Address review feedback: fix Go deadlock, bounded channel, per-handler errors

Go SDK:
- Use channel-based event queue to avoid deadlocking the JSON-RPC
  readLoop. Broadcast handlers (tool calls, permission requests) now
  run on a dedicated consumer goroutine instead of inline on the
  readLoop, preventing deadlock when they issue RPC requests.
- Close event channel on Disconnect for clean shutdown.
- Fix unit tests to use channel-based dispatch with goroutines,
  properly validating serialization guarantees.

.NET SDK:
- Switch from unbounded to bounded channel (1024, Wait mode) to
  prevent unbounded memory growth under high event throughput.
- Use GetInvocationList() for per-handler error catching so one
  handler exception does not prevent remaining subscribers from
  seeing the event.
- Add exception filters (when ex is not OperationCanceledException)
  to avoid masking cooperative cancellation.
- Add dedicated test snapshot files for new tests so they don't
  interfere with existing snapshot-based tests.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: stephentoub

dotnet/src/Session.cs

@@ -72,12 +72,18 @@ public sealed partial class CopilotSession : IAsyncDisposable
     private int _isDisposed;
 
     /// <summary>
-    /// Unbounded channel that serializes event dispatch. <see cref="DispatchEvent"/>
+    /// Bounded channel that serializes event dispatch. <see cref="DispatchEvent"/>
     /// enqueues; a single background consumer (<see cref="ProcessEventsAsync"/>) dequeues
     /// and invokes handlers one at a time, preserving arrival order.
+    /// When the channel is full, writers will asynchronously wait, providing backpressure
+    /// instead of unbounded buffering.
     /// </summary>
-    private readonly Channel<SessionEvent> _eventChannel = Channel.CreateUnbounded<SessionEvent>(
-        new() { SingleReader = true });
+    private readonly Channel<SessionEvent> _eventChannel = Channel.CreateBounded<SessionEvent>(
+        new BoundedChannelOptions(1024)
+        {
+            SingleReader = true,
+            FullMode = BoundedChannelFullMode.Wait
+        });
 
     /// <summary>
     /// Gets the unique identifier for this session.
@@ -315,19 +321,27 @@ private async Task ProcessEventsAsync()
             {
                 await HandleBroadcastEventAsync(sessionEvent);
             }
-            catch (Exception ex)
+            catch (Exception ex) when (ex is not OperationCanceledException)
             {
                 LogBroadcastHandlerError(ex);
             }
 
-            // Invoke user handlers serially. Catch exceptions to keep the loop alive.
-            try
+            // Invoke user handlers serially. Catch exceptions per handler to keep the
+            // loop alive and ensure all subscribers see the event even if one fails.
+            var handlers = EventHandlers;
+            if (handlers is not null)
             {
-                EventHandlers?.Invoke(sessionEvent);
-            }
-            catch (Exception ex)
-            {
-                LogEventHandlerError(ex);
+                foreach (var handler in handlers.GetInvocationList())
+                {
+                    try
+                    {
+                        handler.DynamicInvoke(sessionEvent);
+                    }
+                    catch (Exception ex) when (ex is not OperationCanceledException)
+                    {
+                        LogEventHandlerError(ex);
+                    }
+                }
             }
         }
     }

dotnet/test/SessionTests.cs

@@ -470,9 +470,6 @@ await WaitForAsync(() =>
     [Fact]
     public async Task Handler_Exception_Does_Not_Halt_Event_Delivery()
     {
-        // Reuse an existing snapshot — this test only customizes the C# handler.
-        await Ctx.ConfigureForTestAsync("session", "should_have_stateful_conversation");
-
         var session = await CreateSessionAsync();
         var eventCount = 0;
         var gotIdle = new TaskCompletionSource();
@@ -500,9 +497,6 @@ public async Task Handler_Exception_Does_Not_Halt_Event_Delivery()
     [Fact]
     public async Task DisposeAsync_From_Handler_Does_Not_Deadlock()
     {
-        // Reuse an existing snapshot — this test only customizes the C# handler.
-        await Ctx.ConfigureForTestAsync("session", "should_have_stateful_conversation");
-
         var session = await CreateSessionAsync();
         var disposed = new TaskCompletionSource();
 

go/session.go

@@ -12,6 +12,8 @@ import (
 	"github.com/github/copilot-sdk/go/rpc"
 )
 
+const eventChannelSize = 1024
+
 type sessionHandler struct {
 	id uint64
 	fn SessionEventHandler
@@ -65,6 +67,12 @@ type Session struct {
 	hooks             *SessionHooks
 	hooksMux          sync.RWMutex
 
+	// eventCh serializes event dispatch. DispatchEvent enqueues; a single
+	// background goroutine (processEvents) dequeues and invokes handlers
+	// one at a time, preserving arrival order and keeping the JSON-RPC
+	// read loop unblocked.
+	eventCh chan SessionEvent
+
 	// RPC provides typed session-scoped RPC methods.
 	RPC *rpc.SessionRpc
 }
@@ -78,14 +86,17 @@ func (s *Session) WorkspacePath() string {
 
 // newSession creates a new session wrapper with the given session ID and client.
 func newSession(sessionID string, client *jsonrpc2.Client, workspacePath string) *Session {
-	return &Session{
+	s := &Session{
 		SessionID:     sessionID,
 		workspacePath: workspacePath,
 		client:        client,
 		handlers:      make([]sessionHandler, 0),
 		toolHandlers:  make(map[string]ToolHandler),
+		eventCh:       make(chan SessionEvent, eventChannelSize),
 		RPC:           rpc.NewSessionRpc(client, sessionID),
 	}
+	go s.processEvents()
+	return s
 }
 
 // Send sends a message to this session and waits for the response.
@@ -435,42 +446,56 @@ func (s *Session) handleHooksInvoke(hookType string, rawInput json.RawMessage) (
 	}
 }
 
-// dispatchEvent dispatches an event to all registered handlers.
+// dispatchEvent enqueues an event for serial dispatch to all registered handlers.
 //
-// Events are processed serially on the JSON-RPC read loop goroutine:
-// broadcast work (tool calls, permission requests) is executed inline before
-// user handlers, and each handler completes before the next is invoked.
-// Panics in user handlers are recovered to prevent crashing the event dispatcher.
+// This method is non-blocking. The event is placed into an in-memory channel and
+// processed by a single background goroutine (processEvents), which guarantees
+// handlers see events one at a time, in order, without blocking the JSON-RPC
+// read loop.
 func (s *Session) dispatchEvent(event SessionEvent) {
-	// Handle broadcast request events inline (serialized with user handlers)
-	s.handleBroadcastEvent(event)
-
-	s.handlerMutex.RLock()
-	handlers := make([]SessionEventHandler, 0, len(s.handlers))
-	for _, h := range s.handlers {
-		handlers = append(handlers, h.fn)
+	select {
+	case s.eventCh <- event:
+	default:
+		fmt.Printf("Warning: event %s dropped; channel full or session shutting down\n", event.Type)
 	}
-	s.handlerMutex.RUnlock()
-
-	for _, handler := range handlers {
-		// Call handler - don't let panics crash the dispatcher
-		func() {
-			defer func() {
-				if r := recover(); r != nil {
-					fmt.Printf("Error in session event handler: %v\n", r)
-				}
+}
+
+// processEvents is the single-goroutine consumer loop that processes events
+// from the channel. Ensures all user code — event handlers, tool handlers,
+// permission handlers — is invoked serially and in FIFO order. Broadcast work
+// (tool calls, permission requests) completes before user handlers see the event.
+func (s *Session) processEvents() {
+	for event := range s.eventCh {
+		// Handle broadcast request events (serialized with user handlers).
+		// Runs off the readLoop goroutine so RPC requests won't deadlock.
+		s.handleBroadcastEvent(event)
+
+		s.handlerMutex.RLock()
+		handlers := make([]SessionEventHandler, 0, len(s.handlers))
+		for _, h := range s.handlers {
+			handlers = append(handlers, h.fn)
+		}
+		s.handlerMutex.RUnlock()
+
+		for _, handler := range handlers {
+			func() {
+				defer func() {
+					if r := recover(); r != nil {
+						fmt.Printf("Error in session event handler: %v\n", r)
+					}
+				}()
+				handler(event)
 			}()
-			handler(event)
-		}()
+		}
 	}
 }
 
 // handleBroadcastEvent handles broadcast request events by executing local handlers
 // and responding via RPC. This implements the protocol v3 broadcast model where tool
 // calls and permission requests are broadcast as session events to all clients.
 //
-// Handlers are executed inline (not in a separate goroutine) so that all user code
-// — tool handlers, permission handlers, and event handlers — runs serially.
+// Handlers are executed on the processEvents goroutine (not the JSON-RPC read loop)
+// so that RPC responses can be received without deadlocking.
 func (s *Session) handleBroadcastEvent(event SessionEvent) {
 	switch event.Type {
 	case ExternalToolRequested:
@@ -641,6 +666,10 @@ func (s *Session) Disconnect() error {
 		return fmt.Errorf("failed to disconnect session: %w", err)
 	}
 
+	// Stop accepting new events. The consumer goroutine will exit naturally
+	// after draining remaining events.
+	close(s.eventCh)
+
 	// Clear handlers
 	s.handlerMutex.Lock()
 	s.handlers = nil