dotnet: emit JSON-RPC error on pending-buffer overflow + guard drop

Carries forward the Rust SDK PR #1394 follow-up review feedback into the
.NET port:

1. Cap the per-session inbound-request parked-waiter list at 128. When
   exceeded, reject the oldest waiter with 'pending session buffer
   overflow'. The custom JsonRpc dispatcher translates the thrown
   exception into a JSON-RPC error response (-32603) back to the runtime
   so the request id isn't left hanging — silently dropping it would
   leave the runtime waiting on the response until its own timeout.
   Mirrors Rust commit 491b442 and TS commit c167bc3.

2. Use a distinct message ('pending session routing ended before
   session was registered') when the pending guard drops without
   registration. Lets debugging tell the overflow path from the
   create-failed path. Also fixes the pre-existing bug where the
   waiter-fault loop ran inside _pendingLock despite the comment saying
   otherwise — moved it outside the lock so TCS continuations can't
   deadlock against the lock. Mirrors Rust commit e0ff254.

Adds two tests:
- overflow path: 129 early inbound requests → oldest gets error with
  'pending session buffer overflow', remaining 128 resolve after
  registration
- guard-drop path: session.create fails with parked request → error
  response with 'pending session routing ended before session was
  registered'

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

Author: tclem

dotnet/src/Client.cs

@@ -904,23 +904,29 @@ private PendingSessionRoutingGuard BeginPendingSessionRouting()
 
     private void EndPendingSessionRouting()
     {
+        List<TaskCompletionSource<CopilotSession>>? waiters = null;
         lock (_pendingLock)
         {
             _pendingRoutingCount--;
             if (_pendingRoutingCount == 0)
             {
                 _pendingSessionEvents.Clear();
-                var waiters = new List<TaskCompletionSource<CopilotSession>>();
+                waiters = new List<TaskCompletionSource<CopilotSession>>();
                 foreach (var list in _pendingSessionWaiters.Values)
                     waiters.AddRange(list);
                 _pendingSessionWaiters.Clear();
+            }
+        }
 
-                // Fault pending waiters outside the lock so TCS continuations don't run under it.
-                foreach (var tcs in waiters)
-                {
-                    tcs.TrySetException(new InvalidOperationException(
-                        "Cloud session.create completed without registering this sessionId; request dropped."));
-                }
+        // Fault pending waiters outside the lock so TCS continuations don't run under it.
+        // Distinct phrasing from the overflow-eviction path so the runtime / debugging can tell
+        // the two cases apart. Matches the Rust SDK message in PR #1394 (commit e0ff254f).
+        if (waiters != null)
+        {
+            foreach (var tcs in waiters)
+            {
+                tcs.TrySetException(new InvalidOperationException(
+                    "pending session routing ended before session was registered"));
             }
         }
     }
@@ -979,6 +985,18 @@ private Task<CopilotSession> ResolveSessionAsync(string sessionId)
                     list = [];
                     _pendingSessionWaiters[sessionId] = list;
                 }
+
+                // Cap parked waiters per session id. When exceeded, reject the oldest with a
+                // distinct message so the runtime isn't left waiting on a hung request id.
+                // RunContinuationsAsynchronously ensures the continuation won't run under this lock.
+                // Matches the Rust SDK fix in PR #1394 (commit 491b4427).
+                if (list.Count >= PendingSessionBufferLimit)
+                {
+                    var oldest = list[0];
+                    list.RemoveAt(0);
+                    oldest.TrySetException(new InvalidOperationException("pending session buffer overflow"));
+                }
+
                 list.Add(tcs);
                 return tcs.Task;
             }

dotnet/test/Unit/CloudSessionTests.cs

@@ -267,6 +267,93 @@ public async Task CreateCloudSessionAsync_Parks_Inbound_Requests_Until_Registrat
         Assert.Equal("blue", response!["answer"]?.ToString());
     }
 
+    // -------------------------------------------------------------------------
+    // 8. Pending-waiter overflow: oldest is rejected, remaining 128 succeed
+    // -------------------------------------------------------------------------
+
+    [Fact]
+    public async Task PendingRequestWaiterOverflow_RejectsOldestWithOverflowMessage()
+    {
+        const string cloudId = "overflow-session";
+        const int requestCount = 129; // one beyond the 128-waiter cap
+
+        await using var server = await FakeCloudServer.StartAsync(
+            cloudSessionId: cloudId,
+            earlyInboundRequestCount: requestCount);
+
+        await using var client = new CopilotClient(new CopilotClientOptions
+        { Connection = RuntimeConnection.ForUri(server.Url) });
+
+        await using var _ = await client.CreateCloudSessionAsync(new SessionConfig
+        {
+            OnPermissionRequest = PermissionHandler.ApproveAll,
+            Cloud = new CloudSessionOptions
+            {
+                Repository = new CloudSessionRepository { Owner = "github", Name = "copilot-sdk" }
+            },
+            OnUserInputRequest = (_, _) => Task.FromResult(new UserInputResponse { Answer = "yes", WasFreeform = false })
+        });
+
+        var responses = await server.WaitForAllInboundResponses(requestCount, TimeSpan.FromSeconds(10));
+
+        // Exactly one overflow eviction, 128 successful completions.
+        Assert.Equal(1, responses.Count(r => r.IsError));
+        Assert.Equal(128, responses.Count(r => !r.IsError));
+
+        var err = responses.Single(r => r.IsError);
+        Assert.Contains("pending session buffer overflow", err.ErrorMessage ?? "");
+    }
+
+    // -------------------------------------------------------------------------
+    // 9. Guard-drop path: parked requests are rejected with distinct message
+    // -------------------------------------------------------------------------
+
+    [Fact]
+    public async Task PendingSessionGuardDrop_RejectsParkedRequestWithDistinctMessage()
+    {
+        const string cloudId = "guard-drop-session";
+        const int inboundRequestId = 500;
+
+        await using var server = await FakeCloudServer.StartAsync(
+            cloudSessionId: cloudId,
+            failSessionCreate: true,
+            earlyInboundRequest: new Dictionary<string, object?>
+            {
+                ["jsonrpc"] = "2.0",
+                ["id"] = inboundRequestId,
+                ["method"] = "userInput.request",
+                ["params"] = new Dictionary<string, object?>
+                {
+                    ["sessionId"] = cloudId,
+                    ["question"] = "Color?",
+                    ["choices"] = new object?[] { "red", "blue" },
+                    ["allowFreeform"] = false
+                }
+            });
+
+        await using var client = new CopilotClient(new CopilotClientOptions
+        { Connection = RuntimeConnection.ForUri(server.Url) });
+
+        await Assert.ThrowsAnyAsync<Exception>(() =>
+            client.CreateCloudSessionAsync(new SessionConfig
+            {
+                OnPermissionRequest = PermissionHandler.ApproveAll,
+                Cloud = new CloudSessionOptions
+                {
+                    Repository = new CloudSessionRepository { Owner = "github", Name = "copilot-sdk" }
+                }
+            }));
+
+        // The parked request must have been rejected with the guard-drop message (not the overflow message).
+        var responses = await server.WaitForAllInboundResponses(1, TimeSpan.FromSeconds(5));
+
+        Assert.Single(responses);
+        Assert.True(responses[0].IsError);
+        Assert.Contains(
+            "pending session routing ended before session was registered",
+            responses[0].ErrorMessage ?? "");
+    }
+
     // =========================================================================
     // Fake server infrastructure
     // =========================================================================
@@ -280,21 +367,35 @@ private sealed class FakeCloudServer : IAsyncDisposable
         private readonly string _cloudSessionId;
         private readonly Dictionary<string, object?>? _earlyNotification;
         private readonly Dictionary<string, object?>? _earlyInboundRequest;
+        private readonly int _earlyInboundRequestCount;
+        private readonly bool _failSessionCreate;
         private readonly TaskCompletionSource<Dictionary<string, object?>?> _userInputResponseTcs =
             new(TaskCreationOptions.RunContinuationsAsynchronously);
 
+        // Response tracking for overflow / guard-drop tests.
+        private readonly object _inboundResponsesLock = new();
+        private readonly List<InboundResponse> _collectedInboundResponses = [];
+        private int _waitForInboundResponseCount;
+        private TaskCompletionSource<IReadOnlyList<InboundResponse>>? _allInboundResponsesTcs;
+
         public JsonElement? LastCreatePayload { get; private set; }
 
+        public record InboundResponse(int RequestId, bool IsError, string? ErrorMessage);
+
         private FakeCloudServer(
             TcpListener listener,
             string cloudSessionId,
             Dictionary<string, object?>? earlyNotification,
-            Dictionary<string, object?>? earlyInboundRequest)
+            Dictionary<string, object?>? earlyInboundRequest,
+            int earlyInboundRequestCount,
+            bool failSessionCreate)
         {
             _listener = listener;
             _cloudSessionId = cloudSessionId;
             _earlyNotification = earlyNotification;
             _earlyInboundRequest = earlyInboundRequest;
+            _earlyInboundRequestCount = earlyInboundRequestCount;
+            _failSessionCreate = failSessionCreate;
             _serverTask = RunAsync();
         }
 
@@ -310,16 +411,55 @@ public string Url
         public static Task<FakeCloudServer> StartAsync(
             string cloudSessionId = "cloud-session-id",
             Dictionary<string, object?>? earlyNotification = null,
-            Dictionary<string, object?>? earlyInboundRequest = null)
+            Dictionary<string, object?>? earlyInboundRequest = null,
+            int earlyInboundRequestCount = 0,
+            bool failSessionCreate = false)
         {
             var listener = new TcpListener(IPAddress.Loopback, 0);
             listener.Start();
-            return Task.FromResult(new FakeCloudServer(listener, cloudSessionId, earlyNotification, earlyInboundRequest));
+            return Task.FromResult(new FakeCloudServer(
+                listener, cloudSessionId, earlyNotification, earlyInboundRequest,
+                earlyInboundRequestCount, failSessionCreate));
         }
 
         public Task<Dictionary<string, object?>?> WaitForUserInputResponse(TimeSpan timeout)
             => _userInputResponseTcs.Task.WaitAsync(timeout);
 
+        /// <summary>
+        /// Waits until the server has collected <paramref name="count"/> responses (error or success)
+        /// from the client for inbound requests. Used by overflow and guard-drop tests.
+        /// </summary>
+        public Task<IReadOnlyList<InboundResponse>> WaitForAllInboundResponses(int count, TimeSpan timeout)
+        {
+            var tcs = new TaskCompletionSource<IReadOnlyList<InboundResponse>>(
+                TaskCreationOptions.RunContinuationsAsynchronously);
+            lock (_inboundResponsesLock)
+            {
+                _waitForInboundResponseCount = count;
+                _allInboundResponsesTcs = tcs;
+                if (_collectedInboundResponses.Count >= count)
+                    tcs.TrySetResult(new List<InboundResponse>(_collectedInboundResponses));
+            }
+            return tcs.Task.WaitAsync(timeout);
+        }
+
+        private void RecordInboundResponse(InboundResponse response)
+        {
+            TaskCompletionSource<IReadOnlyList<InboundResponse>>? tcs = null;
+            IReadOnlyList<InboundResponse>? snapshot = null;
+            lock (_inboundResponsesLock)
+            {
+                _collectedInboundResponses.Add(response);
+                if (_allInboundResponsesTcs != null &&
+                    _collectedInboundResponses.Count >= _waitForInboundResponseCount)
+                {
+                    tcs = _allInboundResponsesTcs;
+                    snapshot = new List<InboundResponse>(_collectedInboundResponses);
+                }
+            }
+            tcs?.TrySetResult(snapshot!);
+        }
+
         public async ValueTask DisposeAsync()
         {
             _cts.Cancel();
@@ -373,6 +513,16 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel
                         };
                     }
                     _userInputResponseTcs.TrySetResult(dict);
+
+                    if (idElement.ValueKind == JsonValueKind.Number && idElement.TryGetInt32(out var successId))
+                        RecordInboundResponse(new InboundResponse(successId, IsError: false, null));
+                }
+                else if (request.TryGetProperty("error", out var errorEl))
+                {
+                    var requestId = idElement.ValueKind == JsonValueKind.Number && idElement.TryGetInt32(out var errId)
+                        ? errId : -1;
+                    var msg = errorEl.TryGetProperty("message", out var msgEl) ? msgEl.GetString() : null;
+                    RecordInboundResponse(new InboundResponse(requestId, IsError: true, msg));
                 }
                 return;
             }
@@ -424,6 +574,45 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel
                     await Task.Delay(50, cancellationToken);
                 }
 
+                // For overflow tests: send N inbound requests to exercise the buffer cap.
+                if (_earlyInboundRequestCount > 0)
+                {
+                    for (var i = 1; i <= _earlyInboundRequestCount; i++)
+                    {
+                        await WriteMessageAsync(stream, new Dictionary<string, object?>
+                        {
+                            ["jsonrpc"] = "2.0",
+                            ["id"] = i,
+                            ["method"] = "userInput.request",
+                            ["params"] = new Dictionary<string, object?>
+                            {
+                                ["sessionId"] = _cloudSessionId,
+                                ["question"] = $"Question {i}",
+                                ["choices"] = new object?[] { "yes", "no" },
+                                ["allowFreeform"] = false
+                            }
+                        }, cancellationToken);
+                    }
+
+                    // Give the client time to park/overflow all requests before responding.
+                    await Task.Delay(100, cancellationToken);
+                }
+
+                if (_failSessionCreate)
+                {
+                    await WriteMessageAsync(stream, new Dictionary<string, object?>
+                    {
+                        ["jsonrpc"] = "2.0",
+                        ["id"] = id,
+                        ["error"] = new Dictionary<string, object?>
+                        {
+                            ["code"] = -32603,
+                            ["message"] = "session.create failed (test-induced failure)"
+                        }
+                    }, cancellationToken);
+                    return;
+                }
+
                 await WriteMessageAsync(stream, new Dictionary<string, object?>
                 {
                     ["jsonrpc"] = "2.0",