fix(dotnet): serialize event enqueue barriers

Author: hydraxman

dotnet/src/Session.cs

@@ -81,6 +81,7 @@ private sealed record EventSubscription(Type EventType, Action<SessionEvent> Han
     private IReadOnlyList<OpenCanvasInstance> _openCanvases = Array.Empty<OpenCanvasInstance>();
 
     private int _isDisposed;
+    private readonly object _eventDispatchGate = new();
     private long _eventEnqueueVersion;
 
     private abstract record EventDispatchItem;
@@ -417,8 +418,7 @@ async Task MonitorRuntimeCompletionAsync(CancellationToken waitCancellationToken
                         activity = await Rpc.Metadata.ActivityAsync(waitCancellationToken).ConfigureAwait(false);
                         if (!activity.HasActiveWork)
                         {
-                            var stableEventVersion = Volatile.Read(ref _eventEnqueueVersion);
-                            await FlushEventDispatchAsync(waitCancellationToken).ConfigureAwait(false);
+                            var stableEventVersion = await FlushEventDispatchAsync(waitCancellationToken).ConfigureAwait(false);
                             if (stableEventVersion != Volatile.Read(ref _eventEnqueueVersion))
                             {
                                 continue;
@@ -583,27 +583,39 @@ internal void DispatchEvent(SessionEvent sessionEvent)
         // never completes (multi-client permission scenario).
         _ = HandleBroadcastEventAsync(sessionEvent);
 
-        // Queue the event for serial processing by user handlers. Increment first so
-        // a completion monitor can detect an event even if this thread is preempted
-        // before the channel write.
-        Interlocked.Increment(ref _eventEnqueueVersion);
-        _eventChannel.Writer.TryWrite(new EventItem(sessionEvent));
+        // Queueing and publishing the new version share the same gate used to capture
+        // a completion barrier's stable version. Publish first so a monitor whose
+        // barrier is already queued cannot observe the old version after this event
+        // is written behind it; the gate prevents a new barrier from splitting the
+        // version update from the channel write.
+        lock (_eventDispatchGate)
+        {
+            Interlocked.Increment(ref _eventEnqueueVersion);
+            var queued = _eventChannel.Writer.TryWrite(new EventItem(sessionEvent));
+            ObjectDisposedException.ThrowIf(!queued, this);
+        }
     }
 
     /// <summary>
     /// Waits until every event already queued for this session has been delivered to
     /// user handlers. Events arriving after the barrier remain queued behind it.
     /// </summary>
-    private async Task FlushEventDispatchAsync(CancellationToken cancellationToken)
+    private async Task<long> FlushEventDispatchAsync(CancellationToken cancellationToken)
     {
         var completion = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-        var queued = _eventChannel.Writer.TryWrite(new EventBarrier(completion));
-        ObjectDisposedException.ThrowIf(!queued, this);
+        long stableEventVersion;
+        lock (_eventDispatchGate)
+        {
+            stableEventVersion = _eventEnqueueVersion;
+            var queued = _eventChannel.Writer.TryWrite(new EventBarrier(completion));
+            ObjectDisposedException.ThrowIf(!queued, this);
+        }
 
         using var registration = cancellationToken.Register(
             static state => ((TaskCompletionSource<bool>)state!).TrySetCanceled(),
             completion);
         await completion.Task.ConfigureAwait(false);
+        return stableEventVersion;
     }
 
     /// <summary>

dotnet/test/Unit/ClientSessionLifetimeTests.cs

@@ -587,6 +587,83 @@ public async Task SendAndWaitAsync_DroppedIdle_Fallback_Flushes_Event_Enqueued_D
         }
     }
 
+    [Fact]
+    public async Task SendAndWaitAsync_DroppedIdle_Fallback_Serializes_Concurrent_Enqueue_With_Final_Barrier()
+    {
+        await using var server = await FakeCopilotServer.StartAsync();
+        server.ConfigureEventEnqueueDuringFinalBarrier();
+        await using var client = new CopilotClient(new CopilotClientOptions { Connection = RuntimeConnection.ForUri(server.Url) });
+        await using var session = await client.CreateSessionAsync(new SessionConfig
+        {
+            OnPermissionRequest = PermissionHandler.ApproveAll
+        });
+
+        var dispatchGate = typeof(CopilotSession).GetField("_eventDispatchGate", BindingFlags.Instance | BindingFlags.NonPublic)
+            ?.GetValue(session)
+            ?? throw new InvalidOperationException("Event dispatch synchronization gate was not found.");
+        var enqueueStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseEnqueue = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var queuedHandlerStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseQueuedHandler = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        Task? enqueueTask = null;
+        using var subscription = session.On<AssistantTurnEndEvent>(evt =>
+        {
+            if (evt.Data.TurnId == "activity-reactivation-barrier")
+            {
+                lock (dispatchGate)
+                {
+                    var dispatchAttempted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+                    var task = Task.Run(() =>
+                    {
+                        dispatchAttempted.TrySetResult();
+                        DispatchEvent(session, new AssistantTurnEndEvent
+                        {
+                            Data = new AssistantTurnEndData { TurnId = "queued-after-concurrent-enqueue" }
+                        });
+                    });
+                    enqueueTask = task;
+                    dispatchAttempted.Task.GetAwaiter().GetResult();
+                    enqueueStarted.TrySetResult();
+                    releaseEnqueue.Task.GetAwaiter().GetResult();
+                }
+            }
+            else if (evt.Data.TurnId == "queued-after-concurrent-enqueue")
+            {
+                queuedHandlerStarted.TrySetResult();
+                releaseQueuedHandler.Task.GetAwaiter().GetResult();
+            }
+        });
+
+        var completionTask = session.SendAndWaitAsync(
+            new MessageOptions { Prompt = "serialize a concurrent enqueue with the final barrier" },
+            timeout: TimeSpan.FromSeconds(5));
+
+        try
+        {
+            await enqueueStarted.Task.WaitAsync(TimeSpan.FromSeconds(2));
+            await Task.Delay(200);
+            Assert.False(enqueueTask!.IsCompleted, "DispatchEvent must be blocked by the shared enqueue/barrier gate.");
+            Assert.False(completionTask.IsCompleted, "Completion must not cross an event enqueue that is contending with the final barrier.");
+
+            releaseEnqueue.TrySetResult();
+            await enqueueTask!.WaitAsync(TimeSpan.FromSeconds(2));
+            var first = await Task.WhenAny(queuedHandlerStarted.Task, completionTask)
+                .WaitAsync(TimeSpan.FromSeconds(2));
+            Assert.Same(queuedHandlerStarted.Task, first);
+            Assert.False(completionTask.IsCompleted, "Completion must wait for the in-flight event's handler to cross the FIFO barrier.");
+
+            releaseQueuedHandler.TrySetResult();
+            var response = await completionTask;
+            Assert.Equal("completed response", response?.Data.Content);
+        }
+        finally
+        {
+            releaseEnqueue.TrySetResult();
+            releaseQueuedHandler.TrySetResult();
+        }
+    }
+
+
     [MethodImpl(MethodImplOptions.NoInlining)]
     private static async Task<WeakReference<CopilotSession>> CreateDroppedSessionAsync(CopilotClient client)
     {
@@ -800,6 +877,7 @@ public void ConfigureEventEnqueueDuringFinalBarrier()
             _enqueueDuringFinalBarrier = true;
         }
 
+
         public void CompleteReactivatedActivity()
         {
             Volatile.Write(ref _hasActiveWork, false);