diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 417923a179e..bfdb849e6dc 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -67,6 +67,20 @@ Use this to detect mismatch pre-flight: read `workspaceCwd` off `/capabilities` Attaches to existing sessions are NOT counted toward the cap, so an idle daemon's reconnects keep working even when at-capacity. +`RestoreInProgressError` — only emitted by `POST /session/:id/load` and `POST /session/:id/resume` — returns `409` with a `Retry-After: 5` header (matching `session_limit_exceeded`) and: + +```json +{ + "error": "Session \"\" is already being restored via session/; retry session/ after it completes", + "code": "restore_in_progress", + "sessionId": "", + "activeAction": "load", + "requestedAction": "resume" +} +``` + +Fired when a `session/load` is issued for an id that already has a `session/resume` in flight (or vice versa). Wait at least `Retry-After` seconds and retry — the underlying restore completes within `initTimeoutMs` (default 10s). Same-action races (`load` vs `load`, `resume` vs `resume`) coalesce instead of erroring. + ## Capabilities The daemon advertises its supported feature tags from the serve capability @@ -75,12 +89,15 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design ``` ['health', 'capabilities', 'session_create', 'session_scope_override', + 'session_load', 'unstable_session_resume', 'session_list', 'session_prompt', 'session_cancel', 'session_events', 'session_set_model', 'permission_vote'] ``` `session_scope_override` is the negotiation handle for the per-request `sessionScope` field on `POST /session` (see below). Older daemons silently ignore the field, so SDK clients should pre-flight `caps.features` for this tag before sending it. +`session_load` and `unstable_session_resume` advertise the explicit-restore routes (`POST /session/:id/load` and `POST /session/:id/resume`). Older daemons return `404` for these paths, so SDK clients should pre-flight `caps.features` before calling. The `unstable_` prefix on `unstable_session_resume` mirrors the underlying ACP method (`connection.unstable_resumeSession`) — the daemon's wire shape is committed for v1, but the ACP method name itself may change before ACP marks resume stable. + ## Routes > **Stage 1 limitation — no `DELETE /session/:id`.** Sessions live until @@ -174,6 +191,60 @@ Concurrent `POST /session` calls for the same workspace are **coalesced** to one > event (covers the spawn-time `model_switch_failed` even if the > subscribe lands a few ms after the create response). +### `POST /session/:id/load` + +Restore a persisted ACP session by id and replay its history through SSE. The path id is authoritative; any `sessionId` field in the body is ignored. Pre-flight `caps.features.session_load` — older daemons return `404` for this route. + +Request: + +```json +{ + "cwd": "/absolute/path/to/workspace" +} +``` + +| Field | Required | Notes | +| ----- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `cwd` | no | Same canonicalization + `workspace_mismatch` rules as `POST /session`. Omit to inherit `/capabilities.workspaceCwd`. `mcpServers` is intentionally NOT accepted here — daemon-wide MCP is settings-driven (matches `POST /session`). | + +Response: + +```json +{ + "sessionId": "persisted-1", + "workspaceCwd": "/canonical/path", + "attached": false, + "state": { + "models": { ... }, + "modes": { ... }, + "configOptions": [ ... ] + } +} +``` + +`state` mirrors ACP's `LoadSessionResponse` — `models` is a `SessionModelState`, `modes` a `SessionModeState`, `configOptions` an array of `SessionConfigOption`. Missing fields are agent-decided. Late attachers (the `attached: true` paths below) get the SAME `state` snapshot the original load caller saw — the daemon caches it on the entry; runtime mutations (e.g. `model_switched`) are delivered on the SSE stream, not on subsequent attach responses. + +`attached: true` means the session was already live (either from a prior `session/load`/`session/resume`, or because a coalesced concurrent caller raced just ahead). + +**History replay over SSE.** While `loadSession` is in flight on the agent side, the agent emits `session_update` notifications for every persisted turn. The daemon buffers them onto the session's event-bus before the route response returns, so subscribers that immediately call `GET /session/:id/events` with `Last-Event-ID: 0` see the full replay. **The replay ring is bounded** (default 4000 frames per session). Long histories with many tool-call / thought-stream turns can exceed that — the oldest frames are dropped silently. Clients that need full history should subscribe immediately after `load` returns; alternatively they can persist the SSE event ids and use `Last-Event-ID` to resume from a later turn boundary. + +**Errors:** + +- `404` — persisted session id doesn't exist (`SessionNotFoundError`). +- `400` — `workspace_mismatch` (same shape as `POST /session`). +- `503` — `session_limit_exceeded` (counts against `--max-sessions`; in-flight restores are accounted for too). +- `409` — `restore_in_progress` (a `session/resume` for the same id is already in flight). `Retry-After: 5`. Same-action races (two concurrent `session/load` for the same id) coalesce — exactly one returns `attached: false`, the rest return `attached: true` with the same `state`. + +### `POST /session/:id/resume` + +Restore a persisted ACP session by id WITHOUT replaying history through SSE. The model context is restored internally on the agent side (via `geminiClient.initialize` reading `config.getResumedSessionData`); the SSE stream stays clean for clients that already have history rendered. Pre-flight `caps.features.unstable_session_resume`. + +Same request shape as `/load`. Same response shape — `state` mirrors ACP's `ResumeSessionResponse`. Same error envelope, including `409 restore_in_progress` (which fires when a `session/load` is in flight; `session/resume` racing behind another `session/resume` coalesces). + +Use `/load` when the client has no history rendered (cold reconnect, picker → open). Use `/resume` when the client already has the turns on screen and only needs the daemon-side handle back. + +> ⚠️ **Why `unstable_` on the capability tag?** The route is wire-stable for the daemon's v1, but it's backed by ACP's `connection.unstable_resumeSession` which is still subject to ACP-side breaking changes. The daemon insulates the wire shape from those changes; the prefix is a courtesy signal so SDK consumers know the underlying agent contract is not yet locked. + ### `GET /workspace/:id/sessions` List all live sessions whose canonical workspace matches `:id` (URL-encoded absolute cwd). diff --git a/docs/users/qwen-serve.md b/docs/users/qwen-serve.md index fb2610464b6..80655dc269a 100644 --- a/docs/users/qwen-serve.md +++ b/docs/users/qwen-serve.md @@ -173,16 +173,48 @@ To host **multiple workspaces** (one user, several repos; or several users on th To handle multiple **users** (each with their own quota, audit log, sandbox) or to scale beyond one process's reach (cold-start budget, FD count, RSS), spawn one daemon per workspace per user behind an external orchestrator. That orchestrator (multi-tenancy / OIDC / Quota / Audit / k8s) is **out of scope** for the qwen-code project — see issue [#3803](https://github.com/QwenLM/qwen-code/issues/3803) "External Reference Architecture" for the design pointers. +## Loading and resuming a persisted session + +The daemon exposes ACP's `session/load` and `session/unstable_resumeSession` over HTTP via two routes: + +| Route | Use when | +| -------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `POST /session/:id/load` | The client has **no** history rendered (cold reconnect, picker-then-open). The daemon replays every persisted turn through SSE so subscribers see the full transcript. Capability tag: `session_load`. | +| `POST /session/:id/resume` | The client already has the turns on screen and only needs the daemon-side handle back. Model context is restored on the agent side without UI replay — the SSE stream stays clean. Capability tag: `unstable_session_resume`. | + +The TypeScript SDK exposes both as static factories on `DaemonSessionClient`: + +```ts +import { DaemonClient, DaemonSessionClient } from '@qwen-code/sdk'; + +const client = new DaemonClient({ baseUrl: 'http://127.0.0.1:4170' }); + +// Cold reconnect — daemon will replay history through SSE. +const session = await DaemonSessionClient.load(client, 'persisted-id'); + +// Or, if your UI already has the history, skip the replay: +// const session = await DaemonSessionClient.resume(client, 'persisted-id'); + +for await (const event of session.events()) { + // First the replayed `session_update` frames (load only), + // then live events. +} +``` + +Pre-flight `caps.features.session_load` / `caps.features.unstable_session_resume` before calling — older daemons return `404`. Concurrent same-action requests for the same id coalesce; cross-action races (a `load` racing a `resume`) get `409 restore_in_progress` with `Retry-After: 5`. See the [protocol reference](../developers/qwen-serve-protocol.md) for the full error envelope. + +Note: history replay is bounded by the SSE ring (default 4000 frames). Long histories with chatty turns can exceed that — earliest frames are dropped silently. For very long sessions, prefer `resume` and rely on the client's local persisted UI. + ## Durability model -**Sessions are ephemeral in Stage 1.** Plan accordingly: +**Sessions are still ephemeral in Stage 1 across daemon restarts**, but persisted sessions on disk can be reloaded: -- A child process crash publishes `session_died` and removes the session from the daemon's maps. There is **no resume** — clients must `POST /session` again. -- A daemon restart loses every in-flight session. ACP's `loadSession` / `unstable_resumeSession` are **not exposed via HTTP** in Stage 1; sessions don't outlive the daemon. -- Long client disconnects (>5 min on a chatty turn) can outrun the SSE replay ring (default 4000 frames) — `Last-Event-ID` reconnect succeeds but state may be incoherent. For mobile / flaky-network clients, plan to re-create the session and re-open SSE on long drops. +- A child process crash publishes `session_died` and removes the live session from the daemon's maps. The persisted on-disk session **can** be reloaded via `POST /session/:id/load` if a fresh agent child is spawnable. +- A daemon restart loses every in-flight live session. The persisted sessions remain on disk and can be loaded against a new daemon process, subject to the same workspace binding rules. +- Long client disconnects (>5 min on a chatty turn) can outrun the SSE replay ring (default 4000 frames) — `Last-Event-ID` reconnect succeeds but state may be incoherent. For mobile / flaky-network clients, plan to re-open SSE on long drops or call `POST /session/:id/load` to replay from disk. - File operations (`writeTextFile`) are atomic across crashes (write-then-rename); they aren't atomic across daemon restarts in the sense of replaying — the file write either landed or it didn't. -If your integration needs cross-restart durability, you need either Stage 1.5+ (`loadSession` over HTTP, persistence layer) or your own application-level state recovery. Don't hold long-running, restart-sensitive state inside the daemon's session. +If your integration needs server-side cross-restart durability beyond what `session/load` covers (e.g. server-managed retry queues), you still need application-level state recovery. Don't hold long-running, restart-sensitive state inside the daemon's session. ## Stage 1.5+ runtime guarantees diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index 450aecb67d4..45f25783ac6 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -60,6 +60,11 @@ vi.mock('@agentclientprotocol/sdk', () => ({ Object.assign(err, data); return err; }); + static resourceNotFound = vi.fn().mockImplementation((uri: string) => { + const err = new Error(`Resource not found: ${uri}`); + Object.assign(err, { code: -32002, data: { uri } }); + return err; + }); }, PROTOCOL_VERSION: '1.0.0', })); @@ -1681,3 +1686,297 @@ describe('QwenAgent extMethod renameSession routing', () => { await agentPromise; }); }); + +// Tests for QwenAgent.loadSession() and QwenAgent.unstable_resumeSession() +// — locks the session-existence guard, the resourceNotFound error contract, +// and the resume-vs-load semantic difference (load replays UI history, +// resume does not). +describe('QwenAgent loadSession / unstable_resumeSession', () => { + let capturedAgentFactory: + | ((conn: { closed: Promise }) => { + loadSession: (args: Record) => Promise; + unstable_resumeSession: ( + args: Record, + ) => Promise; + }) + | undefined; + + let mockConfig: Config; + let lastSessionMock: + | { + getId: ReturnType; + sendAvailableCommandsUpdate: ReturnType; + replayHistory: ReturnType; + installRewriter: ReturnType; + } + | undefined; + let processExitSpy: MockInstance; + let stdinDestroySpy: MockInstance; + let stdoutDestroySpy: MockInstance; + + const mockArgv = {} as CliArgs; + + beforeEach(() => { + vi.clearAllMocks(); + mockConnectionState.reset(); + lastSessionMock = undefined; + capturedAgentFactory = undefined; + + vi.mocked(AgentSideConnection).mockImplementation((factory: unknown) => { + capturedAgentFactory = factory as typeof capturedAgentFactory; + return { + get closed() { + return mockConnectionState.promise; + }, + } as unknown as InstanceType; + }); + + mockConfig = { + initialize: vi.fn().mockResolvedValue(undefined), + waitForMcpReady: vi.fn().mockResolvedValue(undefined), + getHookSystem: vi.fn().mockReturnValue(undefined), + getDisableAllHooks: vi.fn().mockReturnValue(false), + hasHooksForEvent: vi.fn().mockReturnValue(false), + getModel: vi.fn().mockReturnValue('test-model'), + getModelsConfig: vi.fn().mockReturnValue({ + getCurrentAuthType: vi.fn().mockReturnValue('api-key'), + }), + refreshAuth: vi.fn().mockResolvedValue(undefined), + } as unknown as Config; + + processExitSpy = vi + .spyOn(process, 'exit') + .mockImplementation((() => undefined) as unknown as typeof process.exit); + stdinDestroySpy = vi + .spyOn(process.stdin, 'destroy') + .mockImplementation(() => process.stdin); + stdoutDestroySpy = vi + .spyOn(process.stdout, 'destroy') + .mockImplementation(() => process.stdout); + }); + + afterEach(() => { + processExitSpy.mockRestore(); + stdinDestroySpy.mockRestore(); + stdoutDestroySpy.mockRestore(); + }); + + function makeRestoreInnerConfig( + opts: { + resumedConversation?: { messages: unknown[] }; + } = {}, + ) { + return { + initialize: vi.fn().mockResolvedValue(undefined), + waitForMcpReady: vi.fn().mockResolvedValue(undefined), + getModelsConfig: vi.fn().mockReturnValue({ + getCurrentAuthType: vi.fn().mockReturnValue('api-key'), + }), + refreshAuth: vi.fn().mockResolvedValue(undefined), + getModel: vi.fn().mockReturnValue('m'), + getContentGeneratorConfig: vi.fn().mockReturnValue({}), + getAvailableModels: vi.fn().mockReturnValue([]), + getModes: vi.fn().mockReturnValue([]), + getApprovalMode: vi.fn().mockReturnValue('default'), + getSessionId: vi.fn().mockReturnValue('persisted-1'), + getAuthType: vi.fn().mockReturnValue('api-key'), + getAllConfiguredModels: vi.fn().mockReturnValue([]), + getGeminiClient: vi.fn().mockReturnValue({ + isInitialized: vi.fn().mockReturnValue(true), + initialize: vi.fn().mockResolvedValue(undefined), + waitForMcpReady: vi.fn().mockResolvedValue(undefined), + }), + getFileSystemService: vi.fn().mockReturnValue(undefined), + setFileSystemService: vi.fn(), + getHookSystem: vi.fn().mockReturnValue(undefined), + getDisableAllHooks: vi.fn().mockReturnValue(true), + hasHooksForEvent: vi.fn().mockReturnValue(false), + // load path reads back the persisted conversation here and feeds + // it to `session.replayHistory`. resume path doesn't read this. + getResumedSessionData: vi + .fn() + .mockReturnValue( + opts.resumedConversation + ? { conversation: opts.resumedConversation } + : undefined, + ), + }; + } + + function makeRestoreSettings() { + return { + merged: { mcpServers: {} }, + getUserHooks: vi.fn().mockReturnValue({}), + getProjectHooks: vi.fn().mockReturnValue({}), + } as unknown as LoadedSettings; + } + + function bindRestoreMocks(opts: { + sessionExists: boolean; + resumedConversation?: { messages: unknown[] }; + }) { + const innerConfig = makeRestoreInnerConfig({ + resumedConversation: opts.resumedConversation, + }); + vi.mocked(loadSettings).mockReturnValue(makeRestoreSettings()); + vi.mocked(loadCliConfig).mockResolvedValue( + innerConfig as unknown as Config, + ); + vi.mocked(SessionService).mockImplementation( + () => + ({ + sessionExists: vi.fn().mockResolvedValue(opts.sessionExists), + }) as unknown as InstanceType, + ); + vi.mocked(Session).mockImplementation(() => { + const sessionMock = { + getId: vi.fn().mockReturnValue('persisted-1'), + getConfig: vi.fn().mockReturnValue(innerConfig), + sendAvailableCommandsUpdate: vi.fn().mockResolvedValue(undefined), + replayHistory: vi.fn().mockResolvedValue(undefined), + installRewriter: vi.fn(), + }; + lastSessionMock = sessionMock; + return sessionMock as unknown as InstanceType; + }); + return innerConfig; + } + + async function spawnAgent() { + const agentPromise = runAcpAgent( + mockConfig, + makeRestoreSettings(), + mockArgv, + ); + await vi.waitFor(() => expect(capturedAgentFactory).toBeDefined()); + const agent = capturedAgentFactory!({ + get closed() { + return mockConnectionState.promise; + }, + }); + return { agent, agentPromise }; + } + + it('loadSession throws resourceNotFound when the persisted session is missing', async () => { + bindRestoreMocks({ sessionExists: false }); + const { agent, agentPromise } = await spawnAgent(); + + await expect( + agent.loadSession({ + cwd: '/tmp', + sessionId: 'persisted-missing', + mcpServers: [], + }), + ).rejects.toMatchObject({ + code: -32002, + data: { uri: 'session:persisted-missing' }, + }); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('loadSession returns LoadSessionResponse and replays history on the session', async () => { + bindRestoreMocks({ + sessionExists: true, + resumedConversation: { + messages: [{ role: 'user', parts: [{ text: 'hi' }] }], + }, + }); + const { agent, agentPromise } = await spawnAgent(); + + const response = await agent.loadSession({ + cwd: '/tmp', + sessionId: 'persisted-1', + mcpServers: [], + }); + + expect(response).toMatchObject({ + modes: expect.anything(), + models: expect.anything(), + configOptions: expect.anything(), + }); + // load semantic: history MUST be replayed so SSE subscribers see + // the persisted turns. + expect(lastSessionMock?.replayHistory).toHaveBeenCalledWith([ + { role: 'user', parts: [{ text: 'hi' }] }, + ]); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('loadSession skips history replay when getResumedSessionData() returns undefined', async () => { + // Distinct code path: `createAndStoreSession(config, undefined)` + // takes the no-conversation branch, so `replayHistory` must + // NOT be called even though the persisted session existed + // (covers the case where the on-disk record has a session row + // but no resumable conversation, e.g. corrupted / partially + // written history). + bindRestoreMocks({ sessionExists: true /* no resumedConversation */ }); + const { agent, agentPromise } = await spawnAgent(); + + const response = await agent.loadSession({ + cwd: '/tmp', + sessionId: 'persisted-1', + mcpServers: [], + }); + + expect(response).toMatchObject({ + modes: expect.anything(), + models: expect.anything(), + configOptions: expect.anything(), + }); + expect(lastSessionMock?.replayHistory).not.toHaveBeenCalled(); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('unstable_resumeSession throws resourceNotFound when the persisted session is missing', async () => { + bindRestoreMocks({ sessionExists: false }); + const { agent, agentPromise } = await spawnAgent(); + + await expect( + agent.unstable_resumeSession({ + cwd: '/tmp', + sessionId: 'persisted-missing', + }), + ).rejects.toMatchObject({ + code: -32002, + data: { uri: 'session:persisted-missing' }, + }); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('unstable_resumeSession returns the response without replaying history', async () => { + bindRestoreMocks({ + sessionExists: true, + resumedConversation: { + messages: [{ role: 'user', parts: [{ text: 'hi' }] }], + }, + }); + const { agent, agentPromise } = await spawnAgent(); + + const response = await agent.unstable_resumeSession({ + cwd: '/tmp', + sessionId: 'persisted-1', + }); + + expect(response).toMatchObject({ + modes: expect.anything(), + models: expect.anything(), + configOptions: expect.anything(), + }); + // resume semantic: model context is restored internally via + // geminiClient.initialize(), but UI replay is NOT triggered — + // the SSE stream stays clean for clients that already have the + // history rendered. + expect(lastSessionMock?.replayHistory).not.toHaveBeenCalled(); + + mockConnectionState.resolve(); + await agentPromise; + }); +}); diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index 355b13d35d1..926f1ec1b1b 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -48,6 +48,8 @@ import type { NewSessionResponse, PromptRequest, PromptResponse, + ResumeSessionRequest, + ResumeSessionResponse, SessionConfigOption, SessionInfo, SessionModeState, @@ -337,12 +339,19 @@ class QwenAgent implements Agent { return sessionService.sessionExists(params.sessionId); }, ); + if (!exists) { + throw RequestError.resourceNotFound(`session:${params.sessionId}`); + } const config = await this.newSessionConfig( params.cwd, - params.mcpServers, + // `LoadSessionRequest.mcpServers` is required in today's ACP + // schema, but mirror `unstable_resumeSession` and tolerate a + // future loosening — `newSessionConfig` iterates the list, so + // a `null`/`undefined` would otherwise throw `TypeError`. + params.mcpServers ?? [], params.sessionId, - exists, + true, ); await this.ensureAuthenticated(config); this.setupFileSystem(config); @@ -361,6 +370,43 @@ class QwenAgent implements Agent { }; } + async unstable_resumeSession( + params: ResumeSessionRequest, + ): Promise { + const exists = await runWithAcpRuntimeOutputDir( + this.settings, + params.cwd, + async () => { + const sessionService = new SessionService(params.cwd); + return sessionService.sessionExists(params.sessionId); + }, + ); + if (!exists) { + throw RequestError.resourceNotFound(`session:${params.sessionId}`); + } + + const config = await this.newSessionConfig( + params.cwd, + params.mcpServers ?? [], + params.sessionId, + true, + ); + await this.ensureAuthenticated(config); + this.setupFileSystem(config); + + await this.createAndStoreSession(config); + + const modesData = this.buildModesData(config); + const availableModels = this.buildAvailableModels(config); + const configOptions = this.buildConfigOptions(config); + + return { + modes: modesData, + models: availableModels, + configOptions, + }; + } + async unstable_listSessions( params: ListSessionsRequest, ): Promise { diff --git a/packages/cli/src/serve/capabilities.ts b/packages/cli/src/serve/capabilities.ts index c0b161c8c9d..6cf182fd7ba 100644 --- a/packages/cli/src/serve/capabilities.ts +++ b/packages/cli/src/serve/capabilities.ts @@ -27,6 +27,11 @@ export const SERVE_CAPABILITY_REGISTRY = { capabilities: { since: 'v1' }, session_create: { since: 'v1' }, session_scope_override: { since: 'v1' }, + session_load: { since: 'v1' }, + // ACP backs this with `connection.unstable_resumeSession`. Surface + // the unstable prefix so clients don't pin against a `v1` shape that + // the underlying ACP method may still change. + unstable_session_resume: { since: 'v1' }, session_list: { since: 'v1' }, session_prompt: { since: 'v1' }, session_cancel: { since: 'v1' }, diff --git a/packages/cli/src/serve/httpAcpBridge.test.ts b/packages/cli/src/serve/httpAcpBridge.test.ts index bfeda3f6064..7b6f04fbeeb 100644 --- a/packages/cli/src/serve/httpAcpBridge.test.ts +++ b/packages/cli/src/serve/httpAcpBridge.test.ts @@ -12,6 +12,7 @@ import * as path from 'node:path'; import { AgentSideConnection, PROTOCOL_VERSION, + RequestError, ndJsonStream, } from '@agentclientprotocol/sdk'; import type { @@ -27,6 +28,8 @@ import type { NewSessionResponse, PromptRequest, PromptResponse, + ResumeSessionRequest, + ResumeSessionResponse, SetSessionConfigOptionRequest, SetSessionConfigOptionResponse, SetSessionModeRequest, @@ -37,6 +40,7 @@ import { InvalidPermissionOptionError, InvalidSessionScopeError, MAX_WORKSPACE_PATH_LENGTH, + RestoreInProgressError, SessionNotFoundError, WorkspaceMismatchError, type AcpChannel, @@ -93,10 +97,20 @@ interface FakeAgentOpts { p: NewSessionRequest, self: FakeAgent, ) => Promise | NewSessionResponse; + loadSessionImpl?: ( + p: LoadSessionRequest, + self: FakeAgent, + ) => Promise | LoadSessionResponse; + resumeSessionImpl?: ( + p: ResumeSessionRequest, + self: FakeAgent, + ) => Promise | ResumeSessionResponse; } class FakeAgent implements Agent { newSessionCalls: NewSessionRequest[] = []; + loadSessionCalls: LoadSessionRequest[] = []; + resumeSessionCalls: ResumeSessionRequest[] = []; promptCalls: PromptRequest[] = []; cancelCalls: CancelNotification[] = []; constructor(private readonly opts: FakeAgentOpts = {}) {} @@ -129,8 +143,21 @@ class FakeAgent implements Agent { return { sessionId: `${prefix}:${p.cwd}${suffix}` }; } - async loadSession(_p: LoadSessionRequest): Promise { - throw new Error('not implemented in test fake'); + async loadSession(p: LoadSessionRequest): Promise { + this.loadSessionCalls.push(p); + if (this.opts.loadSessionImpl) { + return this.opts.loadSessionImpl(p, this); + } + return {}; + } + async unstable_resumeSession( + p: ResumeSessionRequest, + ): Promise { + this.resumeSessionCalls.push(p); + if (this.opts.resumeSessionImpl) { + return this.opts.resumeSessionImpl(p, this); + } + return {}; } async authenticate(_p: AuthenticateRequest): Promise { throw new Error('not implemented in test fake'); @@ -282,6 +309,561 @@ describe('createHttpAcpBridge', () => { await bridge.shutdown(); }); + it('loads an existing ACP session and registers it for daemon routes', async () => { + const handles: ChannelHandle[] = []; + const factory: ChannelFactory = async () => { + const h = makeChannel({ + loadSessionImpl: () => ({ configOptions: [] }), + }); + handles.push(h); + return h.channel; + }; + const bridge = makeBridge({ channelFactory: factory }); + + const loaded = await bridge.loadSession({ + sessionId: 'persisted-1', + workspaceCwd: WS_A, + }); + + expect(loaded).toEqual({ + sessionId: 'persisted-1', + workspaceCwd: WS_A, + attached: false, + state: { configOptions: [] }, + }); + expect(handles[0]?.agent.loadSessionCalls).toEqual([ + { sessionId: 'persisted-1', cwd: WS_A, mcpServers: [] }, + ]); + expect(bridge.sessionCount).toBe(1); + + await expect( + bridge.sendPrompt('persisted-1', { + sessionId: 'ignored', + prompt: [{ type: 'text', text: 'hi' }], + }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + expect(handles[0]?.agent.promptCalls[0]?.sessionId).toBe('persisted-1'); + + await bridge.shutdown(); + }); + + it('buffers load replay events until the restored session is registered', async () => { + let capturedConn: AgentSideConnection | undefined; + const factory: ChannelFactory = async () => { + const { clientStream, agentStream } = createInMemoryChannel(); + const fakeAgent = new FakeAgent({ + loadSessionImpl: async (p) => { + await capturedConn!.sessionUpdate({ + sessionId: p.sessionId, + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'replayed' }, + }, + }); + return {}; + }, + }); + capturedConn = new AgentSideConnection(() => fakeAgent, agentStream); + return { + stream: clientStream, + exited: new Promise< + | { exitCode: number | null; signalCode: NodeJS.Signals | null } + | undefined + >(() => {}), + kill: async () => {}, + killSync: () => {}, + }; + }; + const bridge = makeBridge({ channelFactory: factory }); + + const loaded = await bridge.loadSession({ + sessionId: 'persisted-history', + workspaceCwd: WS_A, + }); + const iterator = bridge + .subscribeEvents(loaded.sessionId, { lastEventId: 0 }) + [Symbol.asyncIterator](); + let timer: NodeJS.Timeout | undefined; + const next = await Promise.race([ + iterator.next(), + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error('timed out waiting for replay event')), + 500, + ); + }), + ]); + if (timer) clearTimeout(timer); + + expect(next.value.type).toBe('session_update'); + expect(next.value.data).toMatchObject({ + sessionId: 'persisted-history', + update: { + sessionUpdate: 'agent_message_chunk', + content: { text: 'replayed' }, + }, + }); + + await iterator.return?.(); + await bridge.shutdown(); + }); + + it('resumes an existing ACP session without calling session/load', async () => { + const handles: ChannelHandle[] = []; + const factory: ChannelFactory = async () => { + const h = makeChannel({ + resumeSessionImpl: () => ({ modes: null }), + }); + handles.push(h); + return h.channel; + }; + const bridge = makeBridge({ channelFactory: factory }); + + const resumed = await bridge.resumeSession({ + sessionId: 'persisted-2', + workspaceCwd: WS_A, + }); + + expect(resumed).toEqual({ + sessionId: 'persisted-2', + workspaceCwd: WS_A, + attached: false, + state: { modes: null }, + }); + expect(handles[0]?.agent.loadSessionCalls).toHaveLength(0); + expect(handles[0]?.agent.resumeSessionCalls).toEqual([ + { sessionId: 'persisted-2', cwd: WS_A, mcpServers: [] }, + ]); + + await bridge.shutdown(); + }); + + it('attaches to an already live session and returns the cached restore state', async () => { + const handles: ChannelHandle[] = []; + const factory: ChannelFactory = async () => { + const h = makeChannel({ + // `_meta` is the permissive escape hatch on the ACP response + // schema — any record-shaped payload survives the wire. The + // assertions only need the bridge to forward it intact. + loadSessionImpl: () => ({ _meta: { tag: 'restored-foo' } }), + }); + handles.push(h); + return h.channel; + }; + const bridge = makeBridge({ channelFactory: factory }); + + const loaded = await bridge.loadSession({ + sessionId: 'persisted-3', + workspaceCwd: WS_A, + }); + const attached = await bridge.resumeSession({ + sessionId: 'persisted-3', + workspaceCwd: WS_A, + }); + + expect(loaded.attached).toBe(false); + expect(loaded.state).toEqual({ _meta: { tag: 'restored-foo' } }); + // Late attachers must observe the SAME restore state the original + // caller saw — `entry.restoreState` is cached at load time. + expect(attached).toEqual({ + sessionId: 'persisted-3', + workspaceCwd: WS_A, + attached: true, + state: { _meta: { tag: 'restored-foo' } }, + }); + expect(handles[0]?.agent.loadSessionCalls).toHaveLength(1); + expect(handles[0]?.agent.resumeSessionCalls).toHaveLength(0); + + await bridge.shutdown(); + }); + + it('propagates the original ACP state to coalesced restore waiters', async () => { + let releaseLoad: ((value: LoadSessionResponse) => void) | undefined; + const factory: ChannelFactory = async () => + makeChannel({ + loadSessionImpl: () => + new Promise((resolve) => { + releaseLoad = resolve; + }), + }).channel; + const bridge = makeBridge({ channelFactory: factory }); + + const first = bridge.loadSession({ + sessionId: 'coalesce-state', + workspaceCwd: WS_A, + }); + // Wait for the first call to register inFlight before issuing + // the second. + for (let i = 0; i < 50 && !releaseLoad; i++) { + await new Promise((r) => setTimeout(r, 10)); + } + expect(releaseLoad).toBeDefined(); + const second = bridge.loadSession({ + sessionId: 'coalesce-state', + workspaceCwd: WS_A, + }); + + releaseLoad!({ _meta: { tag: 'restored-baz' } }); + const [r1, r2] = await Promise.all([first, second]); + + expect(r1.attached).toBe(false); + expect(r1.state).toEqual({ _meta: { tag: 'restored-baz' } }); + expect(r2.attached).toBe(true); + // Coalesced waiter sees the same state, not `{}`. + expect(r2.state).toEqual({ _meta: { tag: 'restored-baz' } }); + + await bridge.shutdown(); + }); + + it('survives spawn-owner disconnect kill while a coalesced restore is mid-flight', async () => { + let releaseLoad: ((value: LoadSessionResponse) => void) | undefined; + const factory: ChannelFactory = async () => + makeChannel({ + loadSessionImpl: () => + new Promise((resolve) => { + releaseLoad = resolve; + }), + }).channel; + const bridge = makeBridge({ channelFactory: factory }); + + const first = bridge.loadSession({ + sessionId: 'race-target', + workspaceCwd: WS_A, + }); + for (let i = 0; i < 50 && !releaseLoad; i++) { + await new Promise((r) => setTimeout(r, 10)); + } + expect(releaseLoad).toBeDefined(); + + // Second caller coalesces synchronously and reserves the attach. + const second = bridge.loadSession({ + sessionId: 'race-target', + workspaceCwd: WS_A, + }); + + releaseLoad!({}); + const r1 = await first; + expect(r1.attached).toBe(false); + + // First caller "disconnected" — simulate by issuing the same + // disconnect-cleanup the route handler would. The + // `requireZeroAttaches` guard MUST see B's reserved attach and + // skip the kill, otherwise B observes a 404'd sessionId on its + // next call. + await bridge.killSession(r1.sessionId, { requireZeroAttaches: true }); + + // The session must still be alive for B. + expect(bridge.sessionCount).toBe(1); + const r2 = await second; + expect(r2.attached).toBe(true); + expect(r2.sessionId).toBe('race-target'); + + await bridge.shutdown(); + }); + + it('does not kill the channel when the last live session leaves while a restore is pending', async () => { + let releaseLoad: ((value: LoadSessionResponse) => void) | undefined; + const handles: ChannelHandle[] = []; + const factory: ChannelFactory = async () => { + const h = makeChannel({ + loadSessionImpl: () => + new Promise((resolve) => { + releaseLoad = resolve; + }), + }); + handles.push(h); + return h.channel; + }; + const bridge = makeBridge({ channelFactory: factory }); + + // Spawn a regular session first, then kick off a slow restore on + // the same channel. + const spawned = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const restore = bridge.loadSession({ + sessionId: 'pending-restore', + workspaceCwd: WS_A, + }); + for (let i = 0; i < 50 && !releaseLoad; i++) { + await new Promise((r) => setTimeout(r, 10)); + } + expect(releaseLoad).toBeDefined(); + + // Kill the only registered session; the channel must NOT die + // because pendingRestoreIds is non-empty. + await bridge.killSession(spawned.sessionId); + expect(handles[0]?.killed).toBe(false); + + // Let the restore finish — it joins the channel as the new + // sole session. + releaseLoad!({}); + const restored = await restore; + expect(restored.sessionId).toBe('pending-restore'); + expect(bridge.sessionCount).toBe(1); + expect(handles[0]?.killed).toBe(false); + + await bridge.shutdown(); + }); + + it('does not promote a restored session into the omitted-id attach default', async () => { + const handles: ChannelHandle[] = []; + const factory: ChannelFactory = async () => { + const h = makeChannel({ + loadSessionImpl: () => ({}), + }); + handles.push(h); + return h.channel; + }; + const bridge = makeBridge({ channelFactory: factory }); + + await bridge.loadSession({ + sessionId: 'persisted-explicit', + workspaceCwd: WS_A, + }); + // A subsequent omitted-id `POST /session` (single scope) MUST + // create a fresh session rather than silently attaching to the + // explicitly restored one. + const spawned = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + expect(spawned.sessionId).not.toBe('persisted-explicit'); + expect(spawned.attached).toBe(false); + expect(bridge.sessionCount).toBe(2); + + await bridge.shutdown(); + }); + + it('maps an ACP missing persisted session to SessionNotFoundError', async () => { + const handles: ChannelHandle[] = []; + const factory: ChannelFactory = async () => { + const h = makeChannel({ + loadSessionImpl: (p) => { + throw RequestError.resourceNotFound(`session:${p.sessionId}`); + }, + }); + handles.push(h); + return h.channel; + }; + const bridge = makeBridge({ channelFactory: factory }); + + await expect( + bridge.loadSession({ + sessionId: 'missing-persisted', + workspaceCwd: WS_A, + }), + ).rejects.toMatchObject({ + name: 'SessionNotFoundError', + sessionId: 'missing-persisted', + }); + expect(bridge.sessionCount).toBe(0); + expect(handles[0]?.killed).toBe(false); + + await bridge.shutdown(); + }); + + // The `isAcpSessionResourceNotFound` `message`-fallback path can't + // be exercised through the FakeAgent end-to-end: the ACP SDK + // normalizes non-RequestError throws to `-32603 Internal error`, + // so a fake-agent thrown plain Object with `code: -32002` arrives + // at the bridge as -32603 with the original message buried under + // `data.details`. The fallback covers ACP variants that emit the + // URI in `message` directly (without `data.uri`); the primary + // `data.uri` path is covered by the test above. The exact-match + // tightening (vs. substring) is exercised by inspection. + + it('rejects load while a resume for the same session is in flight', async () => { + let releaseResume: (() => void) | undefined; + const factory: ChannelFactory = async () => + makeChannel({ + resumeSessionImpl: () => + new Promise((resolve) => { + releaseResume = () => resolve({}); + }), + }).channel; + const bridge = makeBridge({ channelFactory: factory }); + + const resume = bridge.resumeSession({ + sessionId: 'persisted-race', + workspaceCwd: WS_A, + }); + for (let i = 0; i < 50 && !releaseResume; i++) { + await new Promise((r) => setTimeout(r, 10)); + } + expect(releaseResume).toBeDefined(); + + await expect( + bridge.loadSession({ + sessionId: 'persisted-race', + workspaceCwd: WS_A, + }), + ).rejects.toBeInstanceOf(RestoreInProgressError); + + releaseResume?.(); + await resume; + await bridge.shutdown(); + }); + + it('rejects resume while a load for the same session is in flight (mirror of load-on-resume)', async () => { + let releaseLoad: (() => void) | undefined; + const factory: ChannelFactory = async () => + makeChannel({ + loadSessionImpl: () => + new Promise((resolve) => { + releaseLoad = () => resolve({}); + }), + }).channel; + const bridge = makeBridge({ channelFactory: factory }); + + const load = bridge.loadSession({ + sessionId: 'persisted-mirror', + workspaceCwd: WS_A, + }); + for (let i = 0; i < 50 && !releaseLoad; i++) { + await new Promise((r) => setTimeout(r, 10)); + } + expect(releaseLoad).toBeDefined(); + + // Resume coalescing onto load would silently subscribe the + // resume client to history-replay frames it explicitly opted + // out of; it must throw instead. + await expect( + bridge.resumeSession({ + sessionId: 'persisted-mirror', + workspaceCwd: WS_A, + }), + ).rejects.toBeInstanceOf(RestoreInProgressError); + + releaseLoad?.(); + await load; + await bridge.shutdown(); + }); + + it('does not kill a shared channel when one of multiple pending restores fails', async () => { + let releaseGood: (() => void) | undefined; + const handles: ChannelHandle[] = []; + const factory: ChannelFactory = async () => { + const h = makeChannel({ + loadSessionImpl: (p) => { + if (p.sessionId === 'bad-restore') { + throw RequestError.resourceNotFound(`session:${p.sessionId}`); + } + return new Promise((resolve) => { + releaseGood = () => resolve({}); + }); + }, + }); + handles.push(h); + return h.channel; + }; + const bridge = makeBridge({ channelFactory: factory }); + + const good = bridge.loadSession({ + sessionId: 'good-restore', + workspaceCwd: WS_A, + }); + for ( + let i = 0; + i < 50 && handles[0]?.agent.loadSessionCalls.length !== 1; + i++ + ) { + await new Promise((r) => setTimeout(r, 10)); + } + expect(handles[0]?.agent.loadSessionCalls[0]?.sessionId).toBe( + 'good-restore', + ); + + await expect( + bridge.loadSession({ + sessionId: 'bad-restore', + workspaceCwd: WS_A, + }), + ).rejects.toBeInstanceOf(SessionNotFoundError); + expect(handles[0]?.killed).toBe(false); + + releaseGood?.(); + await expect(good).resolves.toMatchObject({ + sessionId: 'good-restore', + attached: false, + }); + + await bridge.shutdown(); + }); + + it('does not surface an unhandledRejection when the channel exits after a successful restore', async () => { + // Regression for the dangling-rejection bug: `transportClosed` + // is a fresh `.then(throw)` promise per restore. If `withTimeout` + // wins the race, `transportClosed` stays pending and a later + // channel exit fires the inner `throw` with no observer attached + // — Node 22 logs `unhandledRejection`, and + // `--unhandled-rejections=throw` deployments crash the daemon. + const handles: ChannelHandle[] = []; + const factory: ChannelFactory = async () => { + const h = makeChannel({ loadSessionImpl: () => ({}) }); + handles.push(h); + return h.channel; + }; + const bridge = makeBridge({ channelFactory: factory }); + + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + try { + const restored = await bridge.loadSession({ + sessionId: 'persisted-leak', + workspaceCwd: WS_A, + }); + expect(restored.attached).toBe(false); + // Now resolve `channel.exited` AFTER the restore promise has + // already settled. `transportClosed` was the race-loser, so + // its `.then(throw)` fires now. With the `.catch(() => {})` + // suppression in place, no `unhandledRejection` is emitted; + // without it, the test would observe one. + handles[0]!.crash({ exitCode: null, signalCode: null }); + // Give the rejection a tick to surface if it were unhandled. + await new Promise((r) => setTimeout(r, 50)); + expect(unhandled).toEqual([]); + } finally { + process.off('unhandledRejection', onUnhandled); + await bridge.shutdown(); + } + }); + + it('shutdown awaits in-flight restores before resolving', async () => { + // `shutdown()` adds `inFlightRestoreAwaits` to the wait list so + // shutting the daemon down doesn't orphan a half-completed + // restore. Verify by racing the restore-settled signal against + // the shutdown-resolved signal: if shutdown is awaiting the + // restore, the restore MUST settle first (or simultaneously + // — `Promise.race` ties go to the earlier-registered handler, + // which is the restore here). + let releaseLoad: (() => void) | undefined; + const factory: ChannelFactory = async () => + makeChannel({ + loadSessionImpl: () => + new Promise((resolve) => { + releaseLoad = () => resolve({}); + }), + }).channel; + const bridge = makeBridge({ channelFactory: factory }); + + const restore = bridge.loadSession({ + sessionId: 'persisted-shutdown', + workspaceCwd: WS_A, + }); + for (let i = 0; i < 50 && !releaseLoad; i++) { + await new Promise((r) => setTimeout(r, 10)); + } + expect(releaseLoad).toBeDefined(); + + const restoreFirst = restore + .catch(() => undefined) + .then(() => 'restore' as const); + const shutdownFirst = bridge.shutdown().then(() => 'shutdown' as const); + const winner = await Promise.race([restoreFirst, shutdownFirst]); + expect(winner).toBe('restore'); + // Both must have settled cleanly by the end. + await Promise.all([restoreFirst, shutdownFirst]); + }); + it('rejects cross-workspace requests with WorkspaceMismatchError (#3803 §02)', async () => { // Per #3803 §02 (1 daemon = 1 workspace), `spawnOrAttach` calls // whose canonical `workspaceCwd` doesn't match `boundWorkspace` diff --git a/packages/cli/src/serve/httpAcpBridge.ts b/packages/cli/src/serve/httpAcpBridge.ts index d0d16e7ebc6..3918d08d5d4 100644 --- a/packages/cli/src/serve/httpAcpBridge.ts +++ b/packages/cli/src/serve/httpAcpBridge.ts @@ -23,12 +23,14 @@ import { import type { CancelNotification, Client, + LoadSessionResponse, PromptRequest, PromptResponse, ReadTextFileRequest, ReadTextFileResponse, RequestPermissionRequest, RequestPermissionResponse, + ResumeSessionResponse, SessionNotification, SetSessionModelRequest, SetSessionModelResponse, @@ -96,6 +98,20 @@ export interface BridgeSession { attached: boolean; } +export interface BridgeRestoreSessionRequest { + /** Session id to restore through ACP `session/load` or `session/resume`. */ + sessionId: string; + /** Absolute path to the workspace root the child inherits as cwd. */ + workspaceCwd: string; +} + +export type BridgeSessionState = LoadSessionResponse | ResumeSessionResponse; + +export interface BridgeRestoredSession extends BridgeSession { + /** ACP state returned by `session/load` / `session/resume`. */ + state: BridgeSessionState; +} + /** Sparse summary used by `GET /workspace/:id/sessions`. */ export interface BridgeSessionSummary { sessionId: string; @@ -109,6 +125,22 @@ export interface HttpAcpBridge { */ spawnOrAttach(req: BridgeSpawnRequest): Promise; + /** + * Load an existing persisted session and replay its history through + * session_update notifications. Returns `attached: true` when the requested + * session is already live in this daemon. + */ + loadSession(req: BridgeRestoreSessionRequest): Promise; + + /** + * Resume an existing persisted session without requesting history replay. + * Returns `attached: true` when the requested session is already live in + * this daemon. + */ + resumeSession( + req: BridgeRestoreSessionRequest, + ): Promise; + /** * Forward a prompt to the agent. Concurrent prompts against the same * session FIFO-serialize through a per-session queue (ACP guarantees @@ -246,6 +278,26 @@ export class SessionNotFoundError extends Error { } } +export class RestoreInProgressError extends Error { + readonly sessionId: string; + readonly activeAction: 'load' | 'resume'; + readonly requestedAction: 'load' | 'resume'; + + constructor( + sessionId: string, + activeAction: 'load' | 'resume', + requestedAction: 'load' | 'resume', + ) { + super( + `Session "${sessionId}" is already being restored via session/${activeAction}; retry session/${requestedAction} after it completes`, + ); + this.name = 'RestoreInProgressError'; + this.sessionId = sessionId; + this.activeAction = activeAction; + this.requestedAction = requestedAction; + } +} + /** * Thrown by `spawnOrAttach` when `req.sessionScope` is set to a value * outside the `'single' | 'thread'` enum. The HTTP route validates the @@ -500,6 +552,12 @@ interface ChannelInfo { * BkUyD invariant on `isDying` below). */ sessionIds: Set; + /** + * Restore calls currently executing on this channel but not yet registered + * in `sessionIds`. Used to avoid killing the shared channel when one pending + * restore fails while another is still healthy. + */ + pendingRestoreIds: Set; /** * MUST be set to `true` synchronously by any teardown path BEFORE * awaiting `channel.kill()`. `ensureChannel` treats a dying channel @@ -595,6 +653,14 @@ interface SessionEntry { * session. */ spawnOwnerWantedKill: boolean; + /** + * ACP state captured at `session/load` / `session/resume` time so + * late attachers (existing-byId early-return + coalesced restore + * waiters) get the same payload the original restore caller did. + * `undefined` for sessions created via `doSpawn` — those have never + * had an ACP load/resume response, so attaches return `state: {}`. + */ + restoreState?: BridgeSessionState; } interface PendingPermission { @@ -670,6 +736,9 @@ class BridgeClient implements Client { private readonly resolveEntry: ( sessionId?: string, ) => SessionEntry | undefined, + private readonly resolvePendingRestoreEvents: ( + sessionId?: string, + ) => EventBus | undefined, private readonly registerPending: (pending: PendingPermission) => void, /** * Roll back a `registerPending` call when the subsequent publish @@ -803,8 +872,10 @@ class BridgeClient implements Client { async sessionUpdate(params: SessionNotification): Promise { const entry = this.resolveEntry(params.sessionId); - if (!entry) return; - entry.events.publish({ type: 'session_update', data: params }); + const events = + entry?.events ?? this.resolvePendingRestoreEvents(params.sessionId); + if (!events) return; + events.publish({ type: 'session_update', data: params }); } async writeTextFile( @@ -1171,6 +1242,31 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // `shutdown()`. const inFlightSpawns = new Map>(); + interface InFlightRestore { + action: 'load' | 'resume'; + promise: Promise; + /** + * Synchronous reservation slot for callers that coalesce onto this + * restore. Coalescers do `count++` BEFORE awaiting `promise` so the + * spawn-owner's disconnect-reaper (`killSession({ requireZeroAttaches: + * true })`) sees a non-zero `attachCount` on the freshly registered + * entry and skips the kill. The IIFE folds this counter into + * `entry.attachCount` when it calls `createSessionEntry`. BQ9tV + * race-guard equivalent for coalesced restore waiters. + */ + coalesceState: { count: number }; + } + + // Coalesces concurrent explicit restore calls for the same session id. + // `session/load` replays history through SSE and `session/resume` restores + // context; running either twice for the same id at the same time can + // duplicate history frames or race two entries into `byId`. + const inFlightRestores = new Map(); + // `session/load` emits history replay as session_update notifications before + // the ACP request returns. Keep a temporary bus so those replay frames land in + // the ring, then promote the same bus into the registered SessionEntry. + const pendingRestoreEvents = new Map(); + const registerPending = (p: PendingPermission) => { const entry = byId.get(p.sessionId); if (!entry) { @@ -1255,6 +1351,8 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { } return undefined; }, + (sessionId) => + sessionId ? pendingRestoreEvents.get(sessionId) : undefined, registerPending, (rid) => // Roll back a register-then-publish-failed pending so the agent @@ -1281,6 +1379,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { connection, client, sessionIds: new Set(), + pendingRestoreIds: new Set(), isDying: false, }; aliveChannels.add(info); @@ -1496,20 +1595,11 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { throw new Error('HttpAcpBridge is shutting down'); } - const entry: SessionEntry = { - sessionId: newSessionResp.sessionId, - workspaceCwd: boundWorkspace, - channel: ci.channel, - connection: ci.connection, - events: new EventBus(), - promptQueue: Promise.resolve(), - modelChangeQueue: Promise.resolve(), - pendingPermissionIds: new Set(), - attachCount: 0, - spawnOwnerWantedKill: false, - }; - ci.sessionIds.add(entry.sessionId); - byId.set(entry.sessionId, entry); + const entry = createSessionEntry( + ci, + newSessionResp.sessionId, + boundWorkspace, + ); // `defaultEntry` is the single-scope attach target — only sessions // SPAWNED UNDER `'single'` may claim it. A thread-scope spawn must // never become the attach target, otherwise a later omitted-scope @@ -1683,6 +1773,308 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { return entry.transportClosedReject; }; + const resolveWorkspaceKey = (workspaceCwd: string): string => { + if (!path.isAbsolute(workspaceCwd)) { + throw new Error( + `workspaceCwd must be an absolute path; got "${workspaceCwd}"`, + ); + } + const workspaceKey = + workspaceCwd === boundWorkspace + ? boundWorkspace + : canonicalizeWorkspace(workspaceCwd); + if (workspaceKey !== boundWorkspace) { + throw new WorkspaceMismatchError(boundWorkspace, workspaceKey); + } + return workspaceKey; + }; + + const createSessionEntry = ( + ci: ChannelInfo, + sessionId: string, + workspaceCwd: string, + events = new EventBus(), + ): SessionEntry => { + const entry: SessionEntry = { + sessionId, + workspaceCwd, + channel: ci.channel, + connection: ci.connection, + events, + promptQueue: Promise.resolve(), + modelChangeQueue: Promise.resolve(), + pendingPermissionIds: new Set(), + attachCount: 0, + spawnOwnerWantedKill: false, + }; + ci.sessionIds.add(entry.sessionId); + byId.set(entry.sessionId, entry); + return entry; + }; + + const isAcpSessionResourceNotFound = ( + err: unknown, + sessionId: string, + ): boolean => { + if (!err || typeof err !== 'object') return false; + const maybe = err as { + code?: unknown; + data?: unknown; + message?: unknown; + }; + if (maybe.code !== -32002) return false; + const expectedUri = `session:${sessionId}`; + if ( + maybe.data && + typeof maybe.data === 'object' && + (maybe.data as { uri?: unknown }).uri === expectedUri + ) { + return true; + } + // Fallback for ACP servers that omit `data.uri` and embed the + // URI in the human-readable message. Use exact equality on the + // canonical "Resource not found: " form rather than + // `includes(expectedUri)` — a substring match would cause a + // sessionId of `"a"` to falsely match a message containing + // `"session:abc"`. + return ( + typeof maybe.message === 'string' && + maybe.message === `Resource not found: ${expectedUri}` + ); + }; + + async function restoreSession( + action: 'load' | 'resume', + req: BridgeRestoreSessionRequest, + ): Promise { + if (shuttingDown) { + throw new Error('HttpAcpBridge is shutting down'); + } + const workspaceKey = resolveWorkspaceKey(req.workspaceCwd); + + const existing = byId.get(req.sessionId); + if (existing) { + existing.attachCount++; + return { + sessionId: existing.sessionId, + workspaceCwd: existing.workspaceCwd, + attached: true, + // Late attachers get the same ACP state the original restore + // caller saw; spawn-only sessions don't carry a state payload. + state: existing.restoreState ?? {}, + }; + } + + const inFlight = inFlightRestores.get(req.sessionId); + if (inFlight) { + // Cross-action races BOTH ways must reject. A `resume` arriving + // while a `load` is in flight cannot quietly coalesce: the load + // is replaying full history through SSE on a shared EventBus, + // and `DaemonSessionClient.resume()` seeds `lastEventId: 0`, + // which means the resume client would receive every replayed + // frame — directly violating resume's "no UI replay" contract. + // The mirror direction (`load` onto `resume`) is rejected for + // the same reason: a load caller expects history but resume + // didn't replay any. Same-action coalescing is unaffected. + if (action !== inFlight.action) { + throw new RestoreInProgressError( + req.sessionId, + inFlight.action, + action, + ); + } + // Reserve the attach SYNCHRONOUSLY before awaiting so the spawn + // owner's `requireZeroAttaches` disconnect-reaper observes our + // intent. The IIFE folds this counter into `entry.attachCount` + // at `createSessionEntry` time. + inFlight.coalesceState.count++; + let restored: BridgeRestoredSession; + try { + restored = await inFlight.promise; + } catch (err) { + // Roll back our reservation so a subsequent retry isn't + // permanently skewed if the in-flight restore failed. + inFlight.coalesceState.count--; + throw err; + } + const entry = byId.get(restored.sessionId); + if (!entry) { + // Restore owner's session got reaped before our await + // resumed (channel died mid-microtask, etc). Roll back the + // reservation too — there's no entry for it to live on. + inFlight.coalesceState.count--; + throw new SessionNotFoundError( + restored.sessionId, + 'the agent child likely crashed during session restore — retry to restore the session', + ); + } + // NOTE: do NOT bump entry.attachCount here — `createSessionEntry` + // already initialized it from coalesceState.count synchronously + // when the IIFE registered the entry. Spread `restored` so the + // ACP state propagates to coalesced waiters (BQ9tV-equivalent + // for restore waiter consistency). + return { ...restored, attached: true }; + } + + if ( + byId.size + inFlightSpawns.size + inFlightRestores.size >= + maxSessions + ) { + throw new SessionLimitExceededError(maxSessions); + } + + const restoreEvents = new EventBus(); + let registeredEntry: SessionEntry | undefined; + let ci: ChannelInfo | undefined; + // Live counter shared with coalesced waiters (see InFlightRestore + // doc comment). Mutated synchronously by the coalesce branch above + // and read once by the IIFE when seeding `entry.attachCount`. + const coalesceState = { count: 0 }; + const promise = (async (): Promise => { + pendingRestoreEvents.set(req.sessionId, restoreEvents); + ci = await ensureChannel(); + ci.pendingRestoreIds.add(req.sessionId); + // Restore is a low-frequency one-shot path, so we register a + // fresh `channel.exited` listener per call instead of going + // through `getTransportClosedReject` (which exists to keep + // sendPrompt's per-session listener count at 1 over the + // session's lifetime). The listener is bound to this restore's + // race only — once the race settles, no new awaits attach to + // it, so there's no listener leak across restores. + const transportClosed = ci.channel.exited.then(() => { + throw new Error(`agent channel closed during session/${action}`); + }); + // Suppress the dangling rejection if `withTimeout` wins the + // race below: `transportClosed` then stays pending, and a + // later `channel.exited` settle fires the inner `throw` with + // no observer attached. Node 22 logs `unhandledRejection`; + // under `--unhandled-rejections=throw` (common in container + // deployments) the daemon process crashes. The `Promise.race` + // path's own consumer below catches the rejection in the + // try/catch, so the suppressed rejection here is the + // race-loser case only. + transportClosed.catch(() => {}); + let state: BridgeSessionState; + try { + if (action === 'load') { + state = await Promise.race([ + withTimeout( + ci.connection.loadSession({ + sessionId: req.sessionId, + cwd: workspaceKey, + // Restore path drops per-request `mcpServers` (matches + // `doSpawn`); daemon-wide MCP comes from settings on + // the agent side. The SDK's `RestoreSessionRequest` + // intentionally has no `mcpServers` field for the + // same reason. + mcpServers: [], + }), + initTimeoutMs, + 'loadSession', + ), + transportClosed, + ]); + } else { + state = await Promise.race([ + withTimeout( + ci.connection.unstable_resumeSession({ + sessionId: req.sessionId, + cwd: workspaceKey, + mcpServers: [], + }), + initTimeoutMs, + 'resumeSession', + ), + transportClosed, + ]); + } + } catch (err) { + restoreEvents.close(); + if (isAcpSessionResourceNotFound(err, req.sessionId)) { + throw new SessionNotFoundError(req.sessionId); + } + if ( + ci.sessionIds.size === 0 && + ci.pendingRestoreIds.size === 1 && + ci.pendingRestoreIds.has(req.sessionId) + ) { + ci.isDying = true; + await ci.channel.kill().catch(() => { + /* best-effort — channel.exited handler still runs */ + }); + } + throw err; + } + + if (shuttingDown) { + restoreEvents.close(); + throw new Error('HttpAcpBridge is shutting down'); + } + if (ci.isDying || !aliveChannels.has(ci)) { + restoreEvents.close(); + throw new Error( + `Session ${req.sessionId} restored on a closed agent channel`, + ); + } + const racedEntry = byId.get(req.sessionId); + if (racedEntry) { + restoreEvents.close(); + // Self + any coalescers we accumulated while the restore was + // in flight. Coalescers must not bump attachCount themselves + // (they read it off the registered entry on the next tick). + racedEntry.attachCount += 1 + coalesceState.count; + return { + sessionId: racedEntry.sessionId, + workspaceCwd: racedEntry.workspaceCwd, + attached: true, + state: racedEntry.restoreState ?? {}, + }; + } + + const entry = createSessionEntry( + ci, + req.sessionId, + workspaceKey, + restoreEvents, + ); + entry.restoreState = state; + // Fold synchronous coalesce reservations into the new entry's + // `attachCount`. By this point all coalescers that beat us must + // have hit the inFlightRestores branch and bumped + // `coalesceState.count`; later coalescers will hit the byId + // early-return path instead and increment `entry.attachCount` + // directly. + entry.attachCount = coalesceState.count; + registeredEntry = entry; + // Explicit `session/load` / `session/resume` is "give me THIS + // id"; it must NOT become the implicit attach target for + // subsequent omitted-id `POST /session` callers under `single` + // scope. Those callers asked for "any default", and silently + // joining a restored live history would surprise them. + // `defaultEntry` is reserved for sessions created through + // `doSpawn` under `'single'` scope. + return { + sessionId: entry.sessionId, + workspaceCwd: entry.workspaceCwd, + attached: false, + state, + }; + })().finally(() => { + ci?.pendingRestoreIds.delete(req.sessionId); + pendingRestoreEvents.delete(req.sessionId); + if (!registeredEntry) { + restoreEvents.close(); + } + }); + + inFlightRestores.set(req.sessionId, { action, promise, coalesceState }); + try { + return await promise; + } finally { + inFlightRestores.delete(req.sessionId); + } + } + return { get sessionCount() { return byId.size; @@ -1692,6 +2084,14 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { return pendingPermissions.size; }, + async loadSession(req) { + return restoreSession('load', req); + }, + + async resumeSession(req) { + return restoreSession('resume', req); + }, + async spawnOrAttach(req) { if (shuttingDown) { // `runQwenServe.close()` calls `bridge.shutdown()` BEFORE @@ -1701,11 +2101,6 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // see — they'd otherwise leak past `process.exit(0)`. throw new Error('HttpAcpBridge is shutting down'); } - if (!path.isAbsolute(req.workspaceCwd)) { - throw new Error( - `workspaceCwd must be an absolute path; got "${req.workspaceCwd}"`, - ); - } // Fast-path the common §02 case: clients pre-flight `caps.workspaceCwd` // and post back the exact same string, so the equality check // saves a `realpathSync.native` syscall per spawnOrAttach. The @@ -1715,16 +2110,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // sent a non-canonical alias (`/work/./bound`, mixed casing on // case-insensitive FS, a symlinked aliased path, …) — that // still needs the realpath to compare correctly. - const workspaceKey = - req.workspaceCwd === boundWorkspace - ? boundWorkspace - : canonicalizeWorkspace(req.workspaceCwd); - // #3803 §02: reject cross-workspace requests at the daemon - // boundary. The route layer catches `WorkspaceMismatchError` - // and translates to 400 with `code: 'workspace_mismatch'`. - if (workspaceKey !== boundWorkspace) { - throw new WorkspaceMismatchError(boundWorkspace, workspaceKey); - } + const workspaceKey = resolveWorkspaceKey(req.workspaceCwd); // Resolve the effective scope for THIS call. A per-request // `req.sessionScope` overrides the daemon-wide default; omitting @@ -1843,7 +2229,10 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // (a fresh-spawn races that's about to register hasn't hit // `byId` yet but should still count toward the limit). Attaches // returned above bypass this — only NEW children are gated. - if (byId.size + inFlightSpawns.size >= maxSessions) { + if ( + byId.size + inFlightSpawns.size + inFlightRestores.size >= + maxSessions + ) { throw new SessionLimitExceededError(maxSessions); } @@ -2196,17 +2585,23 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { /* bus already closed */ } entry.events.close(); - // Only kill the channel when no other sessions remain. ACP - // doesn't expose a per-session "close" call on the agent side, - // so the agent's `sessions: Map` grows by one - // until the channel dies — bounded by `maxSessions` (default - // 20) so memory is capped. FIXME(stage-1.5): if ACP grows a - // `closeSession` notification, send it here so the agent can - // drop the entry from its map immediately rather than at - // channel exit. (`channelInfo` itself is cleared by the - // `channel.exited` handler once the OS reaps the child — + // Only kill the channel when no other sessions remain AND no + // restore is in flight. ACP doesn't expose a per-session "close" + // call on the agent side, so the agent's `sessions: Map` grows by one until the channel dies — bounded by + // `maxSessions` (default 20) so memory is capped. FIXME(stage- + // 1.5): if ACP grows a `closeSession` notification, send it + // here so the agent can drop the entry from its map immediately + // rather than at channel exit. (`channelInfo` itself is cleared + // by the `channel.exited` handler once the OS reaps the child — // tanzhenxin BkUyD invariant.) - if (ci && ci.sessionIds.size === 0) { + // + // `pendingRestoreIds` covers in-flight `session/load` and + // `session/resume` calls that haven't yet registered into + // `sessionIds`. Killing the channel out from under them would + // SIGTERM the restore mid-flight and 500 the caller for a + // failure orthogonal to their request. + if (ci && ci.sessionIds.size === 0 && ci.pendingRestoreIds.size === 0) { // Mark dying SYNCHRONOUSLY before the await so a concurrent // `spawnOrAttach` arriving during the SIGTERM grace window // doesn't attach to a transport we're tearing down — without @@ -2349,6 +2744,13 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { () => undefined, ), ); + const inFlightRestoreAwaits = Array.from(inFlightRestores.values()).map( + (restore): Promise => + restore.promise.then( + () => undefined, + () => undefined, + ), + ); const inFlightChannelAwait: Promise = inFlightChannelSpawn ? inFlightChannelSpawn.then( () => undefined, @@ -2358,6 +2760,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { await Promise.all([ ...channels.map((ci) => ci.channel.kill().catch(() => {})), ...inFlightSessionAwaits, + ...inFlightRestoreAwaits, inFlightChannelAwait, ]); }, diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 213708c547e..9fd2825ee32 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -29,9 +29,13 @@ import type { } from '@agentclientprotocol/sdk'; import { InvalidPermissionOptionError, + MAX_WORKSPACE_PATH_LENGTH, + RestoreInProgressError, SessionLimitExceededError, SessionNotFoundError, WorkspaceMismatchError, + type BridgeRestoredSession, + type BridgeRestoreSessionRequest, type BridgeSession, type BridgeSessionSummary, type BridgeSpawnRequest, @@ -60,6 +64,8 @@ const EXPECTED_STAGE1_FEATURES = [ 'capabilities', 'session_create', 'session_scope_override', + 'session_load', + 'unstable_session_resume', 'session_list', 'session_prompt', 'session_cancel', @@ -70,6 +76,12 @@ const EXPECTED_STAGE1_FEATURES = [ interface FakeBridgeOpts { spawnImpl?: (req: BridgeSpawnRequest) => Promise; + loadImpl?: ( + req: BridgeRestoreSessionRequest, + ) => Promise; + resumeImpl?: ( + req: BridgeRestoreSessionRequest, + ) => Promise; promptImpl?: ( sessionId: string, req: PromptRequest, @@ -93,6 +105,8 @@ interface FakeBridgeOpts { interface FakeBridge extends HttpAcpBridge { calls: BridgeSpawnRequest[]; + loadCalls: BridgeRestoreSessionRequest[]; + resumeCalls: BridgeRestoreSessionRequest[]; promptCalls: Array<{ sessionId: string; req: PromptRequest; @@ -115,6 +129,8 @@ interface FakeBridge extends HttpAcpBridge { function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { const calls: BridgeSpawnRequest[] = []; + const loadCalls: BridgeRestoreSessionRequest[] = []; + const resumeCalls: BridgeRestoreSessionRequest[] = []; const promptCalls: FakeBridge['promptCalls'] = []; const cancelCalls: FakeBridge['cancelCalls'] = []; const killCalls: Array<{ @@ -133,6 +149,22 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { workspaceCwd: req.workspaceCwd, attached: false, })); + const loadImpl = + opts.loadImpl ?? + (async (req) => ({ + sessionId: req.sessionId, + workspaceCwd: req.workspaceCwd, + attached: false, + state: {}, + })); + const resumeImpl = + opts.resumeImpl ?? + (async (req) => ({ + sessionId: req.sessionId, + workspaceCwd: req.workspaceCwd, + attached: false, + state: {}, + })); const promptImpl = opts.promptImpl ?? (async () => ({ stopReason: 'end_turn' })); const cancelImpl = opts.cancelImpl ?? (async () => {}); @@ -141,6 +173,8 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { const setModelImpl = opts.setModelImpl ?? (async () => ({})); return { calls, + loadCalls, + resumeCalls, promptCalls, cancelCalls, killCalls, @@ -162,6 +196,16 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { calls.push(req); return result; }, + async loadSession(req) { + const result = await loadImpl(req); + loadCalls.push(req); + return result; + }, + async resumeSession(req) { + const result = await resumeImpl(req); + resumeCalls.push(req); + return result; + }, async sendPrompt(sessionId, req, signal) { promptCalls.push({ sessionId, req, signal }); return promptImpl(sessionId, req, signal); @@ -628,6 +672,209 @@ describe('createServeApp', () => { }); }); + describe('POST /session/:id/load and /resume', () => { + it('falls back to bound workspace and uses the route session id', async () => { + for (const action of ['load', 'resume'] as const) { + const bridge = fakeBridge(); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + const res = await request(app) + .post(`/session/persisted-1/${action}`) + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ sessionId: 'spoofed-body-id' }); + + expect(res.status).toBe(200); + expect(res.body).toEqual({ + sessionId: 'persisted-1', + workspaceCwd: WS_BOUND, + attached: false, + state: {}, + }); + const calls = action === 'load' ? bridge.loadCalls : bridge.resumeCalls; + expect(calls).toEqual([ + { sessionId: 'persisted-1', workspaceCwd: WS_BOUND }, + ]); + } + }); + + it('passes explicit cwd through to the bridge', async () => { + const bridge = fakeBridge({ + loadImpl: async (req) => ({ + sessionId: req.sessionId, + workspaceCwd: req.workspaceCwd, + attached: false, + state: { configOptions: [] }, + }), + }); + const app = createServeApp(baseOpts, undefined, { bridge }); + const res = await request(app) + .post('/session/persisted-2/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: '/work/a' }); + + expect(res.status).toBe(200); + expect(res.body.state).toEqual({ configOptions: [] }); + expect(bridge.loadCalls).toEqual([ + { sessionId: 'persisted-2', workspaceCwd: '/work/a' }, + ]); + }); + + it('400s malformed cwd before touching the bridge', async () => { + for (const action of ['load', 'resume'] as const) { + const bridge = fakeBridge(); + const app = createServeApp(baseOpts, undefined, { bridge }); + const res = await request(app) + .post(`/session/persisted-3/${action}`) + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: 'relative/path' }); + + expect(res.status).toBe(400); + expect(bridge.loadCalls).toHaveLength(0); + expect(bridge.resumeCalls).toHaveLength(0); + } + }); + + it('400s a non-string cwd before touching the bridge', async () => { + // Mirrors the `POST /session` malformed-`cwd`-shape test: a + // client/orchestrator serialization bug (`cwd: null`, + // `cwd: 123`, `cwd: {}`) must surface as a typed 400 instead of + // silently falling back to the bound workspace. + for (const action of ['load', 'resume'] as const) { + for (const cwd of [null, 123, {}, []]) { + const bridge = fakeBridge(); + const app = createServeApp(baseOpts, undefined, { bridge }); + const res = await request(app) + .post(`/session/persisted-mal/${action}`) + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd }); + + expect(res.status).toBe(400); + expect(bridge.loadCalls).toHaveLength(0); + expect(bridge.resumeCalls).toHaveLength(0); + } + } + }); + + it('400s a cwd longer than MAX_WORKSPACE_PATH_LENGTH before touching the bridge', async () => { + // Same length cap as `POST /session` (matches Linux PATH_MAX + // 4096) — defends downstream interpolations from + // amplification on the loopback-default-no-token path. + const longCwd = `/${'a'.repeat(MAX_WORKSPACE_PATH_LENGTH)}`; + for (const action of ['load', 'resume'] as const) { + const bridge = fakeBridge(); + const app = createServeApp(baseOpts, undefined, { bridge }); + const res = await request(app) + .post(`/session/persisted-long/${action}`) + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: longCwd }); + + expect(res.status).toBe(400); + expect(res.body.error).toMatch( + new RegExp( + `exceeds the ${MAX_WORKSPACE_PATH_LENGTH}-character limit`, + ), + ); + expect(bridge.loadCalls).toHaveLength(0); + expect(bridge.resumeCalls).toHaveLength(0); + } + }); + + it('404s when the bridge reports an unknown persisted session', async () => { + const bridge = fakeBridge({ + resumeImpl: async (req) => { + throw new SessionNotFoundError(req.sessionId); + }, + }); + const app = createServeApp(baseOpts, undefined, { bridge }); + const res = await request(app) + .post('/session/missing/resume') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({}); + + expect(res.status).toBe(404); + expect(res.body.sessionId).toBe('missing'); + }); + + it('409 + Retry-After when the bridge throws RestoreInProgressError', async () => { + const bridge = fakeBridge({ + loadImpl: async () => { + throw new RestoreInProgressError('persisted-race', 'resume', 'load'); + }, + }); + const app = createServeApp(baseOpts, undefined, { bridge }); + const res = await request(app) + .post('/session/persisted-race/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({}); + + expect(res.status).toBe(409); + expect(res.headers['retry-after']).toBe('5'); + expect(res.body).toMatchObject({ + code: 'restore_in_progress', + sessionId: 'persisted-race', + activeAction: 'resume', + requestedAction: 'load', + }); + }); + + it('400 workspace_mismatch when the bridge throws WorkspaceMismatchError', async () => { + const bridge = fakeBridge({ + loadImpl: async () => { + throw new WorkspaceMismatchError(WS_BOUND, WS_DIFFERENT); + }, + }); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + const res = await request(app) + .post('/session/persisted-x/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: WS_DIFFERENT }); + + expect(res.status).toBe(400); + expect(res.body).toMatchObject({ + code: 'workspace_mismatch', + boundWorkspace: WS_BOUND, + requestedWorkspace: WS_DIFFERENT, + }); + }); + + it('503 + Retry-After: 5 when the bridge throws SessionLimitExceededError', async () => { + const bridge = fakeBridge({ + resumeImpl: async () => { + throw new SessionLimitExceededError(20); + }, + }); + const app = createServeApp(baseOpts, undefined, { bridge }); + const res = await request(app) + .post('/session/persisted-y/resume') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({}); + + expect(res.status).toBe(503); + expect(res.headers['retry-after']).toBe('5'); + expect(res.body).toMatchObject({ + code: 'session_limit_exceeded', + limit: 20, + }); + }); + + // The restore handler's `!res.writable` cleanup branch (kill on + // !attached, detach on attached) is line-for-line identical to + // the matching branch on `POST /session`; routing-side + // disconnect tests for that handler weren't added when the + // cleanup was originally introduced because the supertest + + // Node http close-event timing makes the assertion flaky in + // CI. The same constraint applies here. The cleanup behavior + // is exercised manually via the route handler closure shared + // between both routes in `restoreSessionHandler`. + }); + describe('POST /session/:id/prompt', () => { it('200 with PromptResponse on success; route :id wins over body sessionId', async () => { const bridge = fakeBridge({ diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 124b8ebc901..2c919942506 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -16,6 +16,7 @@ import { InvalidPermissionOptionError, InvalidSessionScopeError, MAX_WORKSPACE_PATH_LENGTH, + RestoreInProgressError, SessionLimitExceededError, SessionNotFoundError, WorkspaceMismatchError, @@ -63,6 +64,8 @@ export interface ServeAppDeps { * - `GET /health` * - `GET /capabilities` * - `POST /session` + * - `POST /session/:id/load` + * - `POST /session/:id/resume` * - `GET /workspace/:id/sessions` * - `POST /session/:id/prompt` * - `POST /session/:id/cancel` @@ -353,6 +356,58 @@ export function createServeApp( } }); + const restoreSessionHandler = + (action: 'load' | 'resume') => + async (req: express.Request, res: express.Response) => { + const sessionId = req.params['id']; + if (!sessionId) { + res + .status(400) + .json({ error: '`sessionId` route parameter is required' }); + return; + } + const body = safeBody(req); + const cwd = parseOptionalWorkspaceCwd(body, boundWorkspace, res); + if (cwd === undefined) return; + try { + const session = + action === 'load' + ? await bridge.loadSession({ sessionId, workspaceCwd: cwd }) + : await bridge.resumeSession({ sessionId, workspaceCwd: cwd }); + // Mirror the `POST /session` disconnect-cleanup path (see the + // long comment above the matching `if (!res.writable)` there + // for the rationale around `res.writable` vs `req.aborted` / + // `req.destroyed`, plus the BQ9tV `requireZeroAttaches` race + // and the tanzhenxin attach-rollback case). Restore needs the + // same cleanup because a client that disconnects during a + // multi-second `session/load` would otherwise leave a freshly + // restored session in `byId` with no client holding its id. + if (!res.writable) { + if (!session.attached) { + bridge + .killSession(session.sessionId, { requireZeroAttaches: true }) + .catch(() => { + // Best-effort cleanup; channel.exited will eventually reap. + }); + } else { + bridge.detachClient(session.sessionId).catch(() => { + // Best-effort cleanup; channel.exited will eventually reap. + }); + } + return; + } + res.status(200).json(session); + } catch (err) { + sendBridgeError(res, err, { + route: `POST /session/:id/${action}`, + sessionId, + }); + } + }; + + app.post('/session/:id/load', restoreSessionHandler('load')); + app.post('/session/:id/resume', restoreSessionHandler('resume')); + app.post('/session/:id/prompt', async (req, res) => { const sessionId = req.params['id']; const body = safeBody(req); @@ -873,6 +928,34 @@ function safeBody(req: import('express').Request): Record { return out; } +function parseOptionalWorkspaceCwd( + body: Record, + boundWorkspace: string, + res: import('express').Response, +): string | undefined { + const hasCwd = 'cwd' in body; + if (hasCwd && typeof body['cwd'] !== 'string') { + res + .status(400) + .json({ error: '`cwd` must be a string absolute path when provided' }); + return undefined; + } + if (hasCwd && (body['cwd'] as string).length > MAX_WORKSPACE_PATH_LENGTH) { + res.status(400).json({ + error: `\`cwd\` exceeds the ${MAX_WORKSPACE_PATH_LENGTH}-character limit`, + }); + return undefined; + } + const cwd = hasCwd ? (body['cwd'] as string) : boundWorkspace; + if (!path.isAbsolute(cwd)) { + res + .status(400) + .json({ error: '`cwd` must be an absolute path when provided' }); + return undefined; + } + return cwd; +} + function isValidOutcome( raw: unknown, ): raw is { outcome: 'cancelled' } | { outcome: 'selected'; optionId: string } { @@ -1033,6 +1116,21 @@ function sendBridgeError( }); return; } + if (err instanceof RestoreInProgressError) { + // Match `SessionLimitExceededError`'s 5s hint (above) — the + // underlying restore can take up to `initTimeoutMs` (default + // 10s) on the agent side, so a 1s retry hint pushed clients + // into tight loops that kept hitting the same 409. + res.set('Retry-After', '5'); + res.status(409).json({ + error: err.message, + code: 'restore_in_progress', + sessionId: err.sessionId, + activeAction: err.activeAction, + requestedAction: err.requestedAction, + }); + return; + } // 5xx is the kind of error operators need to see in their daemon log // — bridge ENOMEM, agent stack trace, unexpected throw, etc. Without // logging here every 500 disappears once the caller consumes the diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index 929656129ef..35a85099e0d 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -8,6 +8,7 @@ import { parseSseStream } from './sse.js'; import type { DaemonCapabilities, DaemonEvent, + DaemonRestoredSession, DaemonSession, DaemonSessionSummary, PermissionResponse, @@ -117,6 +118,14 @@ export interface CreateSessionRequest { sessionScope?: 'single' | 'thread'; } +export interface RestoreSessionRequest { + /** + * Workspace path the daemon must be bound to. Omit to let the daemon use + * its advertised bound workspace, mirroring `createOrAttachSession`. + */ + workspaceCwd?: string; +} + export interface PromptRequest { prompt: PromptContentBlock[]; /** Optional ACP _meta passthrough. */ @@ -335,6 +344,48 @@ export class DaemonClient { ); } + async loadSession( + sessionId: string, + req: RestoreSessionRequest = {}, + ): Promise { + return this.restoreSession('load', sessionId, req); + } + + async resumeSession( + sessionId: string, + req: RestoreSessionRequest = {}, + ): Promise { + return this.restoreSession('resume', sessionId, req); + } + + /** + * Shared transport for `loadSession` / `resumeSession`. Both routes + * share an identical wire shape (POST /session/:id/{load|resume} + * with optional `cwd` body) and identical error envelopes from the + * daemon, so they collapse into a single fetch path that only + * differs in the URL suffix and the route name reported on errors. + */ + private async restoreSession( + action: 'load' | 'resume', + sessionId: string, + req: RestoreSessionRequest, + ): Promise { + return await this.fetchWithTimeout( + `${this.baseUrl}/session/${encodeURIComponent(sessionId)}/${action}`, + { + method: 'POST', + headers: this.headers({ 'Content-Type': 'application/json' }), + body: JSON.stringify({ cwd: req.workspaceCwd }), + }, + async (res) => { + if (!res.ok) { + throw await this.failOnError(res, `POST /session/:id/${action}`); + } + return (await res.json()) as DaemonRestoredSession; + }, + ); + } + /** * Switch the active model for a session. Backed by ACP's currently-unstable * `unstable_setSessionModel`; the daemon also publishes a `model_switched` diff --git a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts index cf993a8c67a..fbb74348bb4 100644 --- a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts @@ -8,10 +8,12 @@ import type { DaemonClient } from './DaemonClient.js'; import { type CreateSessionRequest, type PromptRequest, + type RestoreSessionRequest, type SubscribeOptions, } from './DaemonClient.js'; import type { DaemonEvent, + DaemonSessionState, DaemonSession, PermissionResponse, PromptResult, @@ -21,6 +23,8 @@ import type { export interface DaemonSessionClientOptions { client: DaemonClient; session: DaemonSession; + /** ACP state returned by load/resume; empty for create/attach clients. */ + state?: DaemonSessionState; /** * Seed replay state for callers that persisted the last seen SSE event id. * When omitted, the first event subscription starts live. @@ -50,12 +54,14 @@ export interface DaemonSessionSubscribeOptions extends SubscribeOptions { export class DaemonSessionClient { readonly client: DaemonClient; readonly session: DaemonSession; + readonly state: DaemonSessionState; private lastSeenEventId: number | undefined; private subscriptionActive = false; constructor(opts: DaemonSessionClientOptions) { this.client = opts.client; this.session = { ...opts.session }; + this.state = { ...(opts.state ?? {}) }; this.lastSeenEventId = opts.lastEventId; } @@ -75,6 +81,48 @@ export class DaemonSessionClient { return new DaemonSessionClient({ client, session, lastEventId }); } + /** + * Loads an existing daemon session and seeds the first event subscription + * from the start of the daemon replay ring so history replay frames emitted + * during `session/load` are visible to this client. + */ + static async load( + client: DaemonClient, + sessionId: string, + req: RestoreSessionRequest = {}, + ): Promise { + const { state, ...session } = await client.loadSession(sessionId, req); + return new DaemonSessionClient({ + client, + session, + state, + lastEventId: 0, + }); + } + + /** + * Resumes an existing daemon session without requesting history replay. + * Seeds the first event subscription from the start of the daemon + * replay ring (`lastEventId: 0`) symmetric with `load()` — the agent's + * `unstable_resumeSession` schedules an `available_commands_update` + * via `setTimeout(0)`, which can publish to the daemon bus between + * the HTTP response and the consumer's first `events()` call. Seeding + * ensures that frame is observed instead of dropped. + */ + static async resume( + client: DaemonClient, + sessionId: string, + req: RestoreSessionRequest = {}, + ): Promise { + const { state, ...session } = await client.resumeSession(sessionId, req); + return new DaemonSessionClient({ + client, + session, + state, + lastEventId: 0, + }); + } + get sessionId(): string { return this.session.sessionId; } diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index 8302fcdcf99..e11e3f5563d 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -10,6 +10,7 @@ export { type CreateSessionRequest, type DaemonClientOptions, type PromptRequest, + type RestoreSessionRequest, type SubscribeOptions, } from './DaemonClient.js'; export { @@ -24,7 +25,9 @@ export type { DaemonEvent, DaemonMode, DaemonProtocolVersions, + DaemonRestoredSession, DaemonSession, + DaemonSessionState, DaemonSessionSummary, PermissionOutcome, PermissionOutcomeCancelled, diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 9c880ff55a6..c0703b3460f 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -117,6 +117,36 @@ export interface DaemonSession { attached: boolean; } +/** + * ACP state returned by session load/resume routes. + * + * Fields mirror the ACP `LoadSessionResponse` / `ResumeSessionResponse` + * shapes (see `@agentclientprotocol/sdk`): + * - `models`: the agent's `SessionModelState` — current model id + + * available models the session can switch to. + * - `modes`: the agent's `SessionModeState` — current mode id + + * available approval / interaction modes. + * - `configOptions`: array of `SessionConfigOption` describing + * per-session toggles the client can flip via + * `POST /session/:id/config-option`. + * + * They are typed as `unknown` here to avoid coupling the SDK to ACP's + * internal protocol types, which the SDK doesn't re-export. Callers + * that need richer typing should narrow to the ACP shapes themselves. + */ +export interface DaemonSessionState { + _meta?: Record | null; + models?: unknown; + modes?: unknown; + configOptions?: unknown[] | null; + [key: string]: unknown; +} + +/** Returned from `POST /session/:id/load` and `POST /session/:id/resume`. */ +export interface DaemonRestoredSession extends DaemonSession { + state: DaemonSessionState; +} + /** Sparse session record returned by `GET /workspace/:id/sessions`. */ export interface DaemonSessionSummary { sessionId: string; diff --git a/packages/sdk-typescript/src/index.ts b/packages/sdk-typescript/src/index.ts index 69e27c98b56..78d6e92c649 100644 --- a/packages/sdk-typescript/src/index.ts +++ b/packages/sdk-typescript/src/index.ts @@ -18,9 +18,11 @@ export { type DaemonEvent, type DaemonMode, type DaemonProtocolVersions, + type DaemonRestoredSession, type DaemonSession, type DaemonSessionClientOptions, type DaemonSessionSubscribeOptions, + type DaemonSessionState, type DaemonSessionSummary, type PermissionOutcome, type PermissionOutcomeCancelled, @@ -39,6 +41,7 @@ export { type PromptRequest, type PromptResult, type PromptTextContent, + type RestoreSessionRequest, type SetModelResult, type SubscribeOptions, } from './daemon/index.js'; diff --git a/packages/sdk-typescript/test/unit/DaemonClient.test.ts b/packages/sdk-typescript/test/unit/DaemonClient.test.ts index f5dd9573680..93aaaaf91fe 100644 --- a/packages/sdk-typescript/test/unit/DaemonClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonClient.test.ts @@ -334,6 +334,54 @@ describe('DaemonClient', () => { }); }); + describe('loadSession / resumeSession', () => { + it('POSTs /session/:id/load with optional cwd', async () => { + const { fetch, calls } = recordingFetch(() => + jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: false, + state: { configOptions: [] }, + }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = await client.loadSession('s-1', { + workspaceCwd: '/work/a', + }); + + expect(session.state).toEqual({ configOptions: [] }); + expect(calls[0]?.url).toBe('http://daemon/session/s-1/load'); + expect(calls[0]?.method).toBe('POST'); + expect(JSON.parse(calls[0]!.body!)).toEqual({ cwd: '/work/a' }); + }); + + it('POSTs /session/:id/resume and omits cwd when absent', async () => { + const { fetch, calls } = recordingFetch(() => + jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/bound', + attached: false, + state: {}, + }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + await client.resumeSession('with/slash'); + + expect(calls[0]?.url).toBe('http://daemon/session/with%2Fslash/resume'); + expect(JSON.parse(calls[0]!.body!)).toEqual({}); + }); + + it('throws DaemonHttpError on restore failures', async () => { + const { fetch } = recordingFetch(() => + jsonResponse(404, { error: 'missing' }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + await expect(client.loadSession('missing')).rejects.toMatchObject({ + status: 404, + }); + }); + }); + describe('cancel', () => { it('POSTs /cancel and tolerates 204', async () => { const { fetch, calls } = recordingFetch( diff --git a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts index 9fb6ec73294..e908b183a1c 100644 --- a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts @@ -125,6 +125,67 @@ describe('DaemonSessionClient', () => { expect(calls[1]?.headers['last-event-id']).toBe('0'); }); + it('loads an existing daemon session and seeds replay from the start', async () => { + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/load')) { + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: false, + state: { configOptions: [] }, + }); + } + if (req.url.endsWith('/session/s-1/events')) { + return sseResponse(''); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + + const session = await DaemonSessionClient.load(client, 's-1', { + workspaceCwd: '/work/a', + }); + + expect(session.sessionId).toBe('s-1'); + expect(session.state).toEqual({ configOptions: [] }); + expect(JSON.parse(calls[0]!.body!)).toEqual({ cwd: '/work/a' }); + + for await (const _event of session.events()) { + /* empty */ + } + expect(calls[1]?.headers['last-event-id']).toBe('0'); + }); + + it('resumes an existing daemon session and seeds replay from the start', async () => { + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/resume')) { + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + state: { modes: null }, + }); + } + if (req.url.endsWith('/session/s-1/events')) { + return sseResponse(''); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + + const session = await DaemonSessionClient.resume(client, 's-1'); + + expect(session.attached).toBe(true); + expect(session.state).toEqual({ modes: null }); + for await (const _event of session.events()) { + /* empty */ + } + // Symmetric to load(): `unstable_resumeSession` schedules an + // `available_commands_update` via setTimeout(0) on the agent side, + // so the SDK seeds the subscription from the start of the ring. + expect(calls[1]?.headers['last-event-id']).toBe('0'); + }); + it('forwards session-scoped operations through DaemonClient', async () => { const { fetch, calls } = recordingFetch((req) => { if (req.url.endsWith('/session/s-1/prompt')) {