Restore channel-based dispatch with fire-and-forget broadcast handlers

The channel approach is the right design for .NET:
- Non-blocking enqueue decouples the StreamJsonRpc reader thread from
  user handler execution (StreamJsonRpc dispatches concurrently).
- Single-reader channel guarantees FIFO ordering and serial dispatch.
- Broadcast handlers (tool calls, permissions) fire-and-forget directly
  from DispatchEvent, outside the channel, so a stalled handler cannot
  block event delivery.

Update Should_Receive_Session_Events to await the session.start event
instead of asserting synchronous delivery, since the channel consumer
delivers events asynchronously.

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

Author: stephentoub

dotnet/src/Session.cs

@@ -9,6 +9,7 @@
 using System.Text.Json.Nodes;
 using System.Text.Json.Serialization;
 using System.Threading;
+using System.Threading.Channels;
 using GitHub.Copilot.SDK.Rpc;
 
 namespace GitHub.Copilot.SDK;
@@ -72,22 +73,18 @@ public sealed partial class CopilotSession : IAsyncDisposable
     private int _isDisposed;
 
     /// <summary>
-    /// Guards <see cref="_eventQueue"/> and <see cref="_isDispatching"/>.
+    /// 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 object _dispatchLock = new();
-
-    /// <summary>
-    /// FIFO queue that buffers events arriving while the drain loop is already
-    /// running. Together with <see cref="_isDispatching"/>, this guarantees both
-    /// serial dispatch and strict arrival-order delivery.
-    /// </summary>
-    private readonly Queue<SessionEvent> _eventQueue = new();
-
-    /// <summary>
-    /// True while a thread is draining <see cref="_eventQueue"/>. Only one thread
-    /// at a time runs the drain loop; subsequent callers enqueue and return.
-    /// </summary>
-    private bool _isDispatching;
+    private readonly Channel<SessionEvent> _eventChannel = Channel.CreateBounded<SessionEvent>(
+        new BoundedChannelOptions(1024)
+        {
+            SingleReader = true,
+            FullMode = BoundedChannelFullMode.Wait
+        });
 
     /// <summary>
     /// Gets the unique identifier for this session.
@@ -125,6 +122,7 @@ internal CopilotSession(string sessionId, JsonRpc rpc, ILogger logger, string? w
         _rpc = rpc;
         _logger = logger;
         WorkspacePath = workspacePath;
+        _ = ProcessEventsAsync();
     }
 
     private Task<T> InvokeRpcAsync<T>(string method, object?[]? args, CancellationToken cancellationToken)
@@ -289,61 +287,46 @@ 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>
-    /// Broadcast request events (external_tool.requested, permission.requested) are fired
-    /// concurrently so that a stalled handler does not block event delivery.
-    /// User event handlers are delivered via a queue-drain loop that guarantees both
-    /// serial dispatch and strict FIFO ordering. The first caller to arrive runs the
-    /// drain loop (synchronous delivery); concurrent callers enqueue and return.
+    /// 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)
     {
         // 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).
         _ = HandleBroadcastEventSafe(sessionEvent);
 
-        lock (_dispatchLock)
+        if (!_eventChannel.Writer.TryWrite(sessionEvent))
         {
-            _eventQueue.Enqueue(sessionEvent);
-            if (_isDispatching)
-            {
-                return;
-            }
-            _isDispatching = true;
+            LogEventDropped(sessionEvent.Type);
         }
-
-        DrainEventQueue();
     }
 
     /// <summary>
-    /// Processes queued events one at a time, in FIFO order. Only one thread
-    /// runs this loop at a time; it keeps draining until the queue is empty.
+    /// Single-reader consumer loop that processes events from the channel.
+    /// Ensures user event handlers are invoked serially and in FIFO order.
     /// </summary>
-    private void DrainEventQueue()
+    private async Task ProcessEventsAsync()
     {
-        while (true)
+        await foreach (var sessionEvent in _eventChannel.Reader.ReadAllAsync())
         {
-            SessionEvent evt;
-            lock (_dispatchLock)
-            {
-                if (_eventQueue.Count == 0)
-                {
-                    _isDispatching = false;
-                    return;
-                }
-                evt = _eventQueue.Dequeue();
-            }
-
             var handlers = EventHandlers;
             if (handlers is not null)
             {
                 foreach (var handler in handlers.GetInvocationList())
                 {
                     try
                     {
-                        handler.DynamicInvoke(evt);
+                        handler.DynamicInvoke(sessionEvent);
                     }
                     catch (Exception ex) when (ex is not OperationCanceledException)
                     {
@@ -824,6 +807,8 @@ public async ValueTask DisposeAsync()
             return;
         }
 
+        _eventChannel.Writer.TryComplete();
+
         try
         {
             await InvokeRpcAsync<object>(
@@ -850,6 +835,9 @@ await InvokeRpcAsync<object>(
     [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; channel full")]
+    private partial void LogEventDropped(string eventType);
+
     internal record SendMessageRequest
     {
         public string SessionId { get; init; } = string.Empty;

dotnet/test/SessionTests.cs

@@ -249,11 +249,21 @@ 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>();