From ab8ec0aa9ada604f01ebd889a77666d8b4eaa524 Mon Sep 17 00:00:00 2001 From: Nathan <8344245+hydraxman@users.noreply.github.com> Date: Fri, 17 Jul 2026 10:27:52 +0800 Subject: [PATCH 1/5] fix(dotnet): recover from dropped session idle events --- dotnet/src/Session.cs | 137 +++++++++- .../E2E/SendAndWaitReliabilityE2ETests.cs | 84 ++++++ .../test/Unit/ClientSessionLifetimeTests.cs | 243 ++++++++++++++++++ ...lete_when_live_sessionidle_is_dropped.yaml | 10 + 4 files changed, 464 insertions(+), 10 deletions(-) create mode 100644 dotnet/test/E2E/SendAndWaitReliabilityE2ETests.cs create mode 100644 test/snapshots/send_and_wait_reliability/should_complete_when_live_sessionidle_is_dropped.yaml diff --git a/dotnet/src/Session.cs b/dotnet/src/Session.cs index 04306b7f62..92174190a3 100644 --- a/dotnet/src/Session.cs +++ b/dotnet/src/Session.cs @@ -81,13 +81,18 @@ private sealed record EventSubscription(Type EventType, Action Han private IReadOnlyList _openCanvases = Array.Empty(); private int _isDisposed; + private long _eventEnqueueVersion; + + private abstract record EventDispatchItem; + private sealed record EventItem(SessionEvent Event) : EventDispatchItem; + private sealed record EventBarrier(TaskCompletionSource Completion) : EventDispatchItem; /// /// Channel that serializes event dispatch. enqueues; /// a single background consumer () dequeues and /// invokes handlers one at a time, preserving arrival order. /// - private readonly Channel _eventChannel = Channel.CreateUnbounded( + private readonly Channel _eventChannel = Channel.CreateUnbounded( new() { SingleReader = true }); /// @@ -336,7 +341,8 @@ public async Task SendAsync(MessageOptions options, CancellationToken ca var totalTimestamp = Stopwatch.GetTimestamp(); var effectiveTimeout = timeout ?? TimeSpan.FromSeconds(60); - var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var tcs = new TaskCompletionSource<(AssistantMessageEvent? Message, string CompletedBy)>(TaskCreationOptions.RunContinuationsAsynchronously); + var completionCandidate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); AssistantMessageEvent? lastAssistantMessage = null; var firstAssistantMessageLogged = false; @@ -346,6 +352,7 @@ void Handler(SessionEvent evt) { case AssistantMessageEvent assistantMessage: lastAssistantMessage = assistantMessage; + completionCandidate.TrySetResult(true); if (!firstAssistantMessageLogged) { firstAssistantMessageLogged = true; @@ -356,12 +363,16 @@ void Handler(SessionEvent evt) } break; + case AssistantTurnEndEvent: + completionCandidate.TrySetResult(true); + break; + case SessionIdleEvent: LoggingHelpers.LogTiming(_logger, LogLevel.Debug, null, "CopilotSession.SendAndWaitAsync idle received. Elapsed={Elapsed}, SessionId={SessionId}", totalTimestamp, SessionId); - tcs.TrySetResult(lastAssistantMessage); + tcs.TrySetResult((lastAssistantMessage, "idle")); break; case SessionErrorEvent errorEvent: @@ -371,6 +382,77 @@ void Handler(SessionEvent evt) } } + async Task MonitorRuntimeCompletionAsync(CancellationToken waitCancellationToken) + { + try + { + var candidateOrCompletion = await Task.WhenAny(completionCandidate.Task, tcs.Task).ConfigureAwait(false); + if (candidateOrCompletion != completionCandidate.Task) + { + return; + } + + // In the normal path session.idle follows the assistant events in the + // same notification burst. Give it a brief chance to arrive so the + // fallback adds no RPC traffic to healthy turns. + var graceDelay = Task.Delay(TimeSpan.FromMilliseconds(250), waitCancellationToken); + if (await Task.WhenAny(graceDelay, tcs.Task).ConfigureAwait(false) != graceDelay) + { + return; + } + + while (!tcs.Task.IsCompleted) + { + // The runtime owns the authoritative view of background tasks and + // follow-up turns. This RPC returns only after those have settled. + await Rpc.Tasks.WaitForPendingAsync(waitCancellationToken).ConfigureAwait(false); + + var activity = await Rpc.Metadata.ActivityAsync(waitCancellationToken).ConfigureAwait(false); + if (!activity.HasActiveWork) + { + // RPC responses and session.event notifications share one ordered + // connection, but user handlers run on a separate FIFO channel. + // Flush that channel before confirming the runtime is still idle. + await FlushEventDispatchAsync(waitCancellationToken).ConfigureAwait(false); + activity = await Rpc.Metadata.ActivityAsync(waitCancellationToken).ConfigureAwait(false); + if (!activity.HasActiveWork) + { + var stableEventVersion = Volatile.Read(ref _eventEnqueueVersion); + await FlushEventDispatchAsync(waitCancellationToken).ConfigureAwait(false); + if (stableEventVersion != Volatile.Read(ref _eventEnqueueVersion)) + { + continue; + } + + activity = await Rpc.Metadata.ActivityAsync(waitCancellationToken).ConfigureAwait(false); + if (!activity.HasActiveWork + && stableEventVersion == Volatile.Read(ref _eventEnqueueVersion)) + { + tcs.TrySetResult((lastAssistantMessage, "activity")); + return; + } + } + } + + await Task.Delay(TimeSpan.FromMilliseconds(100), waitCancellationToken).ConfigureAwait(false); + } + } + catch (RemoteRpcException ex) when (ex.ErrorCode == RemoteRpcException.MethodNotFoundErrorCode) + { + // Older runtimes may not expose activity/task-drain RPCs. Preserve the + // existing event-only behavior and let session.idle or the timeout win. + LogRuntimeCompletionFallbackUnavailable(ex, SessionId); + } + catch (OperationCanceledException) when (waitCancellationToken.IsCancellationRequested) + { + // The timeout/caller-cancellation registration completes tcs. + } + catch (Exception ex) when (ex is RemoteRpcException or IOException or ObjectDisposedException or JsonException) + { + tcs.TrySetException(ex); + } + } + using var subscription = On(Handler); await SendAsync(options, cancellationToken); @@ -385,16 +467,17 @@ void Handler(SessionEvent evt) else tcs.TrySetException(new TimeoutException($"SendAndWaitAsync timed out after {effectiveTimeout}")); }); + var runtimeCompletionTask = MonitorRuntimeCompletionAsync(cts.Token); try { - var result = await tcs.Task; + var completion = await tcs.Task; LoggingHelpers.LogTiming(_logger, LogLevel.Debug, null, "CopilotSession.SendAndWaitAsync complete. Elapsed={Elapsed}, SessionId={SessionId}, CompletedBy={CompletedBy}, AssistantMessageReceived={AssistantMessageReceived}", totalTimestamp, SessionId, - "idle", - result is not null); - return result; + completion.CompletedBy, + completion.Message is not null); + return completion.Message; } catch (Exception ex) when (ex is TimeoutException) { @@ -414,6 +497,11 @@ void Handler(SessionEvent evt) "error"); throw; } + finally + { + cts.Cancel(); + await runtimeCompletionTask.ConfigureAwait(false); + } } /// @@ -485,8 +573,27 @@ internal void DispatchEvent(SessionEvent sessionEvent) // never completes (multi-client permission scenario). _ = HandleBroadcastEventAsync(sessionEvent); - // Queue the event for serial processing by user handlers. - _eventChannel.Writer.TryWrite(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)); + } + + /// + /// Waits until every event already queued for this session has been delivered to + /// user handlers. Events arriving after the barrier remain queued behind it. + /// + private async Task FlushEventDispatchAsync(CancellationToken cancellationToken) + { + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var queued = _eventChannel.Writer.TryWrite(new EventBarrier(completion)); + ObjectDisposedException.ThrowIf(!queued, this); + + using var registration = cancellationToken.Register( + static state => ((TaskCompletionSource)state!).TrySetCanceled(), + completion); + await completion.Task.ConfigureAwait(false); } /// @@ -495,8 +602,15 @@ internal void DispatchEvent(SessionEvent sessionEvent) /// private async Task ProcessEventsAsync() { - await foreach (var sessionEvent in _eventChannel.Reader.ReadAllAsync()) + await foreach (var item in _eventChannel.Reader.ReadAllAsync()) { + if (item is EventBarrier barrier) + { + barrier.Completion.TrySetResult(true); + continue; + } + + var sessionEvent = ((EventItem)item).Event; var dispatchTimestamp = Stopwatch.GetTimestamp(); var eventType = sessionEvent.GetType(); foreach (var subscription in _eventHandlers) @@ -1935,6 +2049,9 @@ await InvokeRpcAsync( [LoggerMessage(Level = LogLevel.Error, Message = "Unhandled exception in session event handler")] private partial void LogEventHandlerError(Exception exception); + [LoggerMessage(Level = LogLevel.Debug, Message = "Runtime completion fallback unavailable. SessionId={sessionId}")] + private partial void LogRuntimeCompletionFallbackUnavailable(Exception exception, string sessionId); + [LoggerMessage(Level = LogLevel.Debug, Message = "Failed to fetch tool metadata for {toolName}")] private partial void LogToolMetadataFetchFailed(Exception exception, string toolName); diff --git a/dotnet/test/E2E/SendAndWaitReliabilityE2ETests.cs b/dotnet/test/E2E/SendAndWaitReliabilityE2ETests.cs new file mode 100644 index 0000000000..5a482c5b33 --- /dev/null +++ b/dotnet/test/E2E/SendAndWaitReliabilityE2ETests.cs @@ -0,0 +1,84 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + *--------------------------------------------------------------------------------------------*/ + +using System.Collections; +using System.Reflection; +using Xunit; +using Xunit.Abstractions; + +namespace GitHub.Copilot.Test.E2E; + +/// +/// Verifies that .NET completion remains reliable when the runtime's ephemeral +/// session.idle notification does not reach the SendAndWait handler. +/// +public class SendAndWaitReliabilityE2ETests(E2ETestFixture fixture, ITestOutputHelper output) + : E2ETestBase(fixture, "send_and_wait_reliability", output) +{ + [Fact] + public async Task Should_Complete_When_Live_SessionIdle_Is_Dropped() + { + await using var session = await CreateSessionAsync(); + var userMessageDispatched = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseUserMessage = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var idleObserved = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + using var userSubscription = session.On(_ => + { + userMessageDispatched.TrySetResult(); + releaseUserMessage.Task.GetAwaiter().GetResult(); + }); + using var idleSubscription = session.On(_ => idleObserved.TrySetResult()); + + var completionTask = session.SendAndWaitAsync( + new MessageOptions { Prompt = "What is 5+5? Reply with just the number." }, + timeout: TimeSpan.FromSeconds(30)); + + try + { + await userMessageDispatched.Task.WaitAsync(TimeSpan.FromSeconds(10)); + SuppressIdleForSendAndWaitHandler(session); + } + finally + { + releaseUserMessage.TrySetResult(); + } + + var response = await completionTask; + await idleObserved.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.NotNull(response); + Assert.Contains("10", response.Data.Content); + } + + private static void SuppressIdleForSendAndWaitHandler(CopilotSession session) + { + var handlersField = typeof(CopilotSession).GetField( + "_eventHandlers", + BindingFlags.Instance | BindingFlags.NonPublic) + ?? throw new InvalidOperationException("CopilotSession._eventHandlers was not found."); + var handlers = (IEnumerable)(handlersField.GetValue(session) + ?? throw new InvalidOperationException("CopilotSession._eventHandlers was null.")); + + var sendAndWaitSubscription = handlers.Cast().Last(subscription => + { + var eventType = subscription.GetType().GetProperty("EventType")?.GetValue(subscription); + return Equals(eventType, typeof(SessionEvent)); + }); + var handlerField = sendAndWaitSubscription.GetType().GetField( + "k__BackingField", + BindingFlags.Instance | BindingFlags.NonPublic) + ?? throw new InvalidOperationException("EventSubscription.Handler backing field was not found."); + var original = (Action)(handlerField.GetValue(sendAndWaitSubscription) + ?? throw new InvalidOperationException("SendAndWait event handler was null.")); + + handlerField.SetValue(sendAndWaitSubscription, (Action)(evt => + { + if (evt is not SessionIdleEvent) + { + original(evt); + } + })); + } +} diff --git a/dotnet/test/Unit/ClientSessionLifetimeTests.cs b/dotnet/test/Unit/ClientSessionLifetimeTests.cs index e1143db17c..d2fab66ceb 100644 --- a/dotnet/test/Unit/ClientSessionLifetimeTests.cs +++ b/dotnet/test/Unit/ClientSessionLifetimeTests.cs @@ -415,6 +415,107 @@ public async Task Generated_Session_Rpc_Throws_When_Session_Disposed() await Assert.ThrowsAsync(() => session.Rpc.Model.GetCurrentAsync()); } + [Fact] + public async Task SendAndWaitAsync_Completes_When_SessionIdle_Notification_Is_Dropped() + { + await using var server = await FakeCopilotServer.StartAsync(); + server.ConfigureDroppedIdleCompletion(); + 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 response = await session.SendAndWaitAsync( + new MessageOptions { Prompt = "complete without idle" }, + timeout: TimeSpan.FromSeconds(2)); + + Assert.NotNull(response); + Assert.Equal("completed response", response.Data.Content); + Assert.Contains(server.Requests, request => request.Method == "session.metadata.activity"); + } + + [Fact] + public async Task SendAndWaitAsync_DroppedIdle_Fallback_Flushes_Preceding_Event_Handlers() + { + await using var server = await FakeCopilotServer.StartAsync(); + server.ConfigureDroppedIdleCompletion(delayEvents: true); + 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 completionTask = session.SendAndWaitAsync( + new MessageOptions { Prompt = "flush handlers before completion" }, + timeout: TimeSpan.FromSeconds(5)); + + var handlerStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseHandler = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = session.On(_ => + { + handlerStarted.TrySetResult(); + releaseHandler.Task.GetAwaiter().GetResult(); + }); + + server.ReleaseCompletionEvents(); + await handlerStarted.Task.WaitAsync(TimeSpan.FromSeconds(2)); + await Task.Delay(200); + + Assert.False(completionTask.IsCompleted, "Completion must wait for preceding FIFO event handlers to finish."); + + releaseHandler.TrySetResult(); + var response = await completionTask; + Assert.Equal("completed response", response?.Data.Content); + } + + [Fact] + public async Task SendAndWaitAsync_DroppedIdle_Fallback_Rechecks_Activity_After_Final_Barrier() + { + await using var server = await FakeCopilotServer.StartAsync(); + server.ConfigureActivityReactivationDuringFinalBarrier(); + 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 barrierHandlerStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseBarrierHandler = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = session.On(evt => + { + if (evt.Data.TurnId == "activity-reactivation-barrier") + { + barrierHandlerStarted.TrySetResult(); + releaseBarrierHandler.Task.GetAwaiter().GetResult(); + } + }); + + var completionTask = session.SendAndWaitAsync( + new MessageOptions { Prompt = "recheck activity after final barrier" }, + timeout: TimeSpan.FromSeconds(5)); + + try + { + await barrierHandlerStarted.Task.WaitAsync(TimeSpan.FromSeconds(2)); + await server.ActivityReactivated.WaitAsync(TimeSpan.FromSeconds(2)); + Assert.False(completionTask.IsCompleted); + + releaseBarrierHandler.TrySetResult(); + await server.ReactivatedActivityObserved.WaitAsync(TimeSpan.FromSeconds(2)); + Assert.False(completionTask.IsCompleted, "Completion must not use stale idle activity after the barrier."); + + server.CompleteReactivatedActivity(); + var response = await completionTask; + Assert.Equal("completed response", response?.Data.Content); + } + finally + { + releaseBarrierHandler.TrySetResult(); + server.CompleteReactivatedActivity(); + } + } + [MethodImpl(MethodImplOptions.NoInlining)] private static async Task> CreateDroppedSessionAsync(CopilotClient client) { @@ -512,12 +613,20 @@ private sealed class FakeCopilotServer : IAsyncDisposable private readonly SemaphoreSlim _writeLock = new(1, 1); private readonly TaskCompletionSource _destroyStarted = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _allowDestroy = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _allowCompletionEvents = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _activityReactivated = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _reactivatedActivityObserved = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly Task _serverTask; private readonly List _requests = []; private readonly object _requestsLock = new(); private string? _lastSessionId; private bool _delayDestroy; private bool _failRuntimeShutdown; + private bool _emitCompletionWithoutIdle; + private bool _delayCompletionEvents; + private bool _reactivateDuringFinalBarrier; + private bool _hasActiveWork; + private int _activityRequestCount; private FakeCopilotServer(TcpListener listener) { @@ -543,6 +652,10 @@ public static Task StartAsync() public Task DestroyStarted => _destroyStarted.Task; + public Task ActivityReactivated => _activityReactivated.Task; + + public Task ReactivatedActivityObserved => _reactivatedActivityObserved.Task; + public int RuntimeShutdownCount { get; private set; } public IReadOnlyList Requests @@ -579,6 +692,32 @@ public void FailRuntimeShutdown() _failRuntimeShutdown = true; } + public void ConfigureDroppedIdleCompletion(bool delayEvents = false) + { + _emitCompletionWithoutIdle = true; + _delayCompletionEvents = delayEvents; + if (!delayEvents) + { + _allowCompletionEvents.TrySetResult(); + } + } + + public void ReleaseCompletionEvents() + { + _allowCompletionEvents.TrySetResult(); + } + + public void ConfigureActivityReactivationDuringFinalBarrier() + { + ConfigureDroppedIdleCompletion(); + _reactivateDuringFinalBarrier = true; + } + + public void CompleteReactivatedActivity() + { + Volatile.Write(ref _hasActiveWork, false); + } + public async ValueTask DisposeAsync() { _allowDestroy.TrySetResult(); @@ -646,6 +785,21 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel { _requests.Add(new RpcRequestRecord(method!, paramsElement)); } + + var activityRequestNumber = 0; + if (method == "session.metadata.activity") + { + activityRequestNumber = Interlocked.Increment(ref _activityRequestCount); + if (_reactivateDuringFinalBarrier && activityRequestNumber == 2) + { + await EmitActivityReactivationBarrierEventAsync(stream, cancellationToken); + } + else if (_reactivateDuringFinalBarrier && activityRequestNumber >= 3 && Volatile.Read(ref _hasActiveWork)) + { + _reactivatedActivityObserved.TrySetResult(); + } + } + object? result = method switch { "connect" => new Dictionary @@ -664,6 +818,12 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel { ["messageId"] = "message-1" }, + "session.metadata.activity" => new Dictionary + { + ["abortable"] = false, + ["hasActiveWork"] = Volatile.Read(ref _hasActiveWork) + }, + "session.tasks.waitForPending" => new Dictionary(), "session.mcp.oauth.handlePendingRequest" => new Dictionary { ["success"] = true @@ -683,6 +843,89 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel ["id"] = id, ["result"] = result }, cancellationToken); + + if (_reactivateDuringFinalBarrier && method == "session.metadata.activity" && activityRequestNumber == 2) + { + Volatile.Write(ref _hasActiveWork, true); + _activityReactivated.TrySetResult(); + } + + if (method == "session.send" && _emitCompletionWithoutIdle) + { + _ = EmitCompletionWithoutIdleAsync(stream, cancellationToken); + } + } + + private Task EmitActivityReactivationBarrierEventAsync(Stream stream, CancellationToken cancellationToken) + { + return WriteMessageAsync(stream, new Dictionary + { + ["jsonrpc"] = "2.0", + ["method"] = "session.event", + ["params"] = new Dictionary + { + ["sessionId"] = _lastSessionId, + ["event"] = new Dictionary + { + ["type"] = "assistant.turn_end", + ["id"] = Guid.NewGuid().ToString(), + ["timestamp"] = DateTimeOffset.UtcNow.ToString("O"), + ["data"] = new Dictionary + { + ["turnId"] = "activity-reactivation-barrier" + } + } + } + }, cancellationToken); + } + + private async Task EmitCompletionWithoutIdleAsync(Stream stream, CancellationToken cancellationToken) + { + if (_delayCompletionEvents) + { + await _allowCompletionEvents.Task.WaitAsync(cancellationToken); + } + + await WriteMessageAsync(stream, new Dictionary + { + ["jsonrpc"] = "2.0", + ["method"] = "session.event", + ["params"] = new Dictionary + { + ["sessionId"] = _lastSessionId, + ["event"] = new Dictionary + { + ["type"] = "assistant.message", + ["id"] = Guid.NewGuid().ToString(), + ["timestamp"] = DateTimeOffset.UtcNow.ToString("O"), + ["data"] = new Dictionary + { + ["content"] = "completed response", + ["messageId"] = "assistant-message-1" + } + } + } + }, cancellationToken); + + await WriteMessageAsync(stream, new Dictionary + { + ["jsonrpc"] = "2.0", + ["method"] = "session.event", + ["params"] = new Dictionary + { + ["sessionId"] = _lastSessionId, + ["event"] = new Dictionary + { + ["type"] = "assistant.turn_end", + ["id"] = Guid.NewGuid().ToString(), + ["timestamp"] = DateTimeOffset.UtcNow.ToString("O"), + ["data"] = new Dictionary + { + ["turnId"] = "turn-1" + } + } + } + }, cancellationToken); } private Dictionary CreateSessionResult(JsonElement request) diff --git a/test/snapshots/send_and_wait_reliability/should_complete_when_live_sessionidle_is_dropped.yaml b/test/snapshots/send_and_wait_reliability/should_complete_when_live_sessionidle_is_dropped.yaml new file mode 100644 index 0000000000..48667da723 --- /dev/null +++ b/test/snapshots/send_and_wait_reliability/should_complete_when_live_sessionidle_is_dropped.yaml @@ -0,0 +1,10 @@ +models: + - claude-sonnet-4.5 +conversations: + - messages: + - role: system + content: ${system} + - role: user + content: What is 5+5? Reply with just the number. + - role: assistant + content: "10" From 9d0b5016c4455dd7710b030d9a4b2c0fe0ca666f Mon Sep 17 00:00:00 2001 From: Nathan <8344245+hydraxman@users.noreply.github.com> Date: Thu, 23 Jul 2026 07:51:13 +0800 Subject: [PATCH 2/5] fix(dotnet): preserve legacy idle fallback --- dotnet/src/Session.cs | 10 +++ .../E2E/SendAndWaitReliabilityE2ETests.cs | 84 ------------------- .../test/Unit/ClientSessionLifetimeTests.cs | 66 +++++++++++++++ ...lete_when_live_sessionidle_is_dropped.yaml | 10 --- 4 files changed, 76 insertions(+), 94 deletions(-) delete mode 100644 dotnet/test/E2E/SendAndWaitReliabilityE2ETests.cs delete mode 100644 test/snapshots/send_and_wait_reliability/should_complete_when_live_sessionidle_is_dropped.yaml diff --git a/dotnet/src/Session.cs b/dotnet/src/Session.cs index 92174190a3..83b69a98ba 100644 --- a/dotnet/src/Session.cs +++ b/dotnet/src/Session.cs @@ -443,6 +443,16 @@ async Task MonitorRuntimeCompletionAsync(CancellationToken waitCancellationToken // existing event-only behavior and let session.idle or the timeout win. LogRuntimeCompletionFallbackUnavailable(ex, SessionId); } + catch (IOException ex) when (ex.InnerException is RemoteRpcException + { + ErrorCode: RemoteRpcException.MethodNotFoundErrorCode + }) + { + // Generated RPC methods surface remote errors through CopilotClient, + // which wraps them in IOException. Treat an older runtime's missing + // fallback methods the same as a direct method-not-found response. + LogRuntimeCompletionFallbackUnavailable(ex, SessionId); + } catch (OperationCanceledException) when (waitCancellationToken.IsCancellationRequested) { // The timeout/caller-cancellation registration completes tcs. diff --git a/dotnet/test/E2E/SendAndWaitReliabilityE2ETests.cs b/dotnet/test/E2E/SendAndWaitReliabilityE2ETests.cs deleted file mode 100644 index 5a482c5b33..0000000000 --- a/dotnet/test/E2E/SendAndWaitReliabilityE2ETests.cs +++ /dev/null @@ -1,84 +0,0 @@ -/*--------------------------------------------------------------------------------------------- - * Copyright (c) Microsoft Corporation. All rights reserved. - *--------------------------------------------------------------------------------------------*/ - -using System.Collections; -using System.Reflection; -using Xunit; -using Xunit.Abstractions; - -namespace GitHub.Copilot.Test.E2E; - -/// -/// Verifies that .NET completion remains reliable when the runtime's ephemeral -/// session.idle notification does not reach the SendAndWait handler. -/// -public class SendAndWaitReliabilityE2ETests(E2ETestFixture fixture, ITestOutputHelper output) - : E2ETestBase(fixture, "send_and_wait_reliability", output) -{ - [Fact] - public async Task Should_Complete_When_Live_SessionIdle_Is_Dropped() - { - await using var session = await CreateSessionAsync(); - var userMessageDispatched = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - var releaseUserMessage = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - var idleObserved = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - - using var userSubscription = session.On(_ => - { - userMessageDispatched.TrySetResult(); - releaseUserMessage.Task.GetAwaiter().GetResult(); - }); - using var idleSubscription = session.On(_ => idleObserved.TrySetResult()); - - var completionTask = session.SendAndWaitAsync( - new MessageOptions { Prompt = "What is 5+5? Reply with just the number." }, - timeout: TimeSpan.FromSeconds(30)); - - try - { - await userMessageDispatched.Task.WaitAsync(TimeSpan.FromSeconds(10)); - SuppressIdleForSendAndWaitHandler(session); - } - finally - { - releaseUserMessage.TrySetResult(); - } - - var response = await completionTask; - await idleObserved.Task.WaitAsync(TimeSpan.FromSeconds(10)); - - Assert.NotNull(response); - Assert.Contains("10", response.Data.Content); - } - - private static void SuppressIdleForSendAndWaitHandler(CopilotSession session) - { - var handlersField = typeof(CopilotSession).GetField( - "_eventHandlers", - BindingFlags.Instance | BindingFlags.NonPublic) - ?? throw new InvalidOperationException("CopilotSession._eventHandlers was not found."); - var handlers = (IEnumerable)(handlersField.GetValue(session) - ?? throw new InvalidOperationException("CopilotSession._eventHandlers was null.")); - - var sendAndWaitSubscription = handlers.Cast().Last(subscription => - { - var eventType = subscription.GetType().GetProperty("EventType")?.GetValue(subscription); - return Equals(eventType, typeof(SessionEvent)); - }); - var handlerField = sendAndWaitSubscription.GetType().GetField( - "k__BackingField", - BindingFlags.Instance | BindingFlags.NonPublic) - ?? throw new InvalidOperationException("EventSubscription.Handler backing field was not found."); - var original = (Action)(handlerField.GetValue(sendAndWaitSubscription) - ?? throw new InvalidOperationException("SendAndWait event handler was null.")); - - handlerField.SetValue(sendAndWaitSubscription, (Action)(evt => - { - if (evt is not SessionIdleEvent) - { - original(evt); - } - })); - } -} diff --git a/dotnet/test/Unit/ClientSessionLifetimeTests.cs b/dotnet/test/Unit/ClientSessionLifetimeTests.cs index d2fab66ceb..2057c91836 100644 --- a/dotnet/test/Unit/ClientSessionLifetimeTests.cs +++ b/dotnet/test/Unit/ClientSessionLifetimeTests.cs @@ -435,6 +435,26 @@ public async Task SendAndWaitAsync_Completes_When_SessionIdle_Notification_Is_Dr Assert.Contains(server.Requests, request => request.Method == "session.metadata.activity"); } + [Fact] + public async Task SendAndWaitAsync_Preserves_Event_Completion_When_Legacy_Runtime_Lacks_Fallback_Rpcs() + { + await using var server = await FakeCopilotServer.StartAsync(); + server.ConfigureLegacyRuntimeCompletion(); + 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 response = await session.SendAndWaitAsync( + new MessageOptions { Prompt = "complete from delayed legacy idle" }, + timeout: TimeSpan.FromSeconds(3)); + + Assert.NotNull(response); + Assert.Equal("completed response", response.Data.Content); + Assert.Contains(server.Requests, request => request.Method == "session.tasks.waitForPending"); + } + [Fact] public async Task SendAndWaitAsync_DroppedIdle_Fallback_Flushes_Preceding_Event_Handlers() { @@ -624,6 +644,8 @@ private sealed class FakeCopilotServer : IAsyncDisposable private bool _failRuntimeShutdown; private bool _emitCompletionWithoutIdle; private bool _delayCompletionEvents; + private bool _emitDelayedIdle; + private bool _fallbackMethodsUnavailable; private bool _reactivateDuringFinalBarrier; private bool _hasActiveWork; private int _activityRequestCount; @@ -702,6 +724,13 @@ public void ConfigureDroppedIdleCompletion(bool delayEvents = false) } } + public void ConfigureLegacyRuntimeCompletion() + { + ConfigureDroppedIdleCompletion(); + _emitDelayedIdle = true; + _fallbackMethodsUnavailable = true; + } + public void ReleaseCompletionEvents() { _allowCompletionEvents.TrySetResult(); @@ -786,6 +815,22 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel _requests.Add(new RpcRequestRecord(method!, paramsElement)); } + if (_fallbackMethodsUnavailable + && method is "session.tasks.waitForPending" or "session.metadata.activity") + { + await WriteMessageAsync(stream, new Dictionary + { + ["jsonrpc"] = "2.0", + ["id"] = id, + ["error"] = new Dictionary + { + ["code"] = -32601, + ["message"] = $"Method not found: {method}" + } + }, cancellationToken); + return; + } + var activityRequestNumber = 0; if (method == "session.metadata.activity") { @@ -926,6 +971,27 @@ private async Task EmitCompletionWithoutIdleAsync(Stream stream, CancellationTok } } }, cancellationToken); + + if (_emitDelayedIdle) + { + await Task.Delay(TimeSpan.FromMilliseconds(500), cancellationToken); + await WriteMessageAsync(stream, new Dictionary + { + ["jsonrpc"] = "2.0", + ["method"] = "session.event", + ["params"] = new Dictionary + { + ["sessionId"] = _lastSessionId, + ["event"] = new Dictionary + { + ["type"] = "session.idle", + ["id"] = Guid.NewGuid().ToString(), + ["timestamp"] = DateTimeOffset.UtcNow.ToString("O"), + ["data"] = new Dictionary() + } + } + }, cancellationToken); + } } private Dictionary CreateSessionResult(JsonElement request) diff --git a/test/snapshots/send_and_wait_reliability/should_complete_when_live_sessionidle_is_dropped.yaml b/test/snapshots/send_and_wait_reliability/should_complete_when_live_sessionidle_is_dropped.yaml deleted file mode 100644 index 48667da723..0000000000 --- a/test/snapshots/send_and_wait_reliability/should_complete_when_live_sessionidle_is_dropped.yaml +++ /dev/null @@ -1,10 +0,0 @@ -models: - - claude-sonnet-4.5 -conversations: - - messages: - - role: system - content: ${system} - - role: user - content: What is 5+5? Reply with just the number. - - role: assistant - content: "10" From 05201477d19b63a499bdfe092a56299d6e545758 Mon Sep 17 00:00:00 2001 From: Nathan <8344245+hydraxman@users.noreply.github.com> Date: Fri, 24 Jul 2026 07:03:15 +0800 Subject: [PATCH 3/5] test(dotnet): cover final barrier enqueue race --- .../test/Unit/ClientSessionLifetimeTests.cs | 60 ++++++++++++++++++- 1 file changed, 59 insertions(+), 1 deletion(-) diff --git a/dotnet/test/Unit/ClientSessionLifetimeTests.cs b/dotnet/test/Unit/ClientSessionLifetimeTests.cs index 2057c91836..27674e9aa8 100644 --- a/dotnet/test/Unit/ClientSessionLifetimeTests.cs +++ b/dotnet/test/Unit/ClientSessionLifetimeTests.cs @@ -536,6 +536,57 @@ public async Task SendAndWaitAsync_DroppedIdle_Fallback_Rechecks_Activity_After_ } } + [Fact] + public async Task SendAndWaitAsync_DroppedIdle_Fallback_Flushes_Event_Enqueued_During_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 queuedHandlerStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseQueuedHandler = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = session.On(evt => + { + if (evt.Data.TurnId == "activity-reactivation-barrier") + { + DispatchEvent(session, new AssistantTurnEndEvent + { + Data = new AssistantTurnEndData { TurnId = "queued-during-final-barrier" } + }); + } + else if (evt.Data.TurnId == "queued-during-final-barrier") + { + queuedHandlerStarted.TrySetResult(); + releaseQueuedHandler.Task.GetAwaiter().GetResult(); + } + }); + + var completionTask = session.SendAndWaitAsync( + new MessageOptions { Prompt = "flush an event queued during the final barrier" }, + timeout: TimeSpan.FromSeconds(5)); + + try + { + var first = await Task.WhenAny(queuedHandlerStarted.Task, completionTask) + .WaitAsync(TimeSpan.FromSeconds(2)); + Assert.Same(queuedHandlerStarted.Task, first); + await Task.Delay(200); + Assert.False(completionTask.IsCompleted, "Completion must wait for events enqueued after the stable snapshot."); + + releaseQueuedHandler.TrySetResult(); + var response = await completionTask; + Assert.Equal("completed response", response?.Data.Content); + } + finally + { + releaseQueuedHandler.TrySetResult(); + } + } + [MethodImpl(MethodImplOptions.NoInlining)] private static async Task> CreateDroppedSessionAsync(CopilotClient client) { @@ -647,6 +698,7 @@ private sealed class FakeCopilotServer : IAsyncDisposable private bool _emitDelayedIdle; private bool _fallbackMethodsUnavailable; private bool _reactivateDuringFinalBarrier; + private bool _enqueueDuringFinalBarrier; private bool _hasActiveWork; private int _activityRequestCount; @@ -742,6 +794,12 @@ public void ConfigureActivityReactivationDuringFinalBarrier() _reactivateDuringFinalBarrier = true; } + public void ConfigureEventEnqueueDuringFinalBarrier() + { + ConfigureDroppedIdleCompletion(); + _enqueueDuringFinalBarrier = true; + } + public void CompleteReactivatedActivity() { Volatile.Write(ref _hasActiveWork, false); @@ -835,7 +893,7 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel if (method == "session.metadata.activity") { activityRequestNumber = Interlocked.Increment(ref _activityRequestCount); - if (_reactivateDuringFinalBarrier && activityRequestNumber == 2) + if ((_reactivateDuringFinalBarrier || _enqueueDuringFinalBarrier) && activityRequestNumber == 2) { await EmitActivityReactivationBarrierEventAsync(stream, cancellationToken); } From 59c07028869c5df01e1b037f08cfa64cc7371979 Mon Sep 17 00:00:00 2001 From: Nathan <8344245+hydraxman@users.noreply.github.com> Date: Fri, 24 Jul 2026 09:12:49 +0800 Subject: [PATCH 4/5] fix(dotnet): serialize event enqueue barriers --- dotnet/src/Session.cs | 32 +++++--- .../test/Unit/ClientSessionLifetimeTests.cs | 78 +++++++++++++++++++ 2 files changed, 100 insertions(+), 10 deletions(-) diff --git a/dotnet/src/Session.cs b/dotnet/src/Session.cs index 83b69a98ba..011ae3099b 100644 --- a/dotnet/src/Session.cs +++ b/dotnet/src/Session.cs @@ -81,6 +81,7 @@ private sealed record EventSubscription(Type EventType, Action Han private IReadOnlyList _openCanvases = Array.Empty(); 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); + } } /// /// Waits until every event already queued for this session has been delivered to /// user handlers. Events arriving after the barrier remain queued behind it. /// - private async Task FlushEventDispatchAsync(CancellationToken cancellationToken) + private async Task FlushEventDispatchAsync(CancellationToken cancellationToken) { var completion = new TaskCompletionSource(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)state!).TrySetCanceled(), completion); await completion.Task.ConfigureAwait(false); + return stableEventVersion; } /// diff --git a/dotnet/test/Unit/ClientSessionLifetimeTests.cs b/dotnet/test/Unit/ClientSessionLifetimeTests.cs index 27674e9aa8..5389c1848d 100644 --- a/dotnet/test/Unit/ClientSessionLifetimeTests.cs +++ b/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(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> CreateDroppedSessionAsync(CopilotClient client) { @@ -800,6 +877,7 @@ public void ConfigureEventEnqueueDuringFinalBarrier() _enqueueDuringFinalBarrier = true; } + public void CompleteReactivatedActivity() { Volatile.Write(ref _hasActiveWork, false); From 998f42d2ed29f44bf76a3bbd9246f3f183e535fd Mon Sep 17 00:00:00 2001 From: Nathan <8344245+hydraxman@users.noreply.github.com> Date: Sat, 25 Jul 2026 06:56:34 +0800 Subject: [PATCH 5/5] fix(dotnet): harden event barrier lifecycle --- dotnet/src/Session.cs | 15 +- .../test/Unit/ClientSessionLifetimeTests.cs | 197 ++++++++++-------- 2 files changed, 122 insertions(+), 90 deletions(-) diff --git a/dotnet/src/Session.cs b/dotnet/src/Session.cs index 011ae3099b..ebb3f3c32d 100644 --- a/dotnet/src/Session.cs +++ b/dotnet/src/Session.cs @@ -590,9 +590,17 @@ internal void DispatchEvent(SessionEvent sessionEvent) // version update from the channel write. lock (_eventDispatchGate) { + // DisposeAsync completes the channel under this same gate. Notifications + // that race with session.destroy are intentionally ignored, matching the + // pre-barrier behavior instead of failing the JSON-RPC notification pump. + if (Volatile.Read(ref _isDisposed) != 0) + { + return; + } + Interlocked.Increment(ref _eventEnqueueVersion); var queued = _eventChannel.Writer.TryWrite(new EventItem(sessionEvent)); - ObjectDisposedException.ThrowIf(!queued, this); + Debug.Assert(queued, "The event channel cannot complete outside the dispatch gate."); } } @@ -2033,7 +2041,10 @@ public async ValueTask DisposeAsync() return; } - _eventChannel.Writer.TryComplete(); + lock (_eventDispatchGate) + { + _eventChannel.Writer.TryComplete(); + } try { diff --git a/dotnet/test/Unit/ClientSessionLifetimeTests.cs b/dotnet/test/Unit/ClientSessionLifetimeTests.cs index 5389c1848d..a4662a5991 100644 --- a/dotnet/test/Unit/ClientSessionLifetimeTests.cs +++ b/dotnet/test/Unit/ClientSessionLifetimeTests.cs @@ -10,6 +10,7 @@ using System.Runtime.CompilerServices; using System.Text; using System.Text.Json; +using Microsoft.Extensions.Logging; using Xunit; namespace GitHub.Copilot.Test.Unit; @@ -18,6 +19,40 @@ public sealed class ClientSessionLifetimeTests { private sealed record RpcRequestRecord(string Method, JsonElement Params); + private sealed class RecordingLogger : ILogger + { + private readonly object _gate = new(); + private readonly List _messages = []; + + public IReadOnlyList Messages + { + get + { + lock (_gate) + { + return _messages.ToArray(); + } + } + } + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + lock (_gate) + { + _messages.Add(formatter(state, exception)); + } + } + } + [Fact] public async Task StopAsync_Requests_Runtime_Shutdown_For_Owned_Process() { @@ -141,6 +176,39 @@ public async Task Disposing_Session_Remains_Rooted_Until_Destroy_Completes() AssertSessionCount(client, sessions: 0); } + [Fact] + public async Task Disposing_Session_Ignores_Racing_Inbound_Event() + { + await using var server = await FakeCopilotServer.StartAsync(); + server.DelayDestroy(); + var logger = new RecordingLogger(); + await using var client = new CopilotClient(new CopilotClientOptions + { + Connection = RuntimeConnection.ForUri(server.Url), + Logger = logger + }); + + var session = await client.CreateSessionAsync(new SessionConfig + { + OnPermissionRequest = PermissionHandler.ApproveAll + }); + + var disposeTask = session.DisposeAsync().AsTask(); + await server.DestroyStarted; + await server.EmitTurnEndEventAsync("event-during-destroy"); + await Task.Delay(100); + server.CompleteDestroy(); + await disposeTask; + + await using var nextSession = await client.CreateSessionAsync(new SessionConfig + { + OnPermissionRequest = PermissionHandler.ApproveAll + }); + Assert.NotNull(nextSession); + Assert.DoesNotContain(logger.Messages, message => + message.Contains("Error handling JSON-RPC method session.event", StringComparison.Ordinal)); + } + [Fact] public async Task StopAsync_Removes_Rooted_Sessions() { @@ -537,7 +605,7 @@ public async Task SendAndWaitAsync_DroppedIdle_Fallback_Rechecks_Activity_After_ } [Fact] - public async Task SendAndWaitAsync_DroppedIdle_Fallback_Flushes_Event_Enqueued_During_Final_Barrier() + public async Task SendAndWaitAsync_DroppedIdle_Fallback_Flushes_Inbound_Event_Enqueued_During_Final_Barrier() { await using var server = await FakeCopilotServer.StartAsync(); server.ConfigureEventEnqueueDuringFinalBarrier(); @@ -553,10 +621,7 @@ public async Task SendAndWaitAsync_DroppedIdle_Fallback_Flushes_Event_Enqueued_D { if (evt.Data.TurnId == "activity-reactivation-barrier") { - DispatchEvent(session, new AssistantTurnEndEvent - { - Data = new AssistantTurnEndData { TurnId = "queued-during-final-barrier" } - }); + server.EmitTurnEndEventAsync("queued-during-final-barrier").GetAwaiter().GetResult(); } else if (evt.Data.TurnId == "queued-during-final-barrier") { @@ -587,83 +652,6 @@ 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(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> CreateDroppedSessionAsync(CopilotClient client) { @@ -767,6 +755,7 @@ private sealed class FakeCopilotServer : IAsyncDisposable private readonly Task _serverTask; private readonly List _requests = []; private readonly object _requestsLock = new(); + private Stream? _stream; private string? _lastSessionId; private bool _delayDestroy; private bool _failRuntimeShutdown; @@ -883,6 +872,30 @@ public void CompleteReactivatedActivity() Volatile.Write(ref _hasActiveWork, false); } + public Task EmitTurnEndEventAsync(string turnId) + { + var stream = _stream ?? throw new InvalidOperationException("The test transport is not connected."); + return WriteMessageAsync(stream, new Dictionary + { + ["jsonrpc"] = "2.0", + ["method"] = "session.event", + ["params"] = new Dictionary + { + ["sessionId"] = _lastSessionId, + ["event"] = new Dictionary + { + ["type"] = "assistant.turn_end", + ["id"] = Guid.NewGuid().ToString(), + ["timestamp"] = DateTimeOffset.UtcNow.ToString("O"), + ["data"] = new Dictionary + { + ["turnId"] = turnId + } + } + } + }, _cts.Token); + } + public async ValueTask DisposeAsync() { _allowDestroy.TrySetResult(); @@ -905,16 +918,24 @@ private async Task RunAsync() { using var tcpClient = await _listener.AcceptTcpClientAsync(_cts.Token); using var stream = tcpClient.GetStream(); + _stream = stream; - while (!_cts.Token.IsCancellationRequested) + try { - using var request = await ReadMessageAsync(stream, _cts.Token); - if (request is null) + while (!_cts.Token.IsCancellationRequested) { - return; - } + using var request = await ReadMessageAsync(stream, _cts.Token); + if (request is null) + { + return; + } - await HandleRequestAsync(stream, request.RootElement, _cts.Token); + await HandleRequestAsync(stream, request.RootElement, _cts.Token); + } + } + finally + { + _stream = null; } }