Serialize event dispatch in .NET and Go SDKs (#791)

* Serialize event dispatch in .NET and Go SDKs

In .NET, StreamJsonRpc dispatches notifications concurrently on the
thread pool. The old code invoked user event handlers inline from
DispatchEvent, which meant handlers could run concurrently and out of
order.

In Go, the JSON-RPC read loop is single-threaded, so user handlers were
already serialized. However, broadcast handlers (tool calls, permission
requests) ran inline on the read loop, which deadlocked when a handler
issued an RPC request back through the same connection.

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 (void this._executeToolAndRespond()) and Python
(asyncio.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>

* Fix panic on send to closed event channel in Go SDK

Protect dispatchEvent with a recover guard so that a notification
arriving after Disconnect does not crash the process. Also wrap the
channel close in sync.Once so Disconnect is safe to call more than once.

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

---------

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

Author: 3 people

dotnet/src/Client.cs

@@ -412,7 +412,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)
@@ -516,7 +516,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)
@@ -186,7 +199,7 @@ public async Task<string> SendAsync(MessageOptions options, CancellationToken ca
         CancellationToken cancellationToken = default)
     {
         var effectiveTimeout = timeout ?? TimeSpan.FromSeconds(60);
-        var tcs = new TaskCompletionSource<AssistantMessageEvent?>();
+        var tcs = new TaskCompletionSource<AssistantMessageEvent?>(TaskCreationOptions.RunContinuationsAsynchronously);
         AssistantMessageEvent? lastAssistantMessage = null;
 
         void Handler(SessionEvent evt)
@@ -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);
         }
     }
 
@@ -707,6 +755,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
@@ -735,6 +788,8 @@ public async ValueTask DisposeAsync()
             return;
         }
 
+        _eventChannel.Writer.TryComplete();
+
         try
         {
             await InvokeRpcAsync<object>(
@@ -749,12 +804,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/Harness/TestHelper.cs

@@ -10,7 +10,7 @@ public static class TestHelper
         CopilotSession session,
         TimeSpan? timeout = null)
     {
-        var tcs = new TaskCompletionSource<AssistantMessageEvent>();
+        var tcs = new TaskCompletionSource<AssistantMessageEvent>(TaskCreationOptions.RunContinuationsAsynchronously);
         using var cts = new CancellationTokenSource(timeout ?? TimeSpan.FromSeconds(60));
 
         AssistantMessageEvent? finalAssistantMessage = null;
@@ -78,7 +78,7 @@ public static async Task<T> GetNextEventOfTypeAsync<T>(
         CopilotSession session,
         TimeSpan? timeout = null) where T : SessionEvent
     {
-        var tcs = new TaskCompletionSource<T>();
+        var tcs = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
         using var cts = new CancellationTokenSource(timeout ?? TimeSpan.FromSeconds(60));
 
         using var subscription = session.On(evt =>

dotnet/test/MultiClientTests.cs

@@ -109,10 +109,10 @@ public async Task Both_Clients_See_Tool_Request_And_Completion_Events()
         });
 
         // Set up event waiters BEFORE sending the prompt to avoid race conditions
-        var client1Requested = new TaskCompletionSource<bool>();
-        var client2Requested = new TaskCompletionSource<bool>();
-        var client1Completed = new TaskCompletionSource<bool>();
-        var client2Completed = new TaskCompletionSource<bool>();
+        var client1Requested = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var client2Requested = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var client1Completed = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var client2Completed = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 
         using var sub1 = session1.On(evt =>
         {