diff --git a/dotnet/src/Session.cs b/dotnet/src/Session.cs index 7c34ded16..12626b444 100644 --- a/dotnet/src/Session.cs +++ b/dotnet/src/Session.cs @@ -1920,8 +1920,12 @@ public async ValueTask DisposeAsync() try { - await InvokeRpcAsync( - "session.destroy", [new SessionDestroyRequest() { SessionId = SessionId }], CancellationToken.None); + var response = await InvokeRpcAsync( + "session.detach", [new SessionDetachRequest() { SessionId = SessionId }], CancellationToken.None); + if (!response.Success) + { + LogSessionDetachFailed(SessionId, response.Error ?? "unknown error"); + } } catch (ObjectDisposedException) { @@ -1934,18 +1938,17 @@ await InvokeRpcAsync( finally { RemoveFromClient(); + _eventHandlers = ImmutableInterlocked.InterlockedExchange(ref _eventHandlers, ImmutableArray.Empty); + _toolHandlers.Clear(); + _commandHandlers.Clear(); + + _permissionHandler = null; + _userInputHandler = null; + _elicitationHandler = null; + _exitPlanModeHandler = null; + _autoModeSwitchHandler = null; GC.SuppressFinalize(this); } - - _eventHandlers = ImmutableInterlocked.InterlockedExchange(ref _eventHandlers, ImmutableArray.Empty); - _toolHandlers.Clear(); - _commandHandlers.Clear(); - - _permissionHandler = null; - _userInputHandler = null; - _elicitationHandler = null; - _exitPlanModeHandler = null; - _autoModeSwitchHandler = null; } [LoggerMessage(Level = LogLevel.Error, Message = "Unhandled exception in broadcast event handler")] @@ -1957,6 +1960,9 @@ await InvokeRpcAsync( [LoggerMessage(Level = LogLevel.Debug, Message = "Failed to fetch tool metadata for {toolName}")] private partial void LogToolMetadataFetchFailed(Exception exception, string toolName); + [LoggerMessage(Level = LogLevel.Warning, Message = "Failed to detach session {sessionId}: {error}")] + private partial void LogSessionDetachFailed(string sessionId, string error); + internal record SendMessageRequest { public string SessionId { get; init; } = string.Empty; @@ -1991,11 +1997,17 @@ internal record SessionAbortRequest public string SessionId { get; init; } = string.Empty; } - internal record SessionDestroyRequest + internal record SessionDetachRequest { public string SessionId { get; init; } = string.Empty; } + internal record SessionDetachResponse + { + public bool Success { get; init; } + public string? Error { get; init; } + } + internal void ThrowIfDisposed() { ObjectDisposedException.ThrowIf(Volatile.Read(ref _isDisposed) != 0, this); @@ -2028,7 +2040,8 @@ internal void ThrowIfDisposed() [JsonSerializable(typeof(SendMessageRequest))] [JsonSerializable(typeof(SendMessageResponse))] [JsonSerializable(typeof(SessionAbortRequest))] - [JsonSerializable(typeof(SessionDestroyRequest))] + [JsonSerializable(typeof(SessionDetachRequest))] + [JsonSerializable(typeof(SessionDetachResponse))] [JsonSerializable(typeof(SessionEndHookInput))] [JsonSerializable(typeof(SessionEndHookOutput))] [JsonSerializable(typeof(SessionStartHookInput))] diff --git a/dotnet/test/E2E/ClientLifecycleE2ETests.cs b/dotnet/test/E2E/ClientLifecycleE2ETests.cs index 4b09c695d..82b2d2bad 100644 --- a/dotnet/test/E2E/ClientLifecycleE2ETests.cs +++ b/dotnet/test/E2E/ClientLifecycleE2ETests.cs @@ -121,7 +121,7 @@ public async Task Should_Receive_Session_Deleted_Lifecycle_Event_When_Deleted() } }); - // Do NOT DisposeAsync the session before deleting: dispose sends session.destroy + // Do NOT DisposeAsync the session before deleting: dispose sends session.detach // which closes in-memory state but does not remove the disk file; calling // delete afterwards still succeeds, but skipping dispose keeps the test minimal. await Client.DeleteSessionAsync(sessionId); diff --git a/dotnet/test/E2E/ClientOptionsE2ETests.cs b/dotnet/test/E2E/ClientOptionsE2ETests.cs index 5391e4bdb..16f820003 100644 --- a/dotnet/test/E2E/ClientOptionsE2ETests.cs +++ b/dotnet/test/E2E/ClientOptionsE2ETests.cs @@ -1068,6 +1068,10 @@ function handleMessage(message) { writeResponse(message.id, { messageId: "fake-message" }); return; } + if (message.method === "session.detach") { + writeResponse(message.id, { success: true }); + return; + } writeResponse(message.id, {}); } diff --git a/dotnet/test/Harness/E2ETestBase.cs b/dotnet/test/Harness/E2ETestBase.cs index 3eb0f0e97..ecc003b44 100644 --- a/dotnet/test/Harness/E2ETestBase.cs +++ b/dotnet/test/Harness/E2ETestBase.cs @@ -113,7 +113,7 @@ protected static async Task SuspendAndUntrackSessionForResumeAsync(CopilotSessio { await session.Rpc.SuspendAsync(); - // In-process clients host separate runtimes, while session.destroy removes the + // In-process clients host separate runtimes, while session.detach removes the // session from the current runtime. Untrack locally to exercise resume without // either replacing an active wrapper or destroying the session first. var removeFromClient = typeof(CopilotSession).GetMethod( diff --git a/dotnet/test/Harness/E2ETestContext.cs b/dotnet/test/Harness/E2ETestContext.cs index 1c88d809b..3d53fb820 100644 --- a/dotnet/test/Harness/E2ETestContext.cs +++ b/dotnet/test/Harness/E2ETestContext.cs @@ -569,7 +569,7 @@ private static async Task StopClientForCleanupAsync(CopilotClient client) $"Graceful in-process client cleanup exceeded {s_gracefulClientStopTimeout}; forcing shutdown."); await client.ForceStopAsync(); - // Disposing the connection completes any session.destroy RPC that + // Disposing the connection completes any session.detach RPC that // blocked graceful cleanup. Observe that task before continuing. await gracefulStop.WaitAsync(s_gracefulClientStopTimeout); } diff --git a/dotnet/test/Unit/ClientSessionLifetimeTests.cs b/dotnet/test/Unit/ClientSessionLifetimeTests.cs index da04ff38b..4dff86f59 100644 --- a/dotnet/test/Unit/ClientSessionLifetimeTests.cs +++ b/dotnet/test/Unit/ClientSessionLifetimeTests.cs @@ -118,6 +118,23 @@ public async Task Disposed_Session_Is_Removed_From_Client() AssertSessionCount(client, sessions: 0); } + [Fact] + public async Task DisposeAsync_Does_Not_Throw_When_Detach_Fails() + { + await using var server = await FakeCopilotServer.StartAsync(); + server.FailDetach(); + await using var client = new CopilotClient(new CopilotClientOptions { Connection = RuntimeConnection.ForUri(server.Url) }); + + var session = await client.CreateSessionAsync(new SessionConfig + { + OnPermissionRequest = PermissionHandler.ApproveAll + }); + + await session.DisposeAsync(); + + AssertSessionCount(client, sessions: 0); + } + [Fact] public async Task Disposing_Session_Remains_Rooted_Until_Destroy_Completes() { @@ -692,6 +709,7 @@ private sealed class FakeCopilotServer : IAsyncDisposable private readonly object _requestsLock = new(); private string? _lastSessionId; private bool _delayDestroy; + private bool _failDetach; private bool _failRuntimeShutdown; private FakeCopilotServer(TcpListener listener) @@ -754,6 +772,11 @@ public void FailRuntimeShutdown() _failRuntimeShutdown = true; } + public void FailDetach() + { + _failDetach = true; + } + public async ValueTask DisposeAsync() { _allowDestroy.TrySetResult(); @@ -847,7 +870,7 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel { ["success"] = true }, - "session.destroy" => await DestroySessionAsync(cancellationToken), + "session.detach" => await DetachSessionAsync(cancellationToken), "runtime.shutdown" => HandleRuntimeShutdown(), _ => throw new InvalidOperationException($"Unexpected RPC method '{method}'.") }; @@ -884,7 +907,7 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel }; } - private async Task> DestroySessionAsync(CancellationToken cancellationToken) + private async Task> DetachSessionAsync(CancellationToken cancellationToken) { if (_delayDestroy) { @@ -892,7 +915,9 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel await _allowDestroy.Task.WaitAsync(cancellationToken); } - return []; + return _failDetach + ? new Dictionary { ["success"] = false, ["error"] = "detach failed" } + : new Dictionary { ["success"] = true }; } private Dictionary HandleRuntimeShutdown() diff --git a/dotnet/test/Unit/GitHubTelemetryTests.cs b/dotnet/test/Unit/GitHubTelemetryTests.cs index 24e633387..bd2e2b909 100644 --- a/dotnet/test/Unit/GitHubTelemetryTests.cs +++ b/dotnet/test/Unit/GitHubTelemetryTests.cs @@ -350,7 +350,7 @@ private async Task HandleRequestAsync(Stream stream, JsonElement request, Cancel "session.create" => CaptureCreate(request), "session.resume" => CaptureResume(request), "session.send" => new Dictionary { ["messageId"] = "message-1" }, - "session.destroy" => new Dictionary(), + "session.detach" => new Dictionary { ["success"] = true }, "session.options.update" => new Dictionary { ["success"] = true }, "runtime.shutdown" => new Dictionary(), _ => throw new InvalidOperationException($"Unexpected RPC method '{method}'."), diff --git a/go/client.go b/go/client.go index 856e933ea..05e9f2a05 100644 --- a/go/client.go +++ b/go/client.go @@ -924,6 +924,13 @@ func (c *Client) CreateSession(ctx context.Context, config *SessionConfig) (*Ses "", hasManagedSettings(config.EnableManagedSettings, config.ManagedSettings), ) + s.onDisconnected = func() { + c.sessionsMux.Lock() + defer c.sessionsMux.Unlock() + if c.sessions[sessionID] == s { + delete(c.sessions, sessionID) + } + } s.registerTools(config.Tools) s.registerPermissionHandler(config.OnPermissionRequest) @@ -1258,6 +1265,13 @@ func (c *Client) ResumeSessionWithOptions(ctx context.Context, sessionID string, "", hasManagedSettings(config.EnableManagedSettings, config.ManagedSettings), ) + session.onDisconnected = func() { + c.sessionsMux.Lock() + defer c.sessionsMux.Unlock() + if c.sessions[sessionID] == session { + delete(c.sessions, sessionID) + } + } session.registerTools(config.Tools) session.registerPermissionHandler(config.OnPermissionRequest) diff --git a/go/client_test.go b/go/client_test.go index 3322d7741..f9b61cb37 100644 --- a/go/client_test.go +++ b/go/client_test.go @@ -1939,6 +1939,36 @@ func TestClient_MCPAuthInterestRegistration(t *testing.T) { }) } +func TestSessionDisconnectUnregistersBeforeClientStop(t *testing.T) { + client, requests, cleanup := newInMemoryClient(t) + defer cleanup() + + session, err := client.CreateSession(t.Context(), &SessionConfig{ + OnPermissionRequest: PermissionHandler.ApproveAll, + }) + if err != nil { + t.Fatalf("CreateSession failed: %v", err) + } + requests.clear() + + if err := session.Disconnect(); err != nil { + t.Fatalf("Disconnect failed: %v", err) + } + if err := client.Stop(); err != nil { + t.Fatalf("Stop failed: %v", err) + } + + detachCount := 0 + for _, request := range requests.snapshot() { + if request.Method == "session.detach" { + detachCount++ + } + } + if detachCount != 1 { + t.Fatalf("expected exactly one session.detach request, got %d", detachCount) + } +} + type recordedRequest struct { Method string Params map[string]any @@ -2032,8 +2062,10 @@ func serveInMemoryRuntime(t *testing.T, stdinR *io.PipeReader, stdoutW *io.PipeW result = map[string]any{"id": "interest-1"} case "session.options.update": result = map[string]any{"success": true} - case "session.skills.reload", "session.destroy": + case "session.skills.reload": result = map[string]any{} + case "session.detach": + result = map[string]any{"success": true} default: t.Errorf("unexpected JSON-RPC method %s", request.Method) return diff --git a/go/internal/e2e/client_options_e2e_test.go b/go/internal/e2e/client_options_e2e_test.go index 86332eb6f..13b19e715 100644 --- a/go/internal/e2e/client_options_e2e_test.go +++ b/go/internal/e2e/client_options_e2e_test.go @@ -874,6 +874,10 @@ function handleMessage(message) { writeResponse(message.id, { sessionId, workspacePath: null, capabilities: null }); return; } + if (message.method === "session.detach") { + writeResponse(message.id, { success: true }); + return; + } writeResponse(message.id, {}); } diff --git a/go/session.go b/go/session.go index 99939de4a..df86735c4 100644 --- a/go/session.go +++ b/go/session.go @@ -95,8 +95,11 @@ type Session struct { // eventCh serializes user event handler dispatch. dispatchEvent enqueues; // a single goroutine (processEvents) dequeues and invokes handlers in FIFO order. - eventCh chan SessionEvent - closeOnce sync.Once // guards eventCh close so Disconnect is safe to call more than once + eventCh chan SessionEvent + closeOnce sync.Once // guards eventCh close so Disconnect is safe to call more than once + disconnectMu sync.Mutex + disconnected bool + onDisconnected func() // RPC provides typed session-scoped RPC methods. RPC *rpc.SessionRPC @@ -1716,11 +1719,28 @@ func (s *Session) GetEvents(ctx context.Context) ([]SessionEvent, error) { // log.Printf("Failed to disconnect session: %v", err) // } func (s *Session) Disconnect() error { - _, err := s.client.Request(context.Background(), "session.destroy", sessionDestroyRequest{SessionID: s.SessionID}) + s.disconnectMu.Lock() + defer s.disconnectMu.Unlock() + if s.disconnected { + return nil + } + + result, err := s.client.Request(context.Background(), "session.detach", sessionDetachRequest{SessionID: s.SessionID}) if err != nil { return fmt.Errorf("failed to disconnect session: %w", err) } + var response sessionDetachResponse + if err := json.Unmarshal(result, &response); err != nil { + return fmt.Errorf("failed to decode session detach response: %w", err) + } + if !response.Success { + if response.Error == "" { + response.Error = "unknown error" + } + return fmt.Errorf("failed to disconnect session: %s", response.Error) + } + s.disconnected = true s.closeOnce.Do(func() { close(s.eventCh) }) // Clear handlers @@ -1744,6 +1764,9 @@ func (s *Session) Disconnect() error { s.elicitationHandler = nil s.elicitationMu.Unlock() + if s.onDisconnected != nil { + s.onDisconnected() + } return nil } diff --git a/go/types.go b/go/types.go index 6d6a877d3..4f225d4c9 100644 --- a/go/types.go +++ b/go/types.go @@ -2698,11 +2698,16 @@ type sessionGetMessagesResponse struct { Events []SessionEvent `json:"events"` } -// sessionDestroyRequest is the request for session.destroy -type sessionDestroyRequest struct { +// sessionDetachRequest is the request for session.detach +type sessionDetachRequest struct { SessionID string `json:"sessionId"` } +type sessionDetachResponse struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + // sessionAbortRequest is the request for session.abort type sessionAbortRequest struct { SessionID string `json:"sessionId"` diff --git a/java/src/main/java/com/github/copilot/CopilotClient.java b/java/src/main/java/com/github/copilot/CopilotClient.java index 44878b87e..e655ad1dd 100644 --- a/java/src/main/java/com/github/copilot/CopilotClient.java +++ b/java/src/main/java/com/github/copilot/CopilotClient.java @@ -381,26 +381,32 @@ public CompletableFuture stop() { for (CopilotSession session : new ArrayList<>(sessions.values())) { Runnable closeTask = () -> { - try { - session.close(); - } catch (Exception e) { - LOG.log(Level.WARNING, "Error closing session " + session.getSessionId(), e); - } + session.close(); }; CompletableFuture future; try { future = CompletableFuture.runAsync(closeTask, executor); } catch (RejectedExecutionException e) { LOG.log(Level.WARNING, "Executor rejected session close task; closing inline", e); - closeTask.run(); - future = CompletableFuture.completedFuture(null); + try { + closeTask.run(); + future = CompletableFuture.completedFuture(null); + } catch (RuntimeException closeError) { + future = CompletableFuture.failedFuture(closeError); + } } closeFutures.add(future); } sessions.clear(); return CompletableFuture.allOf(closeFutures.toArray(new CompletableFuture[0])) - .thenCompose(v -> cleanupConnection(true)); + .handle((ignored, closeError) -> closeError) + .thenCompose(closeError -> cleanupConnection(true).thenApply(ignored -> { + if (closeError != null) { + throw new CompletionException(closeError); + } + return null; + })); } /** @@ -571,6 +577,7 @@ public CompletableFuture createSession(SessionConfig config) { long setupNanos = System.nanoTime(); var s = new CopilotSession(sid, connection.rpc); s.setExecutor(executor); + s.setOnClosed(() -> sessions.remove(sid, s)); SessionRequestBuilder.configureSession(s, config); if (extracted.transformCallbacks() != null) { s.registerTransformCallbacks(extracted.transformCallbacks()); @@ -743,6 +750,7 @@ public CompletableFuture resumeSession(String sessionId, ResumeS long setupNanos = System.nanoTime(); var session = new CopilotSession(sessionId, connection.rpc); session.setExecutor(executor); + session.setOnClosed(() -> sessions.remove(sessionId, session)); SessionRequestBuilder.configureSession(session, config); sessions.put(sessionId, session); LoggingHelpers.logTiming(LOG, Level.FINE, diff --git a/java/src/main/java/com/github/copilot/CopilotSession.java b/java/src/main/java/com/github/copilot/CopilotSession.java index 4683fdf01..288724c90 100644 --- a/java/src/main/java/com/github/copilot/CopilotSession.java +++ b/java/src/main/java/com/github/copilot/CopilotSession.java @@ -201,6 +201,8 @@ public final class CopilotSession implements AutoCloseable { /** Tracks whether this session instance has been terminated via close(). */ private volatile boolean isTerminated = false; + private volatile Runnable onClosed = () -> { + }; /** * Creates a new session with the given ID and RPC client. @@ -252,6 +254,10 @@ void setExecutor(Executor executor) { this.executor = executor; } + void setOnClosed(Runnable onClosed) { + this.onClosed = onClosed; + } + /** * Gets the unique identifier for this session. * @@ -2286,9 +2292,10 @@ private void ensureNotTerminated() { /** * Disposes the session and releases all associated resources. *

- * This destroys the session on the server, clears all event handlers, and - * releases tool and permission handlers. After calling this method, the session - * cannot be used again. Subsequent calls to this method have no effect. + * This detaches the session from this client, clears all event handlers, and + * releases tool and permission handlers. Persisted session state remains + * resumable. After calling this method, the session cannot be used again. + * Subsequent calls to this method have no effect. */ @Override public void close() { @@ -2301,10 +2308,20 @@ public void close() { timeoutScheduler.shutdownNow(); + RuntimeException detachFailure = null; try { - rpc.invoke("session.destroy", Map.of("sessionId", sessionId), Void.class).get(5, TimeUnit.SECONDS); + SessionDetachResponse response = rpc + .invoke("session.detach", Map.of("sessionId", sessionId), SessionDetachResponse.class) + .get(5, TimeUnit.SECONDS); + if (response == null || !response.success()) { + String detail = response != null && response.error() != null ? response.error() : "unknown error"; + detachFailure = new IllegalStateException("Failed to detach session " + sessionId + ": " + detail); + } } catch (Exception e) { - LOG.log(Level.FINE, "Error destroying session", e); + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + detachFailure = new IllegalStateException("Failed to detach session " + sessionId, e); } eventHandlers.clear(); @@ -2316,10 +2333,19 @@ public void close() { exitPlanModeHandler.set(null); autoModeSwitchHandler.set(null); hooksHandler.set(null); + onClosed.run(); + + if (detachFailure != null) { + throw detachFailure; + } } // ===== Internal response types for agent API ===== + @JsonIgnoreProperties(ignoreUnknown = true) + record SessionDetachResponse(@JsonProperty("success") boolean success, @JsonProperty("error") String error) { + } + @JsonIgnoreProperties(ignoreUnknown = true) private record AgentListResponse(@JsonProperty("agents") List agents) { } diff --git a/java/src/test/java/com/github/copilot/ClientOptionsE2ETest.java b/java/src/test/java/com/github/copilot/ClientOptionsE2ETest.java index 45056afdb..4a2c3829b 100644 --- a/java/src/test/java/com/github/copilot/ClientOptionsE2ETest.java +++ b/java/src/test/java/com/github/copilot/ClientOptionsE2ETest.java @@ -265,6 +265,8 @@ function resultFor(message) { return { sessionId: message.params?.sessionId ?? 'fake-session', openCanvases: [] }; case 'session.options.update': return { success: true }; + case 'session.detach': + return { success: true }; default: return {}; } diff --git a/java/src/test/java/com/github/copilot/GitHubTelemetryTest.java b/java/src/test/java/com/github/copilot/GitHubTelemetryTest.java index 8e35bd9a9..940fc0c47 100644 --- a/java/src/test/java/com/github/copilot/GitHubTelemetryTest.java +++ b/java/src/test/java/com/github/copilot/GitHubTelemetryTest.java @@ -274,7 +274,8 @@ private void acceptLoop() { respond(server, id, Map.of("sessionId", params.path("sessionId").asText("resume-1"), "workspacePath", "/workspace")); }); - server.registerMethodHandler("session.destroy", (id, params) -> respond(server, id, Map.of())); + server.registerMethodHandler("session.detach", + (id, params) -> respond(server, id, Map.of("success", true))); server.registerMethodHandler("runtime.shutdown", (id, params) -> respond(server, id, Map.of())); ready.complete(server); } catch (IOException e) { diff --git a/java/src/test/java/com/github/copilot/McpAndAgentsTest.java b/java/src/test/java/com/github/copilot/McpAndAgentsTest.java index f39e56eab..0d5602447 100644 --- a/java/src/test/java/com/github/copilot/McpAndAgentsTest.java +++ b/java/src/test/java/com/github/copilot/McpAndAgentsTest.java @@ -452,11 +452,7 @@ void testShouldAcceptDefaultAgentConfigurationOnSessionResume() throws Exception assertNotNull(session.getSessionId()); String sessionId = session.getSessionId(); - // Do not call session.close() here — that invokes session.destroy on the - // server, - // which removes the session and causes the subsequent resumeSession to fail - // with "Session not found". The session handle is simply abandoned and the - // server-side session remains alive for the resume call below. + // Keep the original attachment alive while testing a concurrent resume. CopilotSession resumedSession = client.resumeSession(sessionId, new ResumeSessionConfig().setOnPermissionRequest(PermissionHandler.APPROVE_ALL) diff --git a/java/src/test/java/com/github/copilot/McpAuthInterestRegistrationTest.java b/java/src/test/java/com/github/copilot/McpAuthInterestRegistrationTest.java index 06ac08a2a..b83cedc70 100644 --- a/java/src/test/java/com/github/copilot/McpAuthInterestRegistrationTest.java +++ b/java/src/test/java/com/github/copilot/McpAuthInterestRegistrationTest.java @@ -255,7 +255,10 @@ private static JsonNode resultFor(String method, JsonNode params) { } case "session.eventLog.registerInterest" -> result.put("id", "interest-1"); case "session.options.update" -> result.put("success", true); - case "session.skills.reload", "session.destroy" -> { + case "session.skills.reload" -> { + } + case "session.detach" -> { + result.put("success", true); } default -> throw new IllegalStateException("Unexpected RPC method " + method); } diff --git a/java/src/test/java/com/github/copilot/TimeoutEdgeCaseTest.java b/java/src/test/java/com/github/copilot/TimeoutEdgeCaseTest.java index 17e1851bb..f7172f009 100644 --- a/java/src/test/java/com/github/copilot/TimeoutEdgeCaseTest.java +++ b/java/src/test/java/com/github/copilot/TimeoutEdgeCaseTest.java @@ -5,6 +5,7 @@ package com.github.copilot; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.ByteArrayOutputStream; @@ -64,7 +65,7 @@ public int read() throws IOException { * completed by a stale timeout. *

* Contract: {@code close()} shuts down the timeout scheduler before the - * blocking {@code session.destroy} RPC call, so any pending timeout task is + * blocking {@code session.detach} RPC call, so any pending timeout task is * cancelled and the future remains incomplete (not exceptionally completed with * {@code TimeoutException}). */ @@ -79,9 +80,9 @@ void testTimeoutDoesNotFireAfterSessionClose() throws Exception { assertFalse(result.isDone(), "Future should be pending before timeout fires"); - // close() blocks up to 5s on session.destroy RPC. The 2s timeout + // close() blocks up to 5s on session.detach RPC. The 2s timeout // fires during that window with the current per-call scheduler. - session.close(); + assertThrows(IllegalStateException.class, session::close); assertFalse(result.isDone(), "Future should not be completed by a timeout after session is closed. " + "The per-call ScheduledExecutorService leaked a TimeoutException."); @@ -126,6 +127,7 @@ void testSendAndWaitReusesTimeoutThread() throws Exception { result1.cancel(true); result2.cancel(true); + assertThrows(IllegalStateException.class, session::close); } } finally { rpc.close(); diff --git a/java/src/test/java/com/github/copilot/ZeroTimeoutContractTest.java b/java/src/test/java/com/github/copilot/ZeroTimeoutContractTest.java index 3d986566d..999ccad5f 100644 --- a/java/src/test/java/com/github/copilot/ZeroTimeoutContractTest.java +++ b/java/src/test/java/com/github/copilot/ZeroTimeoutContractTest.java @@ -21,6 +21,21 @@ */ public class ZeroTimeoutContractTest { + @SuppressWarnings("unchecked") + @Test + void closeShouldPropagateDetachFailureAndRemainTerminal() { + var mockRpc = mock(JsonRpcClient.class); + when(mockRpc.invoke(eq("session.detach"), any(), any())).thenReturn( + CompletableFuture.completedFuture(new CopilotSession.SessionDetachResponse(false, "cleanup failed"))); + var session = new CopilotSession("detach-failure-test", mockRpc); + + var error = assertThrows(IllegalStateException.class, session::close); + + assertTrue(error.getMessage().contains("cleanup failed")); + assertThrows(IllegalStateException.class, () -> session.send("test")); + assertDoesNotThrow(session::close); + } + @SuppressWarnings("unchecked") @Test void sendAndWaitWithZeroTimeoutShouldNotTimeOut() throws Exception { @@ -31,9 +46,9 @@ void sendAndWaitWithZeroTimeoutShouldNotTimeOut() throws Exception { var mockRpc = mock(JsonRpcClient.class); when(mockRpc.invoke(any(), any(), any())).thenAnswer(invocation -> { Object method = invocation.getArgument(0); - if ("session.destroy".equals(method)) { - // Make session.close() non-blocking by completing destroy immediately - return CompletableFuture.completedFuture(null); + if ("session.detach".equals(method)) { + // Make session.close() non-blocking by completing detach immediately + return CompletableFuture.completedFuture(new CopilotSession.SessionDetachResponse(true, null)); } // For other calls (e.g., message send), return an incomplete future so the // sendAndWait result does not complete due to a mock response. diff --git a/nodejs/src/client.ts b/nodejs/src/client.ts index c30b2207b..a64dbcd03 100644 --- a/nodejs/src/client.ts +++ b/nodejs/src/client.ts @@ -1406,20 +1406,7 @@ export class CopilotClient { if (Object.keys(patch).length === 0) { return; } - try { - await session.rpc.options.update(patch); - } catch (e) { - // The runtime session exists but the post-create options - // patch failed — best-effort disconnect so we don't leak - // it (in empty mode it would otherwise keep running with - // permissive defaults). - try { - await session.disconnect(); - } catch { - // Swallow: original error is the one the caller needs. - } - throw e; - } + await session.rpc.options.update(patch); } async createSession(config: SessionConfig): Promise { @@ -1470,6 +1457,11 @@ export class CopilotClient { managedSettingsEnabled: config.enableManagedSettings === true || config.managedSettings !== undefined, + onDisconnected: (disconnectedSession) => { + if (this.sessions.get(sessionId) === disconnectedSession) { + this.sessions.delete(sessionId); + } + }, } ); s.registerTools(config.tools); @@ -1507,19 +1499,19 @@ export class CopilotClient { let session: CopilotSession | undefined; let registeredId: string | undefined; - - // Pre-register non-cloud sessions BEFORE issuing the RPC so any - // session-scoped requests the CLI emits during `session.create` - // processing (e.g. sessionFs.writeFile for workspace metadata) can be - // routed to the correct handlers. - if (localSessionId !== undefined) { - session = initializeSession(localSessionId); - registeredId = localSessionId; - } - - const toolFilterOptions = this.resolveToolFilterOptions(config); + let createdSessionId: string | undefined; try { + // Pre-register non-cloud sessions BEFORE issuing the RPC so any + // session-scoped requests the CLI emits during `session.create` + // processing (e.g. sessionFs.writeFile for workspace metadata) can be + // routed to the correct handlers. + if (localSessionId !== undefined) { + registeredId = localSessionId; + session = initializeSession(localSessionId); + } + + const toolFilterOptions = this.resolveToolFilterOptions(config); const response = await this.connection!.sendRequest("session.create", { ...(await getTraceContext(this.onGetTraceContext)), model: config.model, @@ -1626,15 +1618,17 @@ export class CopilotClient { throw new Error("session.create response did not include a sessionId"); } if (localSessionId !== undefined && localSessionId !== returnedSessionId) { + createdSessionId = returnedSessionId; throw new Error( `session.create returned sessionId ${returnedSessionId} but the caller requested ${localSessionId}` ); } + createdSessionId = returnedSessionId; if (session === undefined) { // Cloud / server-assigned path: register the session now that // the CLI has told us which id it chose. - session = initializeSession(returnedSessionId); registeredId = returnedSessionId; + session = initializeSession(returnedSessionId); } if (config.onMcpAuthRequest) { await this.connection!.sendRequest("session.eventLog.registerInterest", { @@ -1647,9 +1641,37 @@ export class CopilotClient { await this.updateSessionOptionsForMode(session, config); } catch (e) { + let cleanupFailed = false; + let cleanupError: unknown; + if (createdSessionId !== undefined) { + try { + if (session?.sessionId === createdSessionId) { + await session.disconnect(); + } else { + const response = (await this.connection!.sendRequest("session.detach", { + sessionId: createdSessionId, + })) as { success: boolean; error?: string }; + if (!response.success) { + throw new Error( + `Failed to detach session ${createdSessionId}: ${response.error ?? "unknown error"}` + ); + } + } + } catch (error) { + cleanupFailed = true; + cleanupError = error; + } + } if (registeredId !== undefined) { this.sessions.delete(registeredId); } + if (cleanupFailed) { + throw new AggregateError( + [e, cleanupError], + "Session creation failed and the created session could not be detached", + { cause: e } + ); + } throw e; } @@ -1713,6 +1735,11 @@ export class CopilotClient { mcpAuthHandler: config.onMcpAuthRequest, managedSettingsEnabled: config.enableManagedSettings === true || config.managedSettings !== undefined, + onDisconnected: (disconnectedSession) => { + if (this.sessions.get(sessionId) === disconnectedSession) { + this.sessions.delete(sessionId); + } + }, } ); session.registerTools(config.tools); @@ -1760,11 +1787,12 @@ export class CopilotClient { session.on(config.onEvent); } this.sessions.set(sessionId, session); - this.setupSessionFs(session, config); - const toolFilterOptions = this.resolveToolFilterOptions(config); + let resumedOnServer = false; try { + this.setupSessionFs(session, config); + const toolFilterOptions = this.resolveToolFilterOptions(config); const response = await this.connection!.sendRequest("session.resume", { ...(await getTraceContext(this.onGetTraceContext)), sessionId, @@ -1861,6 +1889,7 @@ export class CopilotClient { enableManagedSettings: config.enableManagedSettings, managedSettings: config.managedSettings, }); + resumedOnServer = true; const { workspacePath, capabilities, openCanvases } = response as { sessionId: string; @@ -1880,7 +1909,24 @@ export class CopilotClient { await this.updateSessionOptionsForMode(session, config); } catch (e) { + let cleanupFailed = false; + let cleanupError: unknown; + if (resumedOnServer) { + try { + await session.disconnect(); + } catch (error) { + cleanupFailed = true; + cleanupError = error; + } + } this.sessions.delete(sessionId); + if (cleanupFailed) { + throw new AggregateError( + [e, cleanupError], + "Session resume failed and the attachment could not be detached", + { cause: e } + ); + } throw e; } @@ -2063,7 +2109,6 @@ export class CopilotClient { `Please update your SDK or server to ensure compatibility.` ); } - this.negotiatedProtocolVersion = serverVersion; } diff --git a/nodejs/src/session.ts b/nodejs/src/session.ts index 5cc49fb75..0af724275 100644 --- a/nodejs/src/session.ts +++ b/nodejs/src/session.ts @@ -427,6 +427,7 @@ export class CopilotSession { private _capabilities: SessionCapabilities = {}; private openCanvasInstances: OpenCanvasInstance[] = []; private disconnected = false; + private readonly onDisconnected?: (session: CopilotSession) => void; /** @internal Client session API handlers, populated by CopilotClient during create/resume. */ clientSessionApis: ClientSessionApiHandlers = {}; @@ -606,23 +607,35 @@ export class CopilotSession { private connection: MessageConnection, private _workspacePath?: string, traceContextProvider?: TraceContextProvider, - options?: { mcpAuthHandler?: McpAuthHandler; managedSettingsEnabled?: boolean } + options?: { + mcpAuthHandler?: McpAuthHandler; + managedSettingsEnabled?: boolean; + onDisconnected?: (session: CopilotSession) => void; + } ) { this.traceContextProvider = traceContextProvider; this.mcpAuthHandler = options?.mcpAuthHandler; this.managedSettingsEnabled = options?.managedSettingsEnabled === true; + this.onDisconnected = options?.onDisconnected; } /** * Typed session-scoped RPC methods. */ get rpc(): ReturnType { + this.ensureConnected(); if (!this._rpc) { this._rpc = createSessionRpc(this.connection, this.sessionId); } return this._rpc; } + private ensureConnected(): void { + if (this.disconnected) { + throw new Error(`Session ${this.sessionId} has been disconnected`); + } + } + /** * Path to the session workspace directory when infinite sessions are enabled. * Contains checkpoints/, plan.md, and files/ subdirectories. @@ -682,6 +695,7 @@ export class CopilotSession { async send(prompt: string): Promise; async send(options: MessageOptions): Promise; async send(optionsOrPrompt: MessageOptions | string): Promise { + this.ensureConnected(); const options: MessageOptions = typeof optionsOrPrompt === "string" ? { prompt: optionsOrPrompt } : optionsOrPrompt; const response = await this.connection.sendRequest("session.send", { @@ -1922,6 +1936,7 @@ export class CopilotSession { * ``` */ async getEvents(): Promise { + this.ensureConnected(); const response = await this.connection.sendRequest("session.getMessages", { sessionId: this.sessionId, }); @@ -1954,10 +1969,19 @@ export class CopilotSession { if (this.disconnected) { return; } - await this.connection.sendRequest("session.destroy", { - sessionId: this.sessionId, - }); + let response: { success: boolean; error?: string } = { success: false }; + for (let attempt = 0; attempt < 2 && !response.success; attempt++) { + response = (await this.connection.sendRequest("session.detach", { + sessionId: this.sessionId, + })) as { success: boolean; error?: string }; + } + if (!response.success) { + throw new Error( + `Failed to disconnect session ${this.sessionId}: ${response.error || "Unknown error"}` + ); + } this._markDisconnected(); + this.onDisconnected?.(this); } /** Enables `await using session = ...` syntax for automatic cleanup. */ @@ -1986,6 +2010,7 @@ export class CopilotSession { * ``` */ async abort(): Promise { + this.ensureConnected(); await this.connection.sendRequest("session.abort", { sessionId: this.sessionId, }); diff --git a/nodejs/test/client.test.ts b/nodejs/test/client.test.ts index 01a97e980..4ddac71b9 100644 --- a/nodejs/test/client.test.ts +++ b/nodejs/test/client.test.ts @@ -2143,6 +2143,289 @@ describe("CopilotClient", () => { spy.mockRestore(); }); + describe("session disconnect", () => { + it("detaches a session without destroying it", async () => { + const onDisconnected = vi.fn(); + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.detach") { + return { success: true }; + } + if (method === "session.getMessages") { + return { events: [] }; + } + throw new Error(`unexpected method ${method}`); + }); + const session = new CopilotSession( + "test-session", + { sendRequest } as any, + undefined, + undefined, + { + onDisconnected, + } + ); + + await expect(session.getEvents()).resolves.toEqual([]); + await expect(session.disconnect()).resolves.toBeUndefined(); + await expect(session.getEvents()).rejects.toThrow("has been disconnected"); + await expect(session.disconnect()).resolves.toBeUndefined(); + + expect(sendRequest).toHaveBeenCalledWith("session.detach", { + sessionId: "test-session", + }); + expect(sendRequest).not.toHaveBeenCalledWith("session.destroy", expect.anything()); + expect( + sendRequest.mock.calls.filter(([method]) => method === "session.detach") + ).toHaveLength(1); + expect(onDisconnected).toHaveBeenCalledTimes(1); + }); + + it("leaves a session connected when detach fails so it can be retried", async () => { + let detachResponse: { success: boolean; error?: string } = { + success: false, + error: "detach failed", + }; + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.detach") { + return detachResponse; + } + if (method === "session.getMessages") { + return { events: [] }; + } + throw new Error(`unexpected method ${method}`); + }); + const session = new CopilotSession( + "test-session", + { sendRequest } as any, + undefined, + undefined + ); + + await expect(session.disconnect()).rejects.toThrow("detach failed"); + await expect(session.getEvents()).resolves.toEqual([]); + + detachResponse = { success: true }; + await expect(session.disconnect()).resolves.toBeUndefined(); + expect( + sendRequest.mock.calls.filter(([method]) => method === "session.detach") + ).toHaveLength(3); + }); + + it("retries an unsuccessful detach response before disconnecting", async () => { + const sendRequest = vi + .fn() + .mockResolvedValueOnce({ success: false, error: "cleanup raced" }) + .mockResolvedValueOnce({ success: true }); + const session = new CopilotSession( + "test-session", + { sendRequest } as any, + undefined, + undefined + ); + + await expect(session.disconnect()).resolves.toBeUndefined(); + await expect(session.getEvents()).rejects.toThrow("has been disconnected"); + expect(sendRequest).toHaveBeenCalledTimes(2); + }); + + it("detaches a session when asynchronously disposed", async () => { + const sendRequest = vi.fn(async () => ({ success: true })); + const session = new CopilotSession("test-session", { sendRequest } as any, undefined); + + await session[Symbol.asyncDispose](); + + expect(sendRequest).toHaveBeenCalledWith("session.detach", { + sessionId: "test-session", + }); + }); + + it("removes a detached session from the client routing map", async () => { + const client = new CopilotClient(); + const sendRequest = vi.fn(async (method: string, params: any) => { + if (method === "session.create") { + return { sessionId: params.sessionId }; + } + if (method === "session.detach") { + return { success: true }; + } + throw new Error(`unexpected method ${method}`); + }); + (client as any).connection = { sendRequest }; + (client as any).state = "connected"; + + const session = await client.createSession({ + onPermissionRequest: approveAll, + }); + expect((client as any).sessions.get(session.sessionId)).toBe(session); + + await session.disconnect(); + + expect((client as any).sessions.has(session.sessionId)).toBe(false); + }); + + it("detaches a newly created session when initialization fails", async () => { + const client = new CopilotClient({ + mode: "empty", + baseDirectory: "/tmp/copilot-test", + }); + const sendRequest = vi.fn(async (method: string, params: any) => { + if (method === "session.create") { + return { sessionId: params.sessionId }; + } + if (method === "session.options.update") { + throw new Error("options failed"); + } + if (method === "session.detach") { + return { success: true }; + } + throw new Error(`unexpected method ${method}`); + }); + (client as any).connection = { sendRequest }; + (client as any).state = "connected"; + + await expect( + client.createSession({ + onPermissionRequest: approveAll, + availableTools: [], + }) + ).rejects.toThrow("options failed"); + + const sessionId = sendRequest.mock.calls.find( + ([method]) => method === "session.create" + )?.[1].sessionId; + expect(sendRequest).toHaveBeenCalledWith("session.detach", { sessionId }); + expect((client as any).sessions.size).toBe(0); + }); + + it("unregisters a created session when initialization and rollback both fail", async () => { + const client = new CopilotClient({ + mode: "empty", + baseDirectory: "/tmp/copilot-test", + }); + const sendRequest = vi.fn(async (method: string, params: any) => { + if (method === "session.create") { + return { sessionId: params.sessionId }; + } + if (method === "session.options.update") { + throw new Error("options failed"); + } + if (method === "session.detach") { + throw new Error("detach failed"); + } + throw new Error(`unexpected method ${method}`); + }); + (client as any).connection = { sendRequest }; + (client as any).state = "connected"; + + await expect( + client.createSession({ + onPermissionRequest: approveAll, + availableTools: [], + }) + ).rejects.toThrow( + "Session creation failed and the created session could not be detached" + ); + + expect((client as any).sessions.size).toBe(0); + }); + + it("detaches a cloud session when session filesystem initialization fails", async () => { + const client = new CopilotClient({ + sessionFs: { + initialCwd: "/", + sessionStatePath: "/tmp/copilot-test", + conventions: "posix", + }, + }); + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.create") { + return { sessionId: "cloud-session" }; + } + if (method === "session.detach") { + return { success: true }; + } + throw new Error(`unexpected method ${method}`); + }); + (client as any).connection = { sendRequest }; + (client as any).state = "connected"; + + await expect( + client.createSession({ + cloud: {}, + onPermissionRequest: approveAll, + }) + ).rejects.toThrow("createSessionFsProvider is required"); + + expect(sendRequest).toHaveBeenCalledWith("session.detach", { + sessionId: "cloud-session", + }); + expect((client as any).sessions.size).toBe(0); + }); + + it("detaches a resumed session when initialization fails", async () => { + const client = new CopilotClient({ + mode: "empty", + baseDirectory: "/tmp/copilot-test", + }); + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.resume") { + return { sessionId: "test-session" }; + } + if (method === "session.options.update") { + throw new Error("options failed"); + } + if (method === "session.detach") { + return { success: true }; + } + throw new Error(`unexpected method ${method}`); + }); + (client as any).connection = { sendRequest }; + (client as any).state = "connected"; + + await expect( + client.resumeSession("test-session", { + onPermissionRequest: approveAll, + availableTools: [], + }) + ).rejects.toThrow("options failed"); + + expect(sendRequest).toHaveBeenCalledWith("session.detach", { + sessionId: "test-session", + }); + expect((client as any).sessions.size).toBe(0); + }); + + it("unregisters a resumed session when initialization and detach both fail", async () => { + const client = new CopilotClient({ + mode: "empty", + baseDirectory: "/tmp/copilot-test", + }); + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.resume") { + return { sessionId: "test-session" }; + } + if (method === "session.options.update") { + throw new Error("options failed"); + } + if (method === "session.detach") { + throw new Error("detach failed"); + } + throw new Error(`unexpected method ${method}`); + }); + (client as any).connection = { sendRequest }; + (client as any).state = "connected"; + + await expect( + client.resumeSession("test-session", { + onPermissionRequest: approveAll, + availableTools: [], + }) + ).rejects.toThrow("Session resume failed and the attachment could not be detached"); + expect((client as any).sessions.size).toBe(0); + expect((client as any).sessions.size).toBe(0); + }); + }); + describe("URL parsing", () => { it("should parse port-only URL format", () => { const client = new CopilotClient({ @@ -3587,6 +3870,44 @@ describe("CopilotClient", () => { }); describe("shutdown", () => { + it("detaches all active sessions", async () => { + const client = new CopilotClient(); + const sendRequest = vi.fn(async (method: string) => { + if (method === "session.detach") { + return { success: true }; + } + throw new Error(`unexpected method ${method}`); + }); + (client as any).connection = { + sendRequest, + dispose: vi.fn(), + }; + (client as any).isExternalServer = true; + (client as any).state = "connected"; + + const created = new CopilotSession( + "created-session", + (client as any).connection, + undefined + ); + const resumed = new CopilotSession( + "resumed-session", + (client as any).connection, + undefined + ); + (client as any).sessions.set(created.sessionId, created); + (client as any).sessions.set(resumed.sessionId, resumed); + + await expect(client.stop()).resolves.toEqual([]); + + expect(sendRequest).toHaveBeenCalledWith("session.detach", { + sessionId: created.sessionId, + }); + expect(sendRequest).toHaveBeenCalledWith("session.detach", { + sessionId: resumed.sessionId, + }); + }); + it("requests runtime shutdown when stopping an SDK-owned process", async () => { const client = new CopilotClient(); const calls: string[] = []; diff --git a/nodejs/test/e2e/client.e2e.test.ts b/nodejs/test/e2e/client.e2e.test.ts index 89489f78e..53871fb90 100644 --- a/nodejs/test/e2e/client.e2e.test.ts +++ b/nodejs/test/e2e/client.e2e.test.ts @@ -102,7 +102,7 @@ describe("Client", () => { expect(errors[0].message).toContain("Failed to disconnect session"); } }, - // Generous timeout: client.stop() must wait for session.destroy to time out + // Generous timeout: client.stop() must wait for session.detach to time out // when the server process is dead. The default 30s can flake on slow CI under load. 60_000 ); diff --git a/nodejs/test/e2e/client_options.e2e.test.ts b/nodejs/test/e2e/client_options.e2e.test.ts index e3dc41343..77d9cff46 100644 --- a/nodejs/test/e2e/client_options.e2e.test.ts +++ b/nodejs/test/e2e/client_options.e2e.test.ts @@ -110,6 +110,11 @@ function handleMessage(message) { return; } + if (message.method === "session.detach") { + writeResponse(message.id, { success: true }); + return; + } + writeResponse(message.id, {}); } diff --git a/nodejs/test/e2e/multi-client.e2e.test.ts b/nodejs/test/e2e/multi-client.e2e.test.ts index a44ceec3c..c26b797d8 100644 --- a/nodejs/test/e2e/multi-client.e2e.test.ts +++ b/nodejs/test/e2e/multi-client.e2e.test.ts @@ -327,7 +327,7 @@ describe("Multi-client broadcast", async () => { }); // Client 2 resumes with ephemeral_tool - await client2.resumeSession(session1.sessionId, { + const session2 = await client2.resumeSession(session1.sessionId, { onPermissionRequest: approveAll, tools: [toolB], }); @@ -343,28 +343,16 @@ describe("Multi-client broadcast", async () => { }); expect(ephemeralResponse?.data.content).toContain("EPHEMERAL_test2"); - // Disconnect client 2 without destroying the shared session. - // Suppress "Connection is disposed" rejections that occur when the server - // broadcasts events (e.g. tool_changed_notice) to the now-dead connection. - const suppressDisposed = (reason: unknown) => { - if (reason instanceof Error && reason.message.includes("Connection is disposed")) { - return; - } - throw reason; - }; - process.on("unhandledRejection", suppressDisposed); - await client2.forceStop(); - - // Give the server time to process the connection close and remove tools + // Detach client 2's session without destroying client 1's live session. + await session2.disconnect(); + + // Give the server time to remove client 2's tools. await new Promise((resolve) => setTimeout(resolve, 500)); - process.removeListener("unhandledRejection", suppressDisposed); - // Recreate client2 for cleanup in afterAll (but don't rejoin the session) - client2 = new CopilotClient({ - connection: RuntimeConnection.forUri(`localhost:${runtimePort}`, { - connectionToken: tcpConnectionToken, - }), - }); + const toolMetadata = await session1.rpc.tools.getCurrentMetadata(); + const toolNames = toolMetadata.tools?.map((tool) => tool.name) ?? []; + expect(toolNames).toContain("stable_tool"); + expect(toolNames).not.toContain("ephemeral_tool"); // Now only stable_tool should be available const afterResponse = await session1.sendAndWait({ diff --git a/nodejs/test/e2e/session.e2e.test.ts b/nodejs/test/e2e/session.e2e.test.ts index d99a3e392..e9ab7fdfd 100644 --- a/nodejs/test/e2e/session.e2e.test.ts +++ b/nodejs/test/e2e/session.e2e.test.ts @@ -112,7 +112,7 @@ describe("Sessions", () => { ]); await session.disconnect(); - await expect(() => session.getEvents()).rejects.toThrow(/Session not found/); + await expect(() => session.getEvents()).rejects.toThrow(/has been disconnected/); }); // TODO: Re-enable once test harness CAPI proxy supports this test's session lifecycle @@ -334,7 +334,7 @@ describe("Sessions", () => { // All can be disconnected await Promise.all([s1.disconnect(), s2.disconnect(), s3.disconnect()]); for (const s of [s1, s2, s3]) { - await expect(() => s.getEvents()).rejects.toThrow(/Session not found/); + await expect(() => s.getEvents()).rejects.toThrow(/has been disconnected/); } }); diff --git a/nodejs/test/toolSet.test.ts b/nodejs/test/toolSet.test.ts index b77b79707..97b335d07 100644 --- a/nodejs/test/toolSet.test.ts +++ b/nodejs/test/toolSet.test.ts @@ -406,6 +406,7 @@ describe("Empty-mode safe defaults", () => { if (method === "session.options.update") { throw new Error("update rejected"); } + if (method === "session.detach") return { success: true }; throw new Error(`Unexpected method: ${method}`); } ); diff --git a/python/copilot/client.py b/python/copilot/client.py index 21ceb6eee..afb64d373 100644 --- a/python/copilot/client.py +++ b/python/copilot/client.py @@ -2648,6 +2648,7 @@ def _initialize_session(sid: str) -> CopilotSession: workspace_path=None, managed_settings_enabled=enable_managed_settings is True or managed_settings is not None, + on_disconnected=self._unregister_session, ) if self._session_fs_config: if create_session_fs_handler is None: @@ -3318,6 +3319,7 @@ async def resume_session( workspace_path=None, managed_settings_enabled=enable_managed_settings is True or managed_settings is not None, + on_disconnected=self._unregister_session, ) if self._session_fs_config: if create_session_fs_handler is None: @@ -4555,6 +4557,11 @@ def _get_session(self, session_id: str) -> CopilotSession | None: with self._sessions_lock: return self._sessions.get(session_id) + def _unregister_session(self, session: CopilotSession) -> None: + with self._sessions_lock: + if self._sessions.get(session.session_id) is session: + del self._sessions[session.session_id] + async def _set_llm_inference_provider(self) -> None: if self._request_handler is None or self._rpc is None: return diff --git a/python/copilot/session.py b/python/copilot/session.py index 92c24bdd8..3c532c268 100644 --- a/python/copilot/session.py +++ b/python/copilot/session.py @@ -1507,6 +1507,7 @@ def __init__( client: Any, workspace_path: os.PathLike[str] | str | None = None, managed_settings_enabled: bool = False, + on_disconnected: Callable[[CopilotSession], None] | None = None, ): """ Initialize a new CopilotSession. @@ -1522,6 +1523,7 @@ def __init__( (when infinite sessions enabled). managed_settings_enabled: Whether managed settings were enabled when creating or resuming the session. + on_disconnected: Callback invoked after the session disconnects successfully. """ self.session_id = session_id self._managed_settings_enabled = managed_settings_enabled @@ -1559,10 +1561,18 @@ def __init__( self._open_canvases_lock = threading.Lock() self._rpc: SessionRpc | None = None self._destroyed = False + self._disconnect_lock = asyncio.Lock() + self._on_disconnected = on_disconnected + + def _ensure_connected(self) -> None: + with self._event_handlers_lock: + if self._destroyed: + raise RuntimeError(f"Session {self.session_id} has been disconnected") @property def rpc(self) -> SessionRpc: """Typed session-scoped RPC methods.""" + self._ensure_connected() if self._rpc is None: self._rpc = SessionRpc(self._client, self.session_id) return self._rpc @@ -1643,6 +1653,7 @@ async def send( ... attachments=[{"type": "file", "path": "./src/main.py"}], ... ) """ + self._ensure_connected() params: dict[str, Any] = { "sessionId": self.session_id, "prompt": prompt, @@ -2889,6 +2900,7 @@ async def get_events(self) -> list[SessionEvent]: ... case AssistantMessageData() as data: ... print(f"Assistant: {data.content}") """ + self._ensure_connected() response = await self._client.request("session.getMessages", {"sessionId": self.session_id}) # Convert dict events to SessionEvent objects events_dicts = response["events"] @@ -2917,18 +2929,18 @@ async def disconnect(self) -> None: >>> # Clean up when done — session can still be resumed later >>> await session.disconnect() """ - # Ensure that the check and update of _destroyed are atomic so that - # only the first caller proceeds to send the destroy RPC. - with self._event_handlers_lock: - if self._destroyed: - return - self._destroyed = True + async with self._disconnect_lock: + with self._event_handlers_lock: + if self._destroyed: + return + + response = await self._client.request("session.detach", {"sessionId": self.session_id}) + if not response.get("success"): + detail = response.get("error") or "unknown error" + raise RuntimeError(f"Failed to detach session {self.session_id}: {detail}") - try: - await self._client.request("session.destroy", {"sessionId": self.session_id}) - finally: - # Clear handlers even if the request fails. with self._event_handlers_lock: + self._destroyed = True self._event_handlers.clear() with self._tool_handlers_lock: self._tool_handlers.clear() @@ -2942,6 +2954,10 @@ async def disconnect(self) -> None: self._exit_plan_mode_handler = None with self._auto_mode_switch_handler_lock: self._auto_mode_switch_handler = None + on_disconnected = self._on_disconnected + self._on_disconnected = None + if on_disconnected is not None: + on_disconnected(self) async def __aenter__(self) -> CopilotSession: """Enable use as an async context manager.""" diff --git a/python/e2e/test_client_options_e2e.py b/python/e2e/test_client_options_e2e.py index fe1ed5482..df72bfd59 100644 --- a/python/e2e/test_client_options_e2e.py +++ b/python/e2e/test_client_options_e2e.py @@ -168,6 +168,10 @@ def _get_available_port() -> int: writeResponse(message.id, { success: true }); return; } + if (message.method === "session.detach") { + writeResponse(message.id, { success: true }); + return; + } writeResponse(message.id, {}); } diff --git a/python/e2e/test_session_e2e.py b/python/e2e/test_session_e2e.py index b6f173f75..7d666582e 100644 --- a/python/e2e/test_session_e2e.py +++ b/python/e2e/test_session_e2e.py @@ -36,7 +36,7 @@ async def test_should_create_and_disconnect_sessions(self, ctx: E2ETestContext): await session.disconnect() - with pytest.raises(Exception, match="Session not found"): + with pytest.raises(RuntimeError, match="has been disconnected"): await session.get_events() async def test_should_have_stateful_conversation(self, ctx: E2ETestContext): diff --git a/python/test_client.py b/python/test_client.py index 2375bc98a..5db976205 100644 --- a/python/test_client.py +++ b/python/test_client.py @@ -6,6 +6,7 @@ import asyncio import inspect +import threading from datetime import UTC, datetime from tempfile import TemporaryDirectory from unittest.mock import AsyncMock, Mock, patch @@ -2605,6 +2606,77 @@ async def test_aexit_calls_disconnect(self): mock_disconnect.assert_awaited_once() +class TestCopilotSessionDisconnect: + @pytest.mark.asyncio + async def test_concurrent_disconnect_sends_one_request(self): + from copilot.session import CopilotSession + + client = Mock() + client.request = AsyncMock(return_value={"success": True}) + session = CopilotSession("session-id", client) + + await asyncio.gather(session.disconnect(), session.disconnect()) + + client.request.assert_awaited_once_with("session.detach", {"sessionId": "session-id"}) + + @pytest.mark.asyncio + async def test_failed_disconnect_can_be_retried(self): + from copilot.session import CopilotSession + + client = Mock() + client.request = AsyncMock( + side_effect=[ + {"success": False, "error": "temporary failure"}, + {"success": True}, + ] + ) + session = CopilotSession("session-id", client) + + with pytest.raises(RuntimeError, match="temporary failure"): + await session.disconnect() + await session.disconnect() + + assert client.request.await_count == 2 + + @pytest.mark.asyncio + async def test_successful_disconnect_unregisters_session(self): + from copilot.session import CopilotSession + + rpc_client = Mock() + rpc_client.request = AsyncMock(return_value={"success": True}) + sdk_client = CopilotClient.__new__(CopilotClient) + sdk_client._sessions = {} + sdk_client._sessions_lock = threading.Lock() + session = CopilotSession( + "session-id", + rpc_client, + on_disconnected=sdk_client._unregister_session, + ) + sdk_client._sessions[session.session_id] = session + + await session.disconnect() + + assert sdk_client._get_session(session.session_id) is None + + @pytest.mark.asyncio + async def test_disconnected_session_rejects_session_operations_locally(self): + from copilot.session import CopilotSession + + client = Mock() + client.request = AsyncMock(return_value={"success": True}) + session = CopilotSession("session-id", client) + await session.disconnect() + client.request.reset_mock() + + with pytest.raises(RuntimeError, match="Session session-id has been disconnected"): + await session.send("hello") + with pytest.raises(RuntimeError, match="Session session-id has been disconnected"): + await session.get_events() + with pytest.raises(RuntimeError, match="Session session-id has been disconnected"): + _ = session.rpc + client.request.assert_not_awaited() + + class TestCustomAgentWireFormat: def test_model_field_is_forwarded_in_wire_format(self): """The model key in CustomAgentConfig should appear as 'model' in the wire payload.""" diff --git a/rust/src/errors.rs b/rust/src/errors.rs index 6e05bbfae..c772a87f0 100644 --- a/rust/src/errors.rs +++ b/rust/src/errors.rs @@ -152,6 +152,9 @@ pub enum SessionErrorKind { /// Session ID returned by the CLI. returned: SessionId, }, + + /// The CLI could not detach the session. + DetachFailed, } impl fmt::Display for SessionErrorKind { @@ -186,6 +189,7 @@ impl fmt::Display for SessionErrorKind { f, "CLI returned session ID {returned} after SDK registered {requested}" ), + SessionErrorKind::DetachFailed => write!(f, "failed to detach session"), } } } @@ -397,7 +401,7 @@ fn capture_backtrace() -> Option> { /// /// `Client::stop` performs cooperative shutdown across every active /// session before killing the CLI child process. Errors from any -/// per-session `session.destroy` RPC and from the terminal child-kill +/// per-session `session.detach` RPC and from the terminal child-kill /// step are collected here rather than short-circuiting on the first /// failure, so callers see the full picture of what went wrong during /// teardown. @@ -409,7 +413,7 @@ pub struct StopErrors(pub(crate) Vec); impl StopErrors { /// Borrow the collected errors as a slice, in the order they - /// occurred (per-session destroys first, then child-kill last). + /// occurred (per-session detaches first, then child-kill last). pub fn errors(&self) -> &[Error] { &self.0 } diff --git a/rust/src/lib.rs b/rust/src/lib.rs index cafa3c596..5c54e77da 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -968,6 +968,12 @@ fn validate_inprocess_options(options: &ClientOptions) -> Result<()> { Ok(()) } +#[derive(serde::Deserialize)] +struct SessionDetachResponse { + success: bool, + error: Option, +} + /// Connection to a GitHub Copilot CLI server (stdio, TCP, or external). /// /// Cheaply cloneable — cloning shares the underlying connection. @@ -1911,6 +1917,25 @@ impl Client { self.call_with_inline_callback(method, params, None).await } + async fn detach_session(&self, session_id: &str) -> Result<()> { + let value = self + .call( + "session.detach", + Some(serde_json::json!({ "sessionId": session_id })), + ) + .await?; + let response: SessionDetachResponse = serde_json::from_value(value)?; + if response.success { + return Ok(()); + } + Err(Error::with_message( + ErrorKind::Session(SessionErrorKind::DetachFailed), + response + .error + .unwrap_or_else(|| "unknown error".to_string()), + )) + } + /// Same as [`call`](Self::call), but installs an `inline_callback` /// that runs synchronously on the JSON-RPC read task the instant the /// successful response is parsed, before it is delivered to this @@ -2217,12 +2242,7 @@ impl Client { let mut first_error = None; for session_id in self.inner.router.session_ids() { - if let Err(error) = self - .call( - "session.destroy", - Some(serde_json::json!({ "sessionId": session_id })), - ) - .await + if let Err(error) = self.detach_session(&session_id).await && first_error.is_none() { first_error = Some(error); @@ -2347,22 +2367,22 @@ impl Client { /// Cooperatively shut down the client and the CLI child process. /// - /// Walks every still-registered session and sends `session.destroy` + /// Walks every still-registered session and sends `session.detach` /// for each one, asks SDK-owned runtimes to shut down, then kills the - /// CLI child. Errors from per-session destroys, runtime shutdown, and + /// CLI child. Errors from per-session detaches, runtime shutdown, and /// the final child-kill are collected into /// [`StopErrors`] rather than short-circuiting on the first failure /// — so callers see the full picture of teardown. /// /// If you have already called [`Session::disconnect`] on every - /// session this client created, the per-session destroy step is a + /// session this client created, the per-session detach step is a /// no-op (the router map is empty); only the child-kill remains. /// /// [`Session::disconnect`]: crate::session::Session::disconnect /// /// # Cancel safety /// - /// **Cancel-unsafe but recoverable.** The body sequentially destroys + /// **Cancel-unsafe but recoverable.** The body sequentially detaches /// every registered session (each via [`Client::call`](Self::call), /// individually cancel-safe) before killing the child. Cancelling /// `stop()` mid-loop leaves some sessions still in the router map @@ -2377,21 +2397,15 @@ impl Client { let mut errors: Vec = Vec::new(); // Snapshot the registered session IDs without holding the router - // lock across the destroy RPCs. + // lock across the detach RPCs. for session_id in self.inner.router.session_ids() { - match self - .call( - "session.destroy", - Some(serde_json::json!({ "sessionId": session_id })), - ) - .await - { + match self.detach_session(&session_id).await { Ok(_) => {} Err(e) => { warn!( session_id = %session_id, error = %e, - "session.destroy failed during Client::stop", + "session.detach failed during Client::stop", ); errors.push(e); } diff --git a/rust/src/session.rs b/rust/src/session.rs index c6c806b1c..fef5797ea 100644 --- a/rust/src/session.rs +++ b/rust/src/session.rs @@ -554,7 +554,7 @@ impl Session { /// Disconnect this session from the CLI. /// - /// Sends the `session.destroy` RPC, stops the event loop, and unregisters + /// Sends the `session.detach` RPC, stops the event loop, and unregisters /// the session from the client. **Session state on disk** (conversation /// history, planning state, artifacts) is **preserved**, so the /// conversation can be resumed later via [`Client::resume_session`] @@ -569,20 +569,13 @@ impl Session { /// [`Client::delete_session`]: crate::Client::delete_session /// [`send_and_wait`]: Self::send_and_wait pub async fn disconnect(&self) -> Result<(), Error> { - self.client - .call( - "session.destroy", - Some(serde_json::json!({ "sessionId": self.id })), - ) - .await?; + self.client.detach_session(&self.id).await?; self.stop_event_loop().await; self.client.unregister_session(&self.id); Ok(()) } - /// Deprecated alias for [`disconnect`](Self::disconnect). The - /// underlying wire RPC happens to be named `session.destroy`, but it - /// only severs the connection — on-disk session state is preserved. + /// Deprecated alias for [`disconnect`](Self::disconnect). /// Prefer `disconnect` in new code. #[deprecated(since = "0.1.0", note = "Use `disconnect()` instead")] pub async fn destroy(&self) -> Result<(), Error> { diff --git a/rust/tests/e2e/client_options.rs b/rust/tests/e2e/client_options.rs index fc1ceebb8..aa67d8163 100644 --- a/rust/tests/e2e/client_options.rs +++ b/rust/tests/e2e/client_options.rs @@ -491,6 +491,10 @@ function handleMessage(message) { writeResponse(message.id, { success: true }); return; } + if (message.method === "session.detach") { + writeResponse(message.id, { success: true }); + return; + } writeResponse(message.id, {}); } diff --git a/rust/tests/session_test.rs b/rust/tests/session_test.rs index 727911081..41f8794f8 100644 --- a/rust/tests/session_test.rs +++ b/rust/tests/session_test.rs @@ -1384,7 +1384,7 @@ async fn session_rpc_methods_send_correct_method_names() { let cases: Vec<(&str, Option<&str>)> = vec![ ("session.abort", None), ("session.log", Some("message")), - ("session.destroy", None), + ("session.detach", None), ]; for (expected_method, extra_param_key) in cases { @@ -1393,7 +1393,7 @@ async fn session_rpc_methods_send_correct_method_names() { match expected_method { "session.abort" => s.abort().await.map(|_| ()), "session.log" => s.log("test msg", None).await, - "session.destroy" => s.disconnect().await, + "session.detach" => s.disconnect().await, _ => unreachable!(), } }); @@ -1411,6 +1411,7 @@ async fn session_rpc_methods_send_correct_method_names() { "session.log" => { serde_json::json!({ "eventId": "00000000-0000-0000-0000-000000000000" }) } + "session.detach" => serde_json::json!({ "success": true }), _ => serde_json::json!({}), }; server.respond(&request, response).await; @@ -4045,9 +4046,9 @@ async fn rpc_namespace_client_models_list_dispatches_correctly() { } #[tokio::test] -async fn client_stop_sends_session_destroy_for_each_active_session() { +async fn client_stop_sends_session_detach_for_each_active_session() { // One client, two registered sessions. Client::stop must send - // session.destroy for each before returning Ok. + // session.detach for each before returning Ok. let (client, server_read, server_write) = make_client(); let mut server = FakeServer { @@ -4097,31 +4098,33 @@ async fn client_stop_sends_session_destroy_for_each_active_session() { .await; let _session_b = timeout(TIMEOUT, create_b).await.unwrap(); - // Drive Client::stop and respond to each destroy in turn. + // Drive Client::stop and respond to each detach in turn. let stop_handle = tokio::spawn({ let client = client.clone(); async move { client.stop().await } }); - let mut destroyed = Vec::new(); + let mut detached = Vec::new(); for _ in 0..2 { let req = server.read_request().await; - assert_eq!(req["method"], "session.destroy"); - destroyed.push(req["params"]["sessionId"].as_str().unwrap().to_string()); - server.respond(&req, serde_json::json!(null)).await; + assert_eq!(req["method"], "session.detach"); + detached.push(req["params"]["sessionId"].as_str().unwrap().to_string()); + server + .respond(&req, serde_json::json!({ "success": true })) + .await; } - destroyed.sort(); + detached.sort(); let mut expected = [session_id_a.clone(), session_id_b.clone()]; expected.sort(); - assert_eq!(destroyed, expected); + assert_eq!(detached, expected); let stop_result = timeout(TIMEOUT, stop_handle).await.unwrap().unwrap(); assert!(stop_result.is_ok(), "stop returned errors: {stop_result:?}"); } #[tokio::test] -async fn client_stop_aggregates_session_destroy_errors() { - // session.destroy fails on the wire — Client::stop returns +async fn client_stop_aggregates_session_detach_errors() { + // session.detach fails on the wire — Client::stop returns // StopErrors carrying the failure rather than short-circuiting. let (session, mut server) = create_session_pair().await; let client = session.client().clone(); @@ -4129,7 +4132,7 @@ async fn client_stop_aggregates_session_destroy_errors() { let stop_handle = tokio::spawn(async move { client.stop().await }); let req = server.read_request().await; - assert_eq!(req["method"], "session.destroy"); + assert_eq!(req["method"], "session.detach"); let id = req["id"].as_u64().unwrap(); let response = serde_json::json!({ "jsonrpc": "2.0",