Serialize event dispatch in .NET and Go SDKs

Both the .NET and Go SDKs previously invoked user event handlers inline
on the transport reader thread/goroutine. This meant a slow handler
would block all message processing on the connection, and in Go it could
deadlock when a broadcast handler (tool call, permission request) issued
an RPC request back through the same reader.

This PR decouples user handler dispatch from the transport by routing
events through a channel (Go) / Channel<T> (.NET). A single consumer
goroutine/task drains the channel and invokes user handlers serially, in
FIFO order. This matches the guarantees provided by the Node.js and
Python SDKs (which get natural serialization from their single-threaded
event loops) while fitting Go's and .NET's multi-threaded runtimes.

Broadcast handlers (tool calls, permission requests) are fired as
fire-and-forget directly from the dispatch entry point, outside the
channel, so a stalled handler cannot block event delivery. This matches
the existing Node.js (\�oid this._executeToolAndRespond()\) and Python
(\�syncio.ensure_future()\) behavior.

Go changes:
- Add eventCh channel to Session; start processEvents consumer goroutine
- dispatchEvent enqueues to channel and fires broadcast handler goroutine
- Close channel on Disconnect to stop the consumer
- Update unit tests and E2E tests for async delivery

.NET changes:
- Add unbounded Channel<SessionEvent> to CopilotSession; start
  ProcessEventsAsync consumer task in constructor
- DispatchEvent enqueues to channel and fires broadcast handler task
- Complete channel on DisposeAsync
- Per-handler error catching via ImmutableArray iteration
- Cache handler array snapshot to avoid repeated allocation
- Inline broadcast error handling into HandleBroadcastEventAsync
- Update Should_Receive_Session_Events test to await async delivery
- Add Handler_Exception_Does_Not_Halt_Event_Delivery test
- Add DisposeAsync_From_Handler_Does_Not_Deadlock test

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

@@ -2,12 +2,15 @@
  *  Copyright (c) Microsoft Corporation. All rights reserved.
  *--------------------------------------------------------------------------------------------*/
 
+using GitHub.Copilot.SDK.Rpc;
 using Microsoft.Extensions.AI;
+using Microsoft.Extensions.Logging;
 using StreamJsonRpc;
+using System.Collections.Immutable;
 using System.Text.Json;
 using System.Text.Json.Nodes;
 using System.Text.Json.Serialization;
-using GitHub.Copilot.SDK.Rpc;
+using System.Threading.Channels;
 
 namespace GitHub.Copilot.SDK;
 
@@ -52,22 +55,27 @@ namespace GitHub.Copilot.SDK;
 /// </example>
 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).
-    /// 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 ImmutableArray<SessionEventHandler> _eventHandlers = ImmutableArray<SessionEventHandler>.Empty;
+
     private SessionHooks? _hooks;
     private readonly SemaphoreSlim _hooksLock = new(1, 1);
     private SessionRpc? _sessionRpc;
     private int _isDisposed;
 
+    /// <summary>
+    /// 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 +101,20 @@ 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;
+
+        // Start the asynchronous processing loop.
+        _ = ProcessEventsAsync();
     }
 
     private Task<T> InvokeRpcAsync<T>(string method, object?[]? args, CancellationToken cancellationToken)
@@ -236,7 +249,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>
@@ -259,27 +274,53 @@ void Handler(SessionEvent evt)
     /// </example>
     public IDisposable On(SessionEventHandler handler)
     {
-        EventHandlers += handler;
-        return new ActionDisposable(() => EventHandlers -= handler);
+        ImmutableInterlocked.Update(ref _eventHandlers, array => array.Add(handler));
+        return new ActionDisposable(() => ImmutableInterlocked.Update(ref _eventHandlers, array => array.Remove(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.
-    /// Broadcast request events (external_tool.requested, permission.requested) are handled
-    /// internally before being forwarded to user handlers.
+    /// This method is non-blocking. Broadcast request events (external_tool.requested,
+    /// permission.requested) are fired concurrently so that a stalled handler does not
+    /// block event delivery. The event is then placed into an in-memory channel and
+    /// processed by a single background consumer (<see cref="ProcessEventsAsync"/>),
+    /// which guarantees user handlers see events one at a time, in order.
     /// </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);
+        // Fire broadcast work concurrently (fire-and-forget with error logging).
+        // This is done outside the channel so broadcast handlers don't block the
+        // consumer loop — important when a secondary client's handler intentionally
+        // never completes (multi-client permission scenario).
+        _ = HandleBroadcastEventAsync(sessionEvent);
+
+        // Queue the event for serial processing by user handlers.
+        _eventChannel.Writer.TryWrite(sessionEvent);
+    }
 
-        // Reading the field once gives us a snapshot; delegates are immutable.
-        EventHandlers?.Invoke(sessionEvent);
+    /// <summary>
+    /// Single-reader consumer loop that processes events from the channel.
+    /// Ensures user event handlers are invoked serially and in FIFO order.
+    /// </summary>
+    private async Task ProcessEventsAsync()
+    {
+        await foreach (var sessionEvent in _eventChannel.Reader.ReadAllAsync())
+        {
+            foreach (var handler in _eventHandlers)
+            {
+                try
+                {
+                    handler(sessionEvent);
+                }
+                catch (Exception ex)
+                {
+                    LogEventHandlerError(ex);
+                }
+            }
+        }
     }
 
     /// <summary>
@@ -355,37 +396,44 @@ 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)
+        try
         {
-            case ExternalToolRequestedEvent toolEvent:
-                {
-                    var data = toolEvent.Data;
-                    if (string.IsNullOrEmpty(data.RequestId) || string.IsNullOrEmpty(data.ToolName))
-                        return;
-
-                    var tool = GetTool(data.ToolName);
-                    if (tool is null)
-                        return; // This client doesn't handle this tool; another client will.
-
-                    await ExecuteToolAndRespondAsync(data.RequestId, data.ToolName, data.ToolCallId, data.Arguments, tool);
-                    break;
-                }
-
-            case PermissionRequestedEvent permEvent:
-                {
-                    var data = permEvent.Data;
-                    if (string.IsNullOrEmpty(data.RequestId) || data.PermissionRequest is null)
-                        return;
-
-                    var handler = _permissionHandler;
-                    if (handler is null)
-                        return; // This client doesn't handle permissions; another client will.
-
-                    await ExecutePermissionAndRespondAsync(data.RequestId, data.PermissionRequest, handler);
-                    break;
-                }
+            switch (sessionEvent)
+            {
+                case ExternalToolRequestedEvent toolEvent:
+                    {
+                        var data = toolEvent.Data;
+                        if (string.IsNullOrEmpty(data.RequestId) || string.IsNullOrEmpty(data.ToolName))
+                            return;
+
+                        var tool = GetTool(data.ToolName);
+                        if (tool is null)
+                            return; // This client doesn't handle this tool; another client will.
+
+                        await ExecuteToolAndRespondAsync(data.RequestId, data.ToolName, data.ToolCallId, data.Arguments, tool);
+                        break;
+                    }
+
+                case PermissionRequestedEvent permEvent:
+                    {
+                        var data = permEvent.Data;
+                        if (string.IsNullOrEmpty(data.RequestId) || data.PermissionRequest is null)
+                            return;
+
+                        var handler = _permissionHandler;
+                        if (handler is null)
+                            return; // This client doesn't handle permissions; another client will.
+
+                        await ExecutePermissionAndRespondAsync(data.RequestId, data.PermissionRequest, handler);
+                        break;
+                    }
+            }
+        }
+        catch (Exception ex) when (ex is not OperationCanceledException)
+        {
+            LogBroadcastHandlerError(ex);
         }
     }
 
@@ -703,6 +751,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 +784,8 @@ public async ValueTask DisposeAsync()
             return;
         }
 
+        _eventChannel.Writer.TryComplete();
+
         try
         {
             await InvokeRpcAsync<object>(
@@ -745,12 +800,18 @@ await InvokeRpcAsync<object>(
             // Connection is broken or closed
         }
 
-        EventHandlers = null;
+        _eventHandlers = ImmutableInterlocked.InterlockedExchange(ref _eventHandlers, ImmutableArray<SessionEventHandler>.Empty);
         _toolHandlers.Clear();
 
         _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);
+
     internal record SendMessageRequest
     {
         public string SessionId { get; init; } = string.Empty;

dotnet/test/SessionTests.cs

@@ -249,18 +249,40 @@ public async Task Should_Receive_Session_Events()
         // session.start is emitted during the session.create RPC; if the session
         // weren't registered in the sessions map before the RPC, it would be dropped.
         var earlyEvents = new List<SessionEvent>();
+        var sessionStartReceived = new TaskCompletionSource<bool>();
         var session = await CreateSessionAsync(new SessionConfig
         {
-            OnEvent = evt => earlyEvents.Add(evt),
+            OnEvent = evt =>
+            {
+                earlyEvents.Add(evt);
+                if (evt is SessionStartEvent)
+                    sessionStartReceived.TrySetResult(true);
+            },
         });
 
+        // session.start is dispatched asynchronously via the event channel;
+        // wait briefly for the consumer to deliver it.
+        var started = await Task.WhenAny(sessionStartReceived.Task, Task.Delay(TimeSpan.FromSeconds(5)));
+        Assert.Equal(sessionStartReceived.Task, started);
         Assert.Contains(earlyEvents, evt => evt is SessionStartEvent);
 
         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 +303,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 +477,58 @@ await WaitForAsync(() =>
         Assert.Equal("notification", ephemeralEvent.Data.InfoType);
     }
 
+    [Fact]
+    public async Task Handler_Exception_Does_Not_Halt_Event_Delivery()
+    {
+        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()
+    {
+        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;