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

Carries forward the Rust SDK PR #1394 review feedback into the Java port:

1. Cap the per-session inbound-request parked-waiter list at 128
   (BUFFER_LIMIT). When exceeded, evict the oldest waiter via
   completeExceptionally("pending session buffer overflow"). The
   RpcHandlerDispatcher thread blocked in waiter.get() wakes up, catches
   ExecutionException, and resolveSessionForRequest calls
   rpc.sendErrorResponse(-32603, ...) so the runtime isn't left waiting
   on a request id that will never get a reply. Mirrors Rust commit
   491b442 and TS commit c167bc3 on PR #1395.

2. decrementGuard now completes all stale waiters internally with a
   distinct message ("pending session routing ended before session was
   registered") instead of returning them for callers to complete with
   ad-hoc strings. A single canonical message lets debugging tell the
   overflow-eviction path from the create-failed path. Mirrors Rust
   commit e0ff254 and TS commit c167bc3.

3. Fix isDone() fast path in resolveSessionForRequest: the existing
   catch-all "fall through" swallowed ExecutionException from an
   overflow-evicted waiter, sending a generic -32602 "Unknown session"
   error instead of -32603. Now explicitly catches ExecutionException in
   the isDone() branch and sends the same -32603 error as the blocking
   path.

Adds two new unit tests in CloudSessionTest:
- parkedRequestWaiterBuffer_overflow_evictsOldestWithOverflowMessage:
  parks 129 waiters, verifies oldest completes with "pending session
  buffer overflow", remaining 128 resolve normally after registration.
- parkedRequestWaiter_guardDropMessage_isDistinctFromOverflowMessage:
  parks a request via the full handler path, drops the guard without
  registration, verifies the JSON-RPC error response contains "routing
  ended before session was registered" but not "buffer overflow".

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

Author: tclem

java/src/main/java/com/github/copilot/sdk/CopilotClient.java

@@ -583,9 +583,7 @@ public CompletableFuture<CopilotSession> createCloudSession(SessionConfig config
                 if (returnedId == null || returnedId.isEmpty()) {
                     // No id: release the guard and surface the error. Any runtime session
                     // created on the other side may leak — we have nothing to destroy.
-                    var staleWaiters = pendingRoutingState.decrementGuard();
-                    completeWaitersExceptionally(staleWaiters,
-                            "Cloud session.create completed without registering this sessionId; request dropped");
+                    pendingRoutingState.decrementGuard();
                     LOG.warning("Cloud session.create response missing sessionId; runtime session may leak");
                     throw new RuntimeException(
                             "Cloud session.create response did not include a sessionId; cannot register session.");
@@ -619,45 +617,30 @@ public CompletableFuture<CopilotSession> createCloudSession(SessionConfig config
                         waiter.complete(session);
                     }
                 } catch (Exception e) {
-                    // Roll back: remove session from map, release guard with exceptional completion
+                    // Roll back: remove session from map, release guard.
                     sessions.remove(returnedId);
-                    var staleWaiters = pendingRoutingState.decrementGuard();
-                    completeWaitersExceptionally(staleWaiters,
-                            "Cloud session post-registration setup failed; request dropped");
+                    pendingRoutingState.decrementGuard();
                     LoggingHelpers.logTiming(LOG, Level.WARNING, e,
                             "CopilotClient.createCloudSession post-registration setup failed. Elapsed={Elapsed}",
                             totalNanos);
                     throw e instanceof RuntimeException re ? re : new RuntimeException(e);
                 }
 
-                var staleWaiters = pendingRoutingState.decrementGuard();
-                completeWaitersExceptionally(staleWaiters,
-                        "Cloud session.create completed without registering this sessionId; request dropped");
+                pendingRoutingState.decrementGuard();
 
                 LoggingHelpers.logTiming(LOG, Level.FINE,
                         "CopilotClient.createCloudSession complete. Elapsed={Elapsed}, SessionId=" + returnedId,
                         totalNanos);
                 return session;
             }).exceptionally(ex -> {
-                var staleWaiters = pendingRoutingState.decrementGuard();
-                completeWaitersExceptionally(staleWaiters, "Cloud session.create failed; request dropped");
+                pendingRoutingState.decrementGuard();
                 LoggingHelpers.logTiming(LOG, Level.WARNING, ex,
                         "CopilotClient.createCloudSession failed. Elapsed={Elapsed}", totalNanos);
                 throw ex instanceof RuntimeException re ? re : new RuntimeException(ex);
             });
         });
     }
 
-    private static void completeWaitersExceptionally(java.util.List<CompletableFuture<CopilotSession>> waiters,
-            String message) {
-        if (!waiters.isEmpty()) {
-            var ex = new RuntimeException(message);
-            for (var waiter : waiters) {
-                waiter.completeExceptionally(ex);
-            }
-        }
-    }
-
     /**
      * Resumes an existing Copilot session.
      * <p>

java/src/main/java/com/github/copilot/sdk/PendingRoutingState.java

@@ -56,23 +56,32 @@ synchronized void incrementGuard() {
 
     /**
      * Decrement the guard count. If the count reaches zero, clears all buffered
-     * events and completes all parked waiters exceptionally.
-     *
-     * @return list of waiters that should be completed exceptionally (caller must
-     *         complete them outside this lock to avoid re-entrancy)
+     * events and completes all parked request waiters exceptionally with a
+     * canonical message that is distinct from the overflow-eviction path.
      */
-    synchronized List<CompletableFuture<CopilotSession>> decrementGuard() {
+    synchronized void decrementGuard() {
         guardCount = Math.max(0, guardCount - 1);
-        if (guardCount == 0) {
-            pendingEvents.clear();
-            var stale = new ArrayList<CompletableFuture<CopilotSession>>();
-            for (var list : pendingWaiters.values()) {
-                stale.addAll(list);
+        if (guardCount != 0) {
+            return;
+        }
+        pendingEvents.clear();
+        var stale = new ArrayList<CompletableFuture<CopilotSession>>();
+        for (var list : pendingWaiters.values()) {
+            stale.addAll(list);
+        }
+        pendingWaiters.clear();
+        if (!stale.isEmpty()) {
+            // Use a distinct phrasing from the overflow-eviction path so that
+            // debugging can tell the two failure modes apart. Matches the Rust
+            // SDK message (PR #1394 commit e0ff254f) and the TS SDK (commit
+            // c167bc3e).
+            LOG.warning("Pending session routing ended before session was registered; " + "completing " + stale.size()
+                    + " parked request waiter(s) exceptionally");
+            var ex = new RuntimeException("pending session routing ended before session was registered");
+            for (var waiter : stale) {
+                waiter.completeExceptionally(ex);
             }
-            pendingWaiters.clear();
-            return stale;
         }
-        return Collections.emptyList();
     }
 
     /**
@@ -137,7 +146,19 @@ synchronized CompletableFuture<CopilotSession> tryParkRequest(String sessionId,
             return null; // no pending; caller sends error
         }
         var future = new CompletableFuture<CopilotSession>();
-        pendingWaiters.computeIfAbsent(sessionId, k -> new ArrayList<>()).add(future);
+        var list = pendingWaiters.computeIfAbsent(sessionId, k -> new ArrayList<>());
+        if (list.size() >= BUFFER_LIMIT) {
+            // Cap parked waiters per session. When exceeded, evict the oldest
+            // and complete it with a distinct overflow message so the runtime
+            // gets an error response rather than hanging on a reply that will
+            // never arrive. Matches Rust PR #1394 (commit 491b4427) and TS
+            // (commit c167bc3e).
+            var oldest = list.remove(0);
+            LOG.warning("Pending session request waiter buffer full for session " + sessionId + " (limit="
+                    + BUFFER_LIMIT + "); evicting oldest request");
+            oldest.completeExceptionally(new RuntimeException("pending session buffer overflow"));
+        }
+        list.add(future);
         return future;
     }
 

java/src/main/java/com/github/copilot/sdk/RpcHandlerDispatcher.java

@@ -570,9 +570,22 @@ private CopilotSession resolveSessionForRequest(String sessionId, long requestId
             CompletableFuture<CopilotSession> waiter = pendingState.tryParkRequest(sessionId, sessions);
             if (waiter != null) {
                 if (waiter.isDone()) {
-                    // Resolved synchronously (session was already registered under the lock)
+                    // Session was already registered under tryParkRequest's lock —
+                    // complete synchronously. If the waiter was completed exceptionally
+                    // (e.g. overflow eviction just beat us to isDone()), send the
+                    // same -32603 error as the blocking path so the runtime isn't
+                    // left waiting for a reply that will never arrive.
                     try {
                         return waiter.get();
+                    } catch (ExecutionException e) {
+                        Throwable cause = e.getCause();
+                        try {
+                            rpc.sendErrorResponse(requestId, -32603, "Session " + sessionId + " not registered: "
+                                    + (cause != null ? cause.getMessage() : e.getMessage()));
+                        } catch (IOException ioe) {
+                            LOG.log(Level.SEVERE, "Failed to send synchronous registration-failed error", ioe);
+                        }
+                        return null;
                     } catch (Exception e) {
                         // fall through to send error
                     }

java/src/test/java/com/github/copilot/sdk/CloudSessionTest.java

@@ -306,8 +306,7 @@ void bufferEarlySessionEventNotifications() throws Exception {
         }
 
         // Release the guard
-        var staleWaiters = pendingState.decrementGuard();
-        assertTrue(staleWaiters.isEmpty(), "No stale waiters expected");
+        pendingState.decrementGuard();
 
         // The buffered session.event should now have been replayed to the session
         Thread.sleep(50);
@@ -413,21 +412,104 @@ void parkedRequestFailsExceptionallyWhenGuardDroppedWithoutRegistration() throws
 
         Thread.sleep(100);
 
-        // Drop the guard WITHOUT registering the session → waiters should be completed
-        // exceptionally
-        var staleWaiters = pendingState.decrementGuard();
-        // Complete them exceptionally to simulate what CopilotClient does on failure
-        for (var w : staleWaiters) {
-            w.completeExceptionally(new RuntimeException("Cloud session.create failed; request dropped"));
-        }
+        // Drop the guard WITHOUT registering the session. decrementGuard now
+        // completes parked waiters internally with the canonical message.
+        pendingState.decrementGuard();
 
-        // The handler should eventually receive the exceptional completion and send an
+        // The handler should receive the exceptional completion and send an
         // error response
         toolCallFuture.get(5, TimeUnit.SECONDS);
 
         JsonNode response = readResponse();
         assertNotNull(response, "Should have received an error response");
         assertEquals(99, response.get("id").asInt());
         assertNotNull(response.get("error"), "Response should be an error (not a result)");
+        String errorMessage = response.get("error").get("message").asText();
+        assertTrue(errorMessage.contains("routing ended before session was registered"),
+                "Error message should contain the canonical guard-drop phrase; got: " + errorMessage);
+    }
+
+    // =========================================================================
+    // Test 8: overflow path — oldest parked waiter gets the overflow message
+    // =========================================================================
+
+    @Test
+    void parkedRequestWaiterBuffer_overflow_evictsOldestWithOverflowMessage() throws Exception {
+        pendingState.incrementGuard();
+        String sid = "cloud-overflow-requests";
+
+        // Park BUFFER_LIMIT + 1 waiters via tryParkRequest. The 129th call must
+        // evict the very first waiter and complete it with the overflow message.
+        var waiters = new java.util.ArrayList<CompletableFuture<CopilotSession>>();
+        for (int i = 0; i < PendingRoutingState.BUFFER_LIMIT + 1; i++) {
+            waiters.add(pendingState.tryParkRequest(sid, sessions));
+        }
+
+        // The first waiter (oldest) must have been evicted with the overflow message.
+        CompletableFuture<CopilotSession> oldest = waiters.get(0);
+        assertTrue(oldest.isCompletedExceptionally(), "Oldest waiter should be completed exceptionally on overflow");
+        ExecutionException ex = assertThrows(ExecutionException.class, oldest::get);
+        assertEquals("pending session buffer overflow", ex.getCause().getMessage());
+
+        // The remaining BUFFER_LIMIT waiters should still be pending.
+        for (int i = 1; i <= PendingRoutingState.BUFFER_LIMIT; i++) {
+            assertFalse(waiters.get(i).isDone(), "Waiter " + i + " should still be pending after overflow eviction");
+        }
+
+        // Registering the session resolves the remaining 128 waiters normally.
+        var session = new CopilotSession(sid, rpc);
+        var flush = pendingState.registerAndFlush(sid, session, sessions);
+        assertEquals(PendingRoutingState.BUFFER_LIMIT, flush.waiters().size(),
+                "registerAndFlush should return all non-evicted waiters");
+        for (var waiter : flush.waiters()) {
+            waiter.complete(session);
+        }
+        for (int i = 1; i <= PendingRoutingState.BUFFER_LIMIT; i++) {
+            assertFalse(waiters.get(i).isCompletedExceptionally(),
+                    "Waiter " + i + " should complete normally, not exceptionally");
+            assertEquals(session, waiters.get(i).get(1, TimeUnit.SECONDS));
+        }
+
+        pendingState.decrementGuard();
+    }
+
+    // =========================================================================
+    // Test 9: guard-drop message is distinct from overflow message
+    // =========================================================================
+
+    @Test
+    void parkedRequestWaiter_guardDropMessage_isDistinctFromOverflowMessage() throws Exception {
+        String pendingSessionId = "cloud-session-distinct-msg";
+
+        pendingState.incrementGuard();
+
+        // Park a request in the background via the full handler path so the
+        // response travels over the socket — this mirrors the real runtime flow.
+        var toolCallFuture = CompletableFuture.runAsync(() -> {
+            ObjectNode params = MAPPER.createObjectNode();
+            params.put("sessionId", pendingSessionId);
+            params.put("toolCallId", "tc-distinct");
+            params.put("toolName", "noop");
+            params.set("arguments", MAPPER.createObjectNode());
+            invokeHandler("tool.call", "77", params);
+        });
+
+        Thread.sleep(100);
+
+        // Drop the guard without registration. decrementGuard completes waiters
+        // internally with the canonical guard-drop message.
+        pendingState.decrementGuard();
+
+        toolCallFuture.get(5, TimeUnit.SECONDS);
+
+        JsonNode response = readResponse();
+        assertEquals(77, response.get("id").asInt());
+        assertNotNull(response.get("error"), "Should be an error response");
+        String msg = response.get("error").get("message").asText();
+
+        // Must contain the guard-drop phrase — NOT the overflow phrase.
+        assertTrue(msg.contains("routing ended before session was registered"),
+                "Guard-drop error must use the routing-ended phrase; got: " + msg);
+        assertFalse(msg.contains("buffer overflow"), "Guard-drop error must NOT use the overflow phrase; got: " + msg);
     }
 }