Serialize event dispatch in .NET and Go SDKs

Event handlers, tool handlers, and permission handlers are now invoked
serially in event-arrival order. This prevents concurrent handler
execution, preserves event ordering guarantees, and ensures broadcast
work (tool calls, permission requests) completes before user handlers
see the event.

.NET changes:
- Add Channel<SessionEvent> with single-reader consumer loop that
  serializes all user code: broadcast handlers, tool handlers,
  permission handlers, and event handlers.
- Change HandleBroadcastEventAsync from async void to async Task so
  broadcast work is awaited inline by the consumer loop.
- Thread ILogger through from CopilotClient to CopilotSession; use
  LoggerMessage source generator for AOT-friendly diagnostics.
- Log dropped events (during dispose) and handler exceptions instead
  of silently swallowing them.
- Complete the event channel in DisposeAsync; document caller
  responsibility to ensure the session is idle before disposing.

Go changes:
- Remove `go` keyword from executeToolAndRespond and
  executePermissionAndRespond calls in handleBroadcastEvent, so
  broadcast work runs inline on the JSON-RPC read loop goroutine.
- Update Disconnect docs for caller-ensures-idle contract.

Tests:
- .NET: add serial dispatch assertion to existing E2E test; add two
  new E2E tests (handler exception resilience, DisposeAsync from
  handler).
- Go: add two unit tests (serial dispatch, handler panic resilience).

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

Author: stephentoub

dotnet/src/Client.cs

@@ -407,7 +407,7 @@ public async Task<CopilotSession> CreateSessionAsync(SessionConfig config, Cance
 
         // Create and register the session before issuing the RPC so that
         // events emitted by the CLI (e.g. session.start) are not dropped.
-        var session = new CopilotSession(sessionId, connection.Rpc);
+        var session = new CopilotSession(sessionId, connection.Rpc, _logger);
         session.RegisterTools(config.Tools ?? []);
         session.RegisterPermissionHandler(config.OnPermissionRequest);
         if (config.OnUserInputRequest != null)
@@ -511,7 +511,7 @@ public async Task<CopilotSession> ResumeSessionAsync(string sessionId, ResumeSes
 
         // Create and register the session before issuing the RPC so that
         // events emitted by the CLI (e.g. session.start) are not dropped.
-        var session = new CopilotSession(sessionId, connection.Rpc);
+        var session = new CopilotSession(sessionId, connection.Rpc, _logger);
         session.RegisterTools(config.Tools ?? []);
         session.RegisterPermissionHandler(config.OnPermissionRequest);
         if (config.OnUserInputRequest != null)

dotnet/src/Session.cs

@@ -3,10 +3,12 @@
  *--------------------------------------------------------------------------------------------*/
 
 using Microsoft.Extensions.AI;
+using Microsoft.Extensions.Logging;
 using StreamJsonRpc;
 using System.Text.Json;
 using System.Text.Json.Nodes;
 using System.Text.Json.Serialization;
+using System.Threading.Channels;
 using GitHub.Copilot.SDK.Rpc;
 
 namespace GitHub.Copilot.SDK;
@@ -55,19 +57,28 @@ public sealed partial class CopilotSession : IAsyncDisposable
     /// <summary>
     /// Multicast delegate used as a thread-safe, insertion-ordered handler list.
     /// The compiler-generated add/remove accessors use a lock-free CAS loop over the backing field.
-    /// Dispatch reads the field once (inherent snapshot, no allocation).
+    /// Invocation is serialized by the single-reader event channel consumer loop.
     /// Expected handler count is small (typically 1–3), so Delegate.Combine/Remove cost is negligible.
     /// </summary>
     private event SessionEventHandler? EventHandlers;
     private readonly Dictionary<string, AIFunction> _toolHandlers = [];
     private readonly JsonRpc _rpc;
+    private readonly ILogger _logger;
     private volatile PermissionRequestHandler? _permissionHandler;
     private volatile UserInputHandler? _userInputHandler;
     private SessionHooks? _hooks;
     private readonly SemaphoreSlim _hooksLock = new(1, 1);
     private SessionRpc? _sessionRpc;
     private int _isDisposed;
 
+    /// <summary>
+    /// Unbounded 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.
+    /// </summary>
+    private readonly Channel<SessionEvent> _eventChannel = Channel.CreateUnbounded<SessionEvent>(
+        new() { SingleReader = true });
+
     /// <summary>
     /// Gets the unique identifier for this session.
     /// </summary>
@@ -93,15 +104,18 @@ public sealed partial class CopilotSession : IAsyncDisposable
     /// </summary>
     /// <param name="sessionId">The unique identifier for this session.</param>
     /// <param name="rpc">The JSON-RPC connection to the Copilot CLI.</param>
+    /// <param name="logger">Logger for diagnostics.</param>
     /// <param name="workspacePath">The workspace path if infinite sessions are enabled.</param>
     /// <remarks>
     /// This constructor is internal. Use <see cref="CopilotClient.CreateSessionAsync"/> to create sessions.
     /// </remarks>
-    internal CopilotSession(string sessionId, JsonRpc rpc, string? workspacePath = null)
+    internal CopilotSession(string sessionId, JsonRpc rpc, ILogger logger, string? workspacePath = null)
     {
         SessionId = sessionId;
         _rpc = rpc;
+        _logger = logger;
         WorkspacePath = workspacePath;
+        _ = ProcessEventsAsync();
     }
 
     private Task<T> InvokeRpcAsync<T>(string method, object?[]? args, CancellationToken cancellationToken)
@@ -236,7 +250,9 @@ void Handler(SessionEvent evt)
     /// Multiple handlers can be registered and will all receive events.
     /// </para>
     /// <para>
-    /// Handler exceptions are allowed to propagate so they are not lost.
+    /// Handlers are invoked serially in event-arrival order on a background thread.
+    /// A handler will never be called concurrently with itself or with other handlers
+    /// on the same session.
     /// </para>
     /// </remarks>
     /// <example>
@@ -264,22 +280,56 @@ public IDisposable On(SessionEventHandler handler)
     }
 
     /// <summary>
-    /// Dispatches an event to all registered handlers.
+    /// Enqueues an event for serial dispatch to all registered handlers.
     /// </summary>
     /// <param name="sessionEvent">The session event to dispatch.</param>
     /// <remarks>
-    /// This method is internal. Handler exceptions are allowed to propagate so they are not lost.
+    /// This method is non-blocking. The event is placed into an in-memory channel and
+    /// processed by a single background consumer (<see cref="ProcessEventsAsync"/>),
+    /// which guarantees handlers see events one at a time, in order.
     /// Broadcast request events (external_tool.requested, permission.requested) are handled
     /// internally before being forwarded to user handlers.
     /// </remarks>
     internal void DispatchEvent(SessionEvent sessionEvent)
     {
-        // Handle broadcast request events (protocol v3) before dispatching to user handlers.
-        // Fire-and-forget: the response is sent asynchronously via RPC.
-        HandleBroadcastEventAsync(sessionEvent);
+        if (!_eventChannel.Writer.TryWrite(sessionEvent))
+        {
+            LogEventDropped(sessionEvent.Type);
+        }
+    }
+
+    /// <summary>
+    /// Single-reader 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) is awaited inline before dispatching to user handlers.
+    /// </summary>
+    private async Task ProcessEventsAsync()
+    {
+        await foreach (var sessionEvent in _eventChannel.Reader.ReadAllAsync())
+        {
+            // Await broadcast work inline so tool/permission handlers are serialized
+            // with everything else. If two tool requests arrive back-to-back, the second
+            // won't start until the first completes.
+            try
+            {
+                await HandleBroadcastEventAsync(sessionEvent);
+            }
+            catch (Exception ex)
+            {
+                LogBroadcastHandlerError(ex);
+            }
 
-        // Reading the field once gives us a snapshot; delegates are immutable.
-        EventHandlers?.Invoke(sessionEvent);
+            // Invoke user handlers serially. Catch exceptions to keep the loop alive.
+            try
+            {
+                EventHandlers?.Invoke(sessionEvent);
+            }
+            catch (Exception ex)
+            {
+                LogEventHandlerError(ex);
+            }
+        }
     }
 
     /// <summary>
@@ -355,7 +405,7 @@ internal async Task<PermissionRequestResult> HandlePermissionRequestAsync(JsonEl
     /// Implements the protocol v3 broadcast model where tool calls and permission requests
     /// are broadcast as session events to all clients.
     /// </summary>
-    private async void HandleBroadcastEventAsync(SessionEvent sessionEvent)
+    private async Task HandleBroadcastEventAsync(SessionEvent sessionEvent)
     {
         switch (sessionEvent)
         {
@@ -703,6 +753,11 @@ public async Task LogAsync(string message, SessionLogRequestLevel? level = null,
     /// <returns>A task representing the dispose operation.</returns>
     /// <remarks>
     /// <para>
+    /// The caller should ensure the session is idle (e.g., <see cref="SendAndWaitAsync"/>
+    /// has returned) before disposing. If the session is not idle, in-flight event handlers
+    /// or tool handlers may observe failures.
+    /// </para>
+    /// <para>
     /// Session state on disk (conversation history, planning state, artifacts) is
     /// preserved, so the conversation can be resumed later by calling
     /// <see cref="CopilotClient.ResumeSessionAsync"/> with the session ID. To
@@ -731,6 +786,12 @@ public async ValueTask DisposeAsync()
             return;
         }
 
+        // Stop accepting new events. The consumer loop will exit naturally
+        // after completing the current event (if any). The caller is expected
+        // to have waited for the session to become idle before disposing;
+        // if the session is not idle, in-flight handlers may see failures.
+        _eventChannel.Writer.TryComplete();
+
         try
         {
             await InvokeRpcAsync<object>(
@@ -751,6 +812,15 @@ await InvokeRpcAsync<object>(
         _permissionHandler = null;
     }
 
+    [LoggerMessage(Level = LogLevel.Error, Message = "Unhandled exception in broadcast event handler")]
+    private partial void LogBroadcastHandlerError(Exception exception);
+
+    [LoggerMessage(Level = LogLevel.Error, Message = "Unhandled exception in session event handler")]
+    private partial void LogEventHandlerError(Exception exception);
+
+    [LoggerMessage(Level = LogLevel.Warning, Message = "Event {EventType} dropped; session is shutting down")]
+    private partial void LogEventDropped(string eventType);
+
     internal record SendMessageRequest
     {
         public string SessionId { get; init; } = string.Empty;

dotnet/test/SessionTests.cs

@@ -258,9 +258,21 @@ public async Task Should_Receive_Session_Events()
 
         var receivedEvents = new List<SessionEvent>();
         var idleReceived = new TaskCompletionSource<bool>();
+        var concurrentCount = 0;
+        var maxConcurrent = 0;
 
         session.On(evt =>
         {
+            // Track concurrent handler invocations to verify serial dispatch.
+            var current = Interlocked.Increment(ref concurrentCount);
+            var seenMax = Volatile.Read(ref maxConcurrent);
+            if (current > seenMax)
+                Interlocked.CompareExchange(ref maxConcurrent, current, seenMax);
+
+            Thread.Sleep(10);
+
+            Interlocked.Decrement(ref concurrentCount);
+
             receivedEvents.Add(evt);
             if (evt is SessionIdleEvent)
             {
@@ -281,6 +293,9 @@ public async Task Should_Receive_Session_Events()
         Assert.Contains(receivedEvents, evt => evt is AssistantMessageEvent);
         Assert.Contains(receivedEvents, evt => evt is SessionIdleEvent);
 
+        // Events must be dispatched serially — never more than one handler invocation at a time.
+        Assert.Equal(1, maxConcurrent);
+
         // Verify the assistant response contains the expected answer
         var assistantMessage = await TestHelper.GetFinalAssistantMessageAsync(session);
         Assert.NotNull(assistantMessage);
@@ -452,6 +467,60 @@ await WaitForAsync(() =>
         Assert.Equal("notification", ephemeralEvent.Data.InfoType);
     }
 
+    [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();
+
+        session.On(evt =>
+        {
+            eventCount++;
+
+            // Throw on the first event to verify the loop keeps going.
+            if (eventCount == 1)
+                throw new InvalidOperationException("boom");
+
+            if (evt is SessionIdleEvent)
+                gotIdle.TrySetResult();
+        });
+
+        await session.SendAsync(new MessageOptions { Prompt = "What is 1+1?" });
+
+        await gotIdle.Task.WaitAsync(TimeSpan.FromSeconds(30));
+
+        // Handler saw more than just the first (throwing) event.
+        Assert.True(eventCount > 1);
+    }
+
+    [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();
+
+        session.On(evt =>
+        {
+            if (evt is UserMessageEvent)
+            {
+                // Call DisposeAsync from within a handler — must not deadlock.
+                session.DisposeAsync().AsTask().ContinueWith(_ => disposed.TrySetResult());
+            }
+        });
+
+        await session.SendAsync(new MessageOptions { Prompt = "What is 1+1?" });
+
+        // If this times out, we deadlocked.
+        await disposed.Task.WaitAsync(TimeSpan.FromSeconds(10));
+    }
+
     private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
     {
         var deadline = DateTime.UtcNow + timeout;

go/session.go

@@ -436,10 +436,13 @@ func (s *Session) handleHooksInvoke(hookType string, rawInput json.RawMessage) (
 }
 
 // dispatchEvent dispatches an event to all registered handlers.
-// This is an internal method; handlers are called synchronously and any panics
-// are recovered to prevent crashing the event dispatcher.
+//
+// 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.
 func (s *Session) dispatchEvent(event SessionEvent) {
-	// Handle broadcast request events internally (fire-and-forget)
+	// Handle broadcast request events inline (serialized with user handlers)
 	s.handleBroadcastEvent(event)
 
 	s.handlerMutex.RLock()
@@ -465,6 +468,9 @@ func (s *Session) dispatchEvent(event SessionEvent) {
 // 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.
 func (s *Session) handleBroadcastEvent(event SessionEvent) {
 	switch event.Type {
 	case ExternalToolRequested:
@@ -481,7 +487,7 @@ func (s *Session) handleBroadcastEvent(event SessionEvent) {
 		if event.Data.ToolCallID != nil {
 			toolCallID = *event.Data.ToolCallID
 		}
-		go s.executeToolAndRespond(*requestID, *toolName, toolCallID, event.Data.Arguments, handler)
+		s.executeToolAndRespond(*requestID, *toolName, toolCallID, event.Data.Arguments, handler)
 
 	case PermissionRequested:
 		requestID := event.Data.RequestID
@@ -492,7 +498,7 @@ func (s *Session) handleBroadcastEvent(event SessionEvent) {
 		if handler == nil {
 			return
 		}
-		go s.executePermissionAndRespond(*requestID, *event.Data.PermissionRequest, handler)
+		s.executePermissionAndRespond(*requestID, *event.Data.PermissionRequest, handler)
 	}
 }
 
@@ -610,6 +616,10 @@ func (s *Session) GetMessages(ctx context.Context) ([]SessionEvent, error) {
 // Disconnect closes this session and releases all in-memory resources (event
 // handlers, tool handlers, permission handlers).
 //
+// The caller should ensure the session is idle (e.g., [Session.SendAndWait] has
+// returned) before disconnecting. If the session is not idle, in-flight event
+// handlers or tool handlers may observe failures.
+//
 // Session state on disk (conversation history, planning state, artifacts) is
 // preserved, so the conversation can be resumed later by calling
 // [Client.ResumeSession] with the session ID. To permanently remove all