From 393943daaf9532206e18f6bd350b897b5a3f81f3 Mon Sep 17 00:00:00 2001 From: jinye Date: Thu, 9 Jul 2026 06:35:53 +0800 Subject: [PATCH] feat(cli): Add session owner index for workspace runtimes (#6540) * feat(cli): Add session owner index for workspace runtimes Route live session ownership through a registry-backed owner index so multi-workspace sessions can resolve active sessions without scanning every bridge first. Expand trusted workspace load/resume and live read routing while keeping non-session surfaces primary-only. Co-authored-by: Qwen-Coder * fix(cli): avoid partial session owner index updates Co-authored-by: Qwen-Coder * test(cli): relax bridge wiring test timeout Co-authored-by: Qwen-Coder * fix(cli): tighten workspace session owner routing Co-authored-by: Qwen-Coder * fix(cli): normalize restore workspace mismatch handling Co-authored-by: Qwen-Coder * fix(cli): record telemetry for workspace sessions alias Co-authored-by: Qwen-Coder * fix(cli): preserve workspace selector error contract Co-authored-by: Qwen-Coder --------- Co-authored-by: Qwen-Coder --- ...daemon-multi-workspace-phase2a-sessions.md | 84 ++- docs/developers/qwen-serve-protocol.md | 39 +- packages/acp-bridge/src/bridge.test.ts | 75 +++ packages/acp-bridge/src/bridge.ts | 67 +- packages/acp-bridge/src/bridgeOptions.ts | 25 + packages/cli/src/serve/acp-session-bridge.ts | 2 + .../serve/multi-workspace-sessions.test.ts | 98 ++- packages/cli/src/serve/rate-limit.test.ts | 23 + packages/cli/src/serve/routes/session.ts | 629 ++++++++++++------ packages/cli/src/serve/run-qwen-serve.ts | 9 +- .../server-default-bridge-wiring.test.ts | 153 +++++ packages/cli/src/serve/server.test.ts | 481 ++++++++++++-- packages/cli/src/serve/server.ts | 41 +- .../cli/src/serve/server/telemetry.test.ts | 22 + packages/cli/src/serve/server/telemetry.ts | 3 + .../cli/src/serve/workspace-registry.test.ts | 103 +++ packages/cli/src/serve/workspace-registry.ts | 129 +++- 17 files changed, 1643 insertions(+), 340 deletions(-) create mode 100644 packages/cli/src/serve/server-default-bridge-wiring.test.ts diff --git a/docs/design/daemon-multi-workspace-phase2a-sessions.md b/docs/design/daemon-multi-workspace-phase2a-sessions.md index 3297f07953..81c89399f3 100644 --- a/docs/design/daemon-multi-workspace-phase2a-sessions.md +++ b/docs/design/daemon-multi-workspace-phase2a-sessions.md @@ -2,17 +2,23 @@ ## Summary -This document records the Phase 2a contract for issue #6378 after the Phase 1 -`WorkspaceRegistry` PR and the Phase 2a foundation PR. Phase 2a is now split -into two implementation PRs: PR 1 landed env isolation and total-admission -guardrails while multi-workspace remained gated; PR 2 wires non-primary live -session dispatch and publishes the additive capabilities/status schema. +This document records the multi-workspace sessions contract for issue #6378 +after the Phase 1 `WorkspaceRegistry` PR, the Phase 2a foundation PR, and the +first Phase 2b route-expansion PR. Phase 2a was split into two implementation +PRs: PR 1 landed env isolation and total-admission guardrails while +multi-workspace remained gated; PR 2 wired non-primary live session dispatch +and published the additive capabilities/status schema. Phase 2b PR 1 adds a +session owner index and expands the sessions-only route surface without moving +file, memory, MCP, settings, voice, channel workers, ACP, or SDK workspace +clients. -Phase 2a remains sessions-only. It does not add plural routes, a -`WorkspaceDaemonClient`, workspace-qualified ACP/WebSocket, file, memory, MCP, -settings, voice, or channel-worker migration. PR 1 does not add capabilities -`workspaces[]`, `multi_workspace_sessions`, route dispatch, or non-primary -runtime construction. +The multi-workspace work remains sessions-only. Phase 2a did not add plural +routes, a `WorkspaceDaemonClient`, workspace-qualified ACP/WebSocket, file, +memory, MCP, settings, voice, or channel-worker migration. Phase 2b PR 1 adds +only the plural session-list alias described below; it still does not add +workspace client APIs or migrate non-session surfaces. PR 1 did not add +capabilities `workspaces[]`, `multi_workspace_sessions`, route dispatch, or +non-primary runtime construction. ## Foundation Contract @@ -74,10 +80,21 @@ Phase 2a-dispatched routes: - `DELETE /session/:id` - `GET /session/:id/status` +Phase 2b-dispatched additions: + +- `POST /session/:id/load` +- `POST /session/:id/resume` +- `GET /session/:id/context` +- `GET /session/:id/context-usage` +- `GET /session/:id/stats` +- `GET /session/:id/supported-commands` +- `GET /session/:id/tasks` +- `GET /session/:id/lsp` +- `GET /session/:id/hooks` +- `GET /session/:id/artifacts` + Later or primary-only routes: -- non-primary `POST /session/:id/load` -- non-primary `POST /session/:id/resume` - `GET /session/:id/export` - `POST /sessions/delete` - `POST /sessions/archive` @@ -88,9 +105,6 @@ Later or primary-only routes: - non-session `POST /permission/:requestId` - `/acp` -Additional live read routes may be owner-routed in a later Phase 2a slice only -after tests prove they depend solely on the owning live bridge. - ## Phase 2a Cross-PR Requirements - Keep scan misses as `404 session_not_found`; never fall back to primary. @@ -188,6 +202,46 @@ file/memory/MCP/settings/voice/channel-worker migration, dynamic add/remove, non-primary persisted load/resume/export/archive/delete, branch/fork/cd/rewind, shell/model/language migration, or SDK workspace client APIs. +## Phase 2b PR 1 Owner Index And Restore Expansion + +Phase 2b PR 1 adds a bridge lifecycle callback seam and a +`WorkspaceSessionOwnerIndex` owned by `WorkspaceRegistry`. Bridge +register/remove lifecycle events update the index on spawn, load/resume, +channel exit, close, kill, and daemon shutdown. Owner resolution consults the +index first, verifies the indexed runtime with `getSessionSummary`, drops stale +index entries, and falls back to the existing live bridge scan. Fallback hits +are cached back into the index. The index remains an optimization and +consistency seam, not a persisted ownership database. + +`POST /session/:id/load` and `POST /session/:id/resume` now accept explicit +`cwd` for any trusted registered workspace. Omitted `cwd` still resolves to the +primary runtime. Unknown `cwd` returns `400 workspace_mismatch`; untrusted +non-primary `cwd` returns `403 untrusted_workspace`; if the same session id is +already live or being restored in another runtime, restore fails closed with +`409 session_workspace_conflict`. Same-workspace restore races keep the +bridge's existing coalescing and `restore_in_progress` behavior. Restore still +reads persisted session storage from the requested workspace's existing storage +path and does not enable non-primary export/archive/delete. + +The owner-routed read-only live routes now use the owning runtime bridge: +context, context-usage, stats, supported-commands, tasks, lsp, hooks, and +artifacts. These routes do not mutate persisted storage and do not require +ACP/WebSocket connection-local state, so they can safely follow the live owner. +`GET /session/:id/rewind/snapshots` remains primary-only because rewind state is +not part of the sessions-only closed loop. + +`GET /workspaces/:workspace/sessions` is a plural alias for +`GET /workspace/:id/sessions`. Both resolve exact workspace id first and exact +canonical cwd second. Primary workspaces keep persisted/live merge semantics. +Non-primary workspaces stay live-only and continue rejecting archived or +organized list views. + +Phase 2b PR 1 does not add new capability tags, does not alter the +`/capabilities` schema, does not change SDK types, and does not route ACP, +voice, channel-worker, file, memory, MCP, settings, branch/fork/cd/rewind, +shell/model/language, export, archive, delete, or organization surfaces to +non-primary runtimes. + ## Audit Decisions - The foundation PR must not create non-primary runtimes or relax any REST diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 15cfa4d2a4..9c8077852c 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -114,6 +114,22 @@ Attaches to existing sessions are NOT counted toward the cap, so an idle daemon' 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. +`SessionWorkspaceConflictError` — emitted by `POST /session/:id/load` and `POST /session/:id/resume` when the requested `cwd` targets one registered workspace but the same session id is already live or being restored by another runtime — returns `409` with: + +```json +{ + "error": "Session \"\" is already live or restoring in another workspace runtime.", + "code": "session_workspace_conflict", + "sessionId": "", + "workspaceCwd": "/requested/workspace", + "workspaceId": "requested-workspace-id", + "liveWorkspaceCwd": "/live/owner/workspace", + "liveWorkspaceId": "live-owner-workspace-id" +} +``` + +Clients should retry with the owning workspace or wait for the in-flight restore to finish before restoring the id into a different workspace. Same-workspace restore races continue to use the bridge's `restore_in_progress` / coalescing behavior. + `SessionArchivedError` is emitted when a caller tries to load or resume a session whose JSONL is under `chats/archive/`: ```json @@ -415,7 +431,13 @@ the daemon log path; `full` may include it for authenticated operators. "supported": ["v1"] }, "mode": "http-bridge", - "features": ["health", "daemon_status", "capabilities", "multi_workspace_sessions", "..."], + "features": [ + "health", + "daemon_status", + "capabilities", + "multi_workspace_sessions", + "..." + ], "limits": { "maxPendingPromptsPerSession": 5, "maxSessionsPerWorkspace": 20, @@ -457,7 +479,7 @@ do not mutate state, and do not change the serve protocol version. Workspace status routes intentionally do **not** start the ACP child process just because a client polls a GET route: if the daemon is idle, they return `initialized: false` with an empty snapshot. Session status routes require a -live session and use the standard `404 SessionNotFoundError` shape for unknown +live session and return `404 { code: "session_not_found", ... }` for unknown ids. Capability tags: @@ -1258,9 +1280,9 @@ Request: } ``` -| 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`). | +| Field | Required | Notes | +| ----- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `cwd` | no | Same canonicalization + `workspace_mismatch` rules as `POST /session`. Omit to inherit `/capabilities.workspaceCwd`. When `features` contains `multi_workspace_sessions`, callers may pass any trusted registered `workspaces[].cwd`; untrusted non-primary workspaces return `403 untrusted_workspace`. `mcpServers` is intentionally NOT accepted here — daemon-wide MCP is settings-driven (matches `POST /session`). | Response: @@ -1287,8 +1309,10 @@ Response: - `404` — persisted session id doesn't exist (`SessionNotFoundError`). - `400` — `workspace_mismatch` (same shape as `POST /session`). +- `403` — `untrusted_workspace` when `cwd` targets an untrusted non-primary workspace. - `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`. +- `409` — `session_workspace_conflict` when the same session id is already live or being restored by another workspace runtime. - `409` — `session_archived` when the id exists only under `chats/archive/`; call `POST /sessions/unarchive` before `load` or `resume`. - `409` — `session_archiving` when archive or unarchive is in flight for the same id. `Retry-After: 5`. - `409` — `session_conflict` when the id exists in both `chats/` and `chats/archive/`; delete the session with `POST /sessions/delete` before loading. @@ -1303,13 +1327,14 @@ Use `/load` when the client has no history rendered (cold reconnect, picker → > ⚠️ **Why is `unstable_session_resume` still advertised?** The daemon's HTTP route and `session_resume` capability are stable for v1, but the bridge still calls ACP's `connection.unstable_resumeSession`. The old tag remains only so SDKs that shipped before `session_resume` can keep working. -### `GET /workspace/:id/sessions` +### `GET /workspace/:id/sessions` and `GET /workspaces/:workspace/sessions` -List persisted sessions whose canonical workspace matches `:id` (URL-encoded absolute cwd). The default list is active sessions from `chats/`; pass `archiveState=archived` to list archived sessions from `chats/archive/`. `archiveState=all` is not supported in v1. The default response and numeric `cursor` semantics are unchanged by `session_organization`. +List sessions whose canonical workspace matches `:id` or `:workspace`. The path parameter first resolves as an exact workspace id and then as a URL-encoded absolute cwd. `GET /workspaces/:workspace/sessions` is a plural alias with the same response shape. Primary workspaces include the existing persisted/live merge: the default list is active sessions from `chats/`; pass `archiveState=archived` to list archived sessions from `chats/archive/`. Non-primary workspaces are live-only in the current multi-workspace sessions surface and reject archived, organized, or grouped queries. Untrusted non-primary workspaces return `403 { code: "untrusted_workspace" }`. `archiveState=all` is not supported in v1. The default response and numeric `cursor` semantics are unchanged by `session_organization`. ```bash curl http://127.0.0.1:4170/workspace/$(jq -rn --arg c "$PWD" '$c|@uri')/sessions curl http://127.0.0.1:4170/workspace/$(jq -rn --arg c "$PWD" '$c|@uri')/sessions?archiveState=archived +curl http://127.0.0.1:4170/workspaces//sessions ``` Query parameters: diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index a2849340e0..97166e15c7 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -77,6 +77,81 @@ function deferred(): { } describe('createAcpSessionBridge', () => { + it('emits lifecycle events when a fresh session is registered and closed', async () => { + const events: Array<{ + type: string; + sessionId: string; + workspaceCwd: string; + reason?: string; + }> = []; + const bridge = makeBridge({ + channelFactory: async () => makeChannel().channel, + sessionLifecycle: (event) => { + events.push(event); + }, + }); + + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await bridge.closeSession(session.sessionId); + + expect(events).toEqual([ + { + type: 'registered', + sessionId: session.sessionId, + workspaceCwd: WS_A, + reason: 'spawn', + }, + { + type: 'removed', + sessionId: session.sessionId, + workspaceCwd: WS_A, + reason: 'client_close', + }, + ]); + + await bridge.shutdown(); + }); + + it('does not emit another registration for attach-only spawnOrAttach calls', async () => { + const events: Array<{ type: string; sessionId: string }> = []; + const bridge = makeBridge({ + channelFactory: async () => makeChannel().channel, + sessionLifecycle: (event) => { + events.push(event); + }, + }); + + const first = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const second = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + expect(second.sessionId).toBe(first.sessionId); + expect(events.filter((event) => event.type === 'registered')).toEqual([ + expect.objectContaining({ + type: 'registered', + sessionId: first.sessionId, + }), + ]); + + await bridge.closeSession(first.sessionId); + await bridge.shutdown(); + }); + + it('does not fail session lifecycle when the lifecycle callback throws', async () => { + const bridge = makeBridge({ + channelFactory: async () => makeChannel().channel, + sessionLifecycle: () => { + throw new Error('lifecycle sink failed'); + }, + }); + + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await expect(bridge.closeSession(session.sessionId)).resolves.toBe( + undefined, + ); + + await bridge.shutdown(); + }); + it('accepts a valid BridgeOptions.eventRingSize at construction time', () => { // Smoke: positive finite integers are accepted; the underlying // EventBus ring-size threading is exercised end-to-end in diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index e4d586de7a..38f910face 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -107,6 +107,7 @@ import type { BridgeFreshSessionAdmissionContext, BridgeFreshSessionReservation, BridgeOptions, + BridgeSessionLifecycleEvent, BridgeTelemetry, } from './bridgeOptions.js'; import { MCP_RESTART_SERVER_DEADLINE_MS } from './mcpTimeouts.js'; @@ -1134,6 +1135,17 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ); } }; + const emitSessionLifecycle = (event: BridgeSessionLifecycleEvent): void => { + try { + opts.sessionLifecycle?.(event); + } catch (err) { + const message = `qwen serve: session lifecycle callback failed: ${ + err instanceof Error ? err.message : String(err) + }`; + opts.onDiagnosticLine?.(message, 'warn'); + writeStderrLine(message); + } + }; if (defaultSessionScope !== 'single' && defaultSessionScope !== 'thread') { throw new TypeError( `Invalid sessionScope: ${JSON.stringify(defaultSessionScope)}. ` + @@ -1840,6 +1852,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { } byId.delete(sid); telemetry.metrics?.sessionLifecycle('die'); + emitSessionLifecycle({ + type: 'removed', + sessionId: sid, + workspaceCwd: sessEntry.workspaceCwd, + reason: 'channel_closed', + }); // Tombstone the id so any late `extNotification` from the // dying child can't leak into the early-event buffer for a // future load/resume of the same persisted session id. @@ -2649,7 +2667,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { sessionId: string, workspaceCwd: string, events = createSessionEventBus(), - options: { drainEarlyEvents?: boolean } = {}, + options: { drainEarlyEvents?: boolean; lifecycleReason?: string } = {}, ): SessionEntry => { const entry: SessionEntry = { sessionId, @@ -2679,6 +2697,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { byId.set(entry.sessionId, entry); touchActivity(); telemetry.metrics?.sessionLifecycle('spawn'); + emitSessionLifecycle({ + type: 'registered', + sessionId: entry.sessionId, + workspaceCwd: entry.workspaceCwd, + reason: options.lifecycleReason ?? 'spawn', + }); if (options.drainEarlyEvents !== false) { // Drain any guardrail events that fired during this session's // `newSession` handler (before this entry registered) onto the @@ -3077,7 +3101,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { req.sessionId, workspaceKey, restoreEvents, - { drainEarlyEvents: replayUpdates.length === 0 }, + { + drainEarlyEvents: replayUpdates.length === 0, + lifecycleReason: action, + }, ); releaseAdmissionOnce(); entry.restoreState = state; @@ -3129,9 +3156,16 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (!registeredEntry) { restoreEvents.close(); let removedRestoreEntry = false; - if (byId.get(req.sessionId)?.events === restoreEvents) { + const restoreEntry = byId.get(req.sessionId); + if (restoreEntry?.events === restoreEvents) { byId.delete(req.sessionId); ci?.sessionIds.delete(req.sessionId); + emitSessionLifecycle({ + type: 'removed', + sessionId: req.sessionId, + workspaceCwd: restoreEntry.workspaceCwd, + reason: 'restore_failed', + }); removedRestoreEntry = true; } if (removedRestoreEntry && ci && hasNoChannelWork(ci)) { @@ -3230,6 +3264,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { } byId.delete(sessionId); telemetry.metrics?.sessionLifecycle('close'); + emitSessionLifecycle({ + type: 'removed', + sessionId, + workspaceCwd: entry.workspaceCwd, + reason, + }); // Tombstone the closed sessionId so any late `extNotification` // from the (now-defunct) child can't seed the early-event buffer // and leak into a future load/resume of the same persisted id. @@ -6156,6 +6196,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (defaultEntry === entry) defaultEntry = undefined; byId.delete(sessionId); telemetry.metrics?.sessionLifecycle('die'); + emitSessionLifecycle({ + type: 'removed', + sessionId, + workspaceCwd: entry.workspaceCwd, + reason: 'killed', + }); // Detach from the channel. The channel dies only when its LAST // session leaves — other sessions on the same channel keep // running. @@ -6282,8 +6328,17 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { cancelIdleTimer(); stopSessionReaper(); const channels = Array.from(aliveChannels); + const entries = Array.from(byId.values()); defaultEntry = undefined; byId.clear(); + for (const entry of entries) { + emitSessionLifecycle({ + type: 'removed', + sessionId: entry.sessionId, + workspaceCwd: entry.workspaceCwd, + reason: 'kill_all', + }); + } for (const info of channels) { try { info.channel.killSync(); @@ -6335,6 +6390,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // and the automatic publish wouldn't fire. for (const e of entries) { telemetry.metrics?.sessionLifecycle('die'); + emitSessionLifecycle({ + type: 'removed', + sessionId: e.sessionId, + workspaceCwd: e.workspaceCwd, + reason: 'daemon_shutdown', + }); try { e.events.publish({ type: 'session_died', diff --git a/packages/acp-bridge/src/bridgeOptions.ts b/packages/acp-bridge/src/bridgeOptions.ts index 816a4b942a..97ea71f1ad 100644 --- a/packages/acp-bridge/src/bridgeOptions.ts +++ b/packages/acp-bridge/src/bridgeOptions.ts @@ -48,6 +48,24 @@ export type BridgeFreshSessionAdmission = ( context: BridgeFreshSessionAdmissionContext, ) => BridgeFreshSessionReservation | undefined; +export type BridgeSessionLifecycleEvent = + | { + readonly type: 'registered'; + readonly sessionId: string; + readonly workspaceCwd: string; + readonly reason: string; + } + | { + readonly type: 'removed'; + readonly sessionId: string; + readonly workspaceCwd: string; + readonly reason: string; + }; + +export type BridgeSessionLifecycle = ( + event: BridgeSessionLifecycleEvent, +) => void; + /** * Optional injection seam for daemon-host-specific status cells — * `process.env` snapshots and the daemon-side preflight checks @@ -168,6 +186,13 @@ export interface BridgeOptions { * side effect starts. Attaches bypass this hook. */ freshSessionAdmission?: BridgeFreshSessionAdmission; + /** + * Host-level live session owner callback. The bridge emits registration + * only after a live entry is installed, and removal when that live entry is + * removed. Callback failures are diagnostic only and do not fail session + * lifecycle operations. + */ + sessionLifecycle?: BridgeSessionLifecycle; /** * Per-session SSE replay ring depth. Sets `ringSize` on every * `new EventBus(...)` the bridge constructs (both fresh sessions diff --git a/packages/cli/src/serve/acp-session-bridge.ts b/packages/cli/src/serve/acp-session-bridge.ts index fe27e5870c..9913c74017 100644 --- a/packages/cli/src/serve/acp-session-bridge.ts +++ b/packages/cli/src/serve/acp-session-bridge.ts @@ -58,6 +58,8 @@ export type { BridgeFreshSessionAdmission, BridgeFreshSessionAdmissionContext, BridgeFreshSessionReservation, + BridgeSessionLifecycle, + BridgeSessionLifecycleEvent, BridgeOptions, DaemonStatusProvider, } from '@qwen-code/acp-bridge/bridgeOptions'; diff --git a/packages/cli/src/serve/multi-workspace-sessions.test.ts b/packages/cli/src/serve/multi-workspace-sessions.test.ts index b4cf9775d7..56e1a75c62 100644 --- a/packages/cli/src/serve/multi-workspace-sessions.test.ts +++ b/packages/cli/src/serve/multi-workspace-sessions.test.ts @@ -63,6 +63,7 @@ interface FakeBridge extends AcpSessionBridge { action: 'load' | 'resume'; req: BridgeRestoreSessionRequest; }>; + readonly listCalls: string[]; } function makeSummary( @@ -101,6 +102,7 @@ function makeBridge( const pendingPromptCalls: string[] = []; const removePendingPromptCalls: FakeBridge['removePendingPromptCalls'] = []; const restoreCalls: FakeBridge['restoreCalls'] = []; + const listCalls: string[] = []; const bridge = { permissionPolicy: 'first-responder' as const, spawnCalls, @@ -114,6 +116,7 @@ function makeBridge( pendingPromptCalls, removePendingPromptCalls, restoreCalls, + listCalls, get sessionCount() { return live.size; }, @@ -186,6 +189,7 @@ function makeBridge( }; }, listWorkspaceSessions(cwd: string) { + listCalls.push(cwd); return [...live.values()].filter( (summary) => summary.workspaceCwd === cwd, ); @@ -630,7 +634,7 @@ describe('multi-workspace session dispatch', () => { ); }); - it('keeps persisted load and resume primary-only before touching a bridge', async () => { + it('dispatches trusted non-primary persisted load and resume', async () => { const { app, primaryBridge, secondaryBridge } = makeHarness(); for (const action of ['load', 'resume'] as const) { @@ -639,13 +643,61 @@ describe('multi-workspace session dispatch', () => { .set('Host', host()) .send({ cwd: SECONDARY_CWD }); - expect(res.status).toBe(400); - expect(res.body.code).toBe('secondary_workspace_load_not_supported'); + expect(res.status).toBe(200); expect(res.body.workspaceCwd).toBe(SECONDARY_CWD); } expect(primaryBridge.restoreCalls).toEqual([]); - expect(secondaryBridge.restoreCalls).toEqual([]); + expect(secondaryBridge.restoreCalls).toEqual([ + { + action: 'load', + req: expect.objectContaining({ + sessionId: 'secondary-session', + workspaceCwd: SECONDARY_CWD, + }), + }, + { + action: 'resume', + req: expect.objectContaining({ + sessionId: 'secondary-session', + workspaceCwd: SECONDARY_CWD, + }), + }, + ]); + }); + + it('rejects unknown and untrusted restore cwd before touching a bridge', async () => { + const unknown = makeHarness(); + const unknownRes = await request(unknown.app) + .post('/session/unknown-restore/load') + .set('Host', host()) + .send({ cwd: UNKNOWN_CWD }); + + expect(unknownRes.status).toBe(400); + expect(unknownRes.body.code).toBe('workspace_mismatch'); + expect(unknownRes.body.workspaceCount).toBe(2); + expect(unknown.primaryBridge.restoreCalls).toEqual([]); + expect(unknown.secondaryBridge.restoreCalls).toEqual([]); + + const daemonLog = makeDaemonLog(); + const untrusted = makeHarness({ secondaryTrusted: false, daemonLog }); + const untrustedRes = await request(untrusted.app) + .post('/session/untrusted-restore/resume') + .set('Host', host()) + .send({ cwd: SECONDARY_CWD }); + + expect(untrustedRes.status).toBe(403); + expect(untrustedRes.body.code).toBe('untrusted_workspace'); + expect(untrusted.primaryBridge.restoreCalls).toEqual([]); + expect(untrusted.secondaryBridge.restoreCalls).toEqual([]); + expect(daemonLog.warn).toHaveBeenCalledWith( + 'session routing failed', + expect.objectContaining({ + route: 'POST /session/:id/resume', + resolutionKind: 'untrusted_workspace', + workspaceCwd: SECONDARY_CWD, + }), + ); }); it('returns a clear Phase 2a error for non-primary sessions on primary-only routes', async () => { @@ -692,6 +744,44 @@ describe('multi-workspace session dispatch', () => { expect(unknown.body.workspaceCount).toBe(2); }); + it('preserves the legacy invalid workspace selector message', async () => { + const { app } = makeHarness(); + + const res = await request(app) + .get('/workspace/not:an:absolute:path/sessions') + .set('Host', host()); + + expect(res.status).toBe(400); + expect(res.body.error).toBe( + '`:id` must decode to a workspace id or absolute path', + ); + }); + + it('rejects untrusted non-primary workspace session listing', async () => { + const daemonLog = makeDaemonLog(); + const { app, secondaryBridge } = makeHarness({ + secondaryTrusted: false, + daemonLog, + }); + + const res = await request(app) + .get('/workspaces/secondary-id/sessions') + .set('Host', host()); + + expect(res.status).toBe(403); + expect(res.body.code).toBe('untrusted_workspace'); + expect(res.body.workspaceCwd).toBe(SECONDARY_CWD); + expect(secondaryBridge.listCalls).toEqual([]); + expect(daemonLog.warn).toHaveBeenCalledWith( + 'session routing failed', + expect.objectContaining({ + route: 'GET /workspaces/:workspace/sessions', + resolutionKind: 'untrusted_workspace', + workspaceCwd: SECONDARY_CWD, + }), + ); + }); + it('pages live non-primary workspace sessions with a stable cursor', async () => { const { app } = makeHarness({ secondarySummaries: [ diff --git a/packages/cli/src/serve/rate-limit.test.ts b/packages/cli/src/serve/rate-limit.test.ts index 38ac3212e1..8c75573282 100644 --- a/packages/cli/src/serve/rate-limit.test.ts +++ b/packages/cli/src/serve/rate-limit.test.ts @@ -219,6 +219,29 @@ describe('rateLimit', () => { expect(res.body).toMatchObject({ tier: 'read' }); }); + it('classifies plural workspace session listing as read tier', () => { + const next = vi.fn(); + limiter.middleware( + mockReq({ + method: 'GET', + path: '/workspaces/ws-secondary/sessions', + }), + mockRes(), + next, + ); + expect(next).toHaveBeenCalled(); + const res = mockRes(); + limiter.middleware( + mockReq({ + method: 'GET', + path: '/workspaces/ws-secondary/sessions', + }), + res, + vi.fn(), + ); + expect(res.body).toMatchObject({ tier: 'read' }); + }); + it('exempts GET /health', () => { const next = vi.fn(); for (let i = 0; i < 5; i++) { diff --git a/packages/cli/src/serve/routes/session.ts b/packages/cli/src/serve/routes/session.ts index 5a75da304b..31433f78bd 100644 --- a/packages/cli/src/serve/routes/session.ts +++ b/packages/cli/src/serve/routes/session.ts @@ -223,13 +223,14 @@ export function registerSessionRoutes( const resolveRuntimeFromWorkspaceParam = ( req: Request, res: Response, + paramName = 'id', ): WorkspaceRuntime | null => { - const workspaceParam = req.params['id'] ?? ''; + const workspaceParam = req.params[paramName] ?? ''; const byId = workspaceRegistry.getByWorkspaceId(workspaceParam); if (byId) return byId; if (!path.isAbsolute(workspaceParam)) { res.status(400).json({ - error: '`:id` must decode to a workspace id or absolute path', + error: `\`:${paramName}\` must decode to a workspace id or absolute path`, }); return null; } @@ -248,35 +249,164 @@ export function registerSessionRoutes( return runtime; }; - const resolvePrimaryOnlyWorkspaceCwd = ( - cwd: string, + const sendAmbiguousSessionOwner = ( res: Response, route: string, - ): string | undefined => { + sessionId: string, + runtimes: readonly WorkspaceRuntime[], + ): void => { + const workspaceIds = runtimes.map((runtime) => runtime.workspaceId); + logSessionRoutingFailure(route, 'ambiguous', { + sessionId, + workspaceIds, + }); + res.status(500).json({ + error: `Session owner is ambiguous for "${sessionId}"`, + code: 'ambiguous_session_owner', + sessionId, + route, + workspaceIds, + }); + }; + const inFlightRestoreOwners = new Map< + string, + { workspaceCwd: string; count: number } + >(); + + const sendSessionWorkspaceConflict = ( + res: Response, + route: string, + sessionId: string, + runtime: WorkspaceRuntime, + liveRuntime: WorkspaceRuntime, + ): void => { + logSessionRoutingFailure(route, 'workspace_conflict', { + sessionId, + workspaceId: runtime.workspaceId, + workspaceCwd: runtime.workspaceCwd, + liveWorkspaceId: liveRuntime.workspaceId, + liveWorkspaceCwd: liveRuntime.workspaceCwd, + }); + res.status(409).json({ + error: `Session "${sessionId}" is already live or restoring in another workspace runtime.`, + code: 'session_workspace_conflict', + sessionId, + workspaceCwd: runtime.workspaceCwd, + workspaceId: runtime.workspaceId, + liveWorkspaceCwd: liveRuntime.workspaceCwd, + liveWorkspaceId: liveRuntime.workspaceId, + }); + }; + + const enterRestoreOwner = ( + res: Response, + route: string, + sessionId: string, + runtime: WorkspaceRuntime, + ): (() => void) | undefined => { + const existing = inFlightRestoreOwners.get(sessionId); + if (existing && existing.workspaceCwd !== runtime.workspaceCwd) { + const existingRuntime = + workspaceRegistry.getByWorkspaceCwd(existing.workspaceCwd) ?? + workspaceRegistry.primary; + sendSessionWorkspaceConflict( + res, + route, + sessionId, + runtime, + existingRuntime, + ); + return undefined; + } + + if (existing) { + existing.count += 1; + } else { + inFlightRestoreOwners.set(sessionId, { + workspaceCwd: runtime.workspaceCwd, + count: 1, + }); + } + let released = false; + return () => { + if (released) return; + released = true; + const current = inFlightRestoreOwners.get(sessionId); + if (!current || current.workspaceCwd !== runtime.workspaceCwd) return; + current.count -= 1; + if (current.count <= 0) { + inFlightRestoreOwners.delete(sessionId); + } + }; + }; + + const resolveRuntimeForSessionRestore = ( + body: Record, + res: Response, + route: string, + sessionId: string, + ): { runtime: WorkspaceRuntime; workspaceCwd: string } | undefined => { + const cwd = parseOptionalWorkspaceCwd(body, boundWorkspace, res); + if (cwd === undefined) return undefined; let key: string; try { key = canonicalizeWorkspace(cwd); } catch (err) { - sendBridgeError(res, err, { route }); - return undefined; - } - if (key !== boundWorkspace) { - const runtime = workspaceRegistry.getByWorkspaceCwd(key); - if (runtime && !runtime.primary) { - res.status(400).json({ - error: - 'Non-primary persisted session load/resume is not supported in Phase 2a.', - code: 'secondary_workspace_load_not_supported', - workspaceCwd: runtime.workspaceCwd, - workspaceId: runtime.workspaceId, - route, + if ('cwd' in body) { + logSessionRoutingFailure(route, 'workspace_mismatch', { + requestedWorkspace: cwd, }); + sendWorkspaceMismatch(res, cwd); return undefined; } + sendBridgeError(res, err, { route, sessionId }); + return undefined; + } + + const runtime = workspaceRegistry.resolveWorkspaceCwd( + 'cwd' in body ? key : undefined, + ); + if (!runtime) { + logSessionRoutingFailure(route, 'workspace_mismatch', { + requestedWorkspace: key, + }); sendWorkspaceMismatch(res, key); return undefined; } - return key; + if (!runtime.primary && !runtime.trusted) { + logSessionRoutingFailure(route, 'untrusted_workspace', { + workspaceId: runtime.workspaceId, + workspaceCwd: runtime.workspaceCwd, + }); + res.status(403).json({ + error: `Workspace "${runtime.workspaceCwd}" is not trusted.`, + code: 'untrusted_workspace', + workspaceCwd: runtime.workspaceCwd, + workspaceId: runtime.workspaceId, + }); + return undefined; + } + + const liveOwner = workspaceRegistry.resolveLiveSessionOwner(sessionId); + if (liveOwner.kind === 'ambiguous') { + sendAmbiguousSessionOwner(res, route, sessionId, liveOwner.runtimes); + return undefined; + } + if ( + liveOwner.kind === 'found' && + liveOwner.runtime.workspaceCwd !== runtime.workspaceCwd + ) { + sendSessionWorkspaceConflict( + res, + route, + sessionId, + runtime, + liveOwner.runtime, + ); + return undefined; + } + + return { runtime, workspaceCwd: runtime.workspaceCwd }; }; const resolveLiveSessionRuntime = ( @@ -321,9 +451,9 @@ export function registerSessionRoutes( async (req, res) => { const sessionId = requireSessionId(req, res); if (sessionId === null) return; - const runtime = resolveLiveSessionRuntime(sessionId, res, route); - if (!runtime) return; try { + const runtime = resolveLiveSessionRuntime(sessionId, res, route); + if (!runtime) return; await archiveCoordinator.runSharedMany([sessionId], async () => { await handler(req, res, sessionId, runtime); }); @@ -332,6 +462,28 @@ export function registerSessionRoutes( } }; + const withOwnerReadSession = + ( + route: string, + handler: ( + req: Request, + res: Response, + sessionId: string, + runtime: WorkspaceRuntime, + ) => Promise | void, + ): RequestHandler => + async (req, res) => { + const sessionId = requireSessionId(req, res); + if (sessionId === null) return; + try { + const runtime = resolveLiveSessionRuntime(sessionId, res, route); + if (!runtime) return; + await handler(req, res, sessionId, runtime); + } catch (err) { + sendBridgeError(res, err, { route, sessionId }); + } + }; + const parseSessionIdsBody = ( req: Request, res: Response, @@ -523,31 +675,50 @@ export function registerSessionRoutes( const sessionId = requireSessionId(req, res); if (!sessionId) return; const body = safeBody(req); - const cwd = parseOptionalWorkspaceCwd(body, boundWorkspace, res); - if (cwd === undefined) return; - const primaryCwd = resolvePrimaryOnlyWorkspaceCwd( - cwd, + const route = `POST /session/:id/${action}`; + let resolvedRuntime: + | { runtime: WorkspaceRuntime; workspaceCwd: string } + | undefined; + try { + resolvedRuntime = resolveRuntimeForSessionRestore( + body, + res, + route, + sessionId, + ); + } catch (err) { + sendBridgeError(res, err, { route, sessionId }); + return; + } + if (resolvedRuntime === undefined) return; + const { runtime, workspaceCwd } = resolvedRuntime; + const releaseRestoreOwner = enterRestoreOwner( res, - `POST /session/:id/${action}`, + route, + sessionId, + runtime, ); - if (primaryCwd === undefined) return; + if (!releaseRestoreOwner) return; const clientId = parseClientIdHeader(req, res); - if (clientId === null) return; + if (clientId === null) { + releaseRestoreOwner(); + return; + } try { const session = await archiveCoordinator.runSharedMany( [sessionId], async () => { - await assertSessionLoadable(primaryCwd, sessionId); + await assertSessionLoadable(workspaceCwd, sessionId); return action === 'load' - ? await bridge.loadSession({ + ? await runtime.bridge.loadSession({ sessionId, - workspaceCwd: primaryCwd, + workspaceCwd, historyReplay: 'response', ...(clientId !== undefined ? { clientId } : {}), }) - : await bridge.resumeSession({ + : await runtime.bridge.resumeSession({ sessionId, - workspaceCwd: primaryCwd, + workspaceCwd, ...(clientId !== undefined ? { clientId } : {}), }); }, @@ -568,13 +739,13 @@ export function registerSessionRoutes( // restored session in `byId` with no client holding its id. if (!res.writable) { if (!session.attached) { - bridge + runtime.bridge .killSession(session.sessionId, { requireZeroAttaches: true }) .catch(() => { // Best-effort cleanup; channel.exited will eventually reap. }); } else { - bridge + runtime.bridge .detachClient(session.sessionId, session.clientId) .catch(() => { // Best-effort cleanup; channel.exited will eventually reap. @@ -585,9 +756,11 @@ export function registerSessionRoutes( res.status(200).json(session); } catch (err) { sendBridgeError(res, err, { - route: `POST /session/:id/${action}`, + route, sessionId, }); + } finally { + releaseRestoreOwner(); } }; @@ -753,113 +926,106 @@ export function registerSessionRoutes( } }); - app.get('/session/:id/context', async (req, res) => { - const sessionId = requireSessionId(req, res); - if (sessionId === null) return; - try { - res.status(200).json(await bridge.getSessionContextStatus(sessionId)); - } catch (err) { - sendBridgeError(res, err, { - route: 'GET /session/:id/context', - sessionId, - }); - } - }); + app.get( + '/session/:id/context', + withOwnerReadSession( + 'GET /session/:id/context', + async (_req, res, sessionId, runtime) => { + res + .status(200) + .json(await runtime.bridge.getSessionContextStatus(sessionId)); + }, + ), + ); - app.get('/session/:id/context-usage', async (req, res) => { - const sessionId = requireSessionId(req, res); - if (sessionId === null) return; - try { - res.status(200).json( - await bridge.getSessionContextUsageStatus(sessionId, { - detail: req.query['detail'] === 'true', - }), - ); - } catch (err) { - sendBridgeError(res, err, { - route: 'GET /session/:id/context-usage', - sessionId, - }); - } - }); + app.get( + '/session/:id/context-usage', + withOwnerReadSession( + 'GET /session/:id/context-usage', + async (req, res, sessionId, runtime) => { + res.status(200).json( + await runtime.bridge.getSessionContextUsageStatus(sessionId, { + detail: req.query['detail'] === 'true', + }), + ); + }, + ), + ); - app.get('/session/:id/stats', async (req, res) => { - const sessionId = requireSessionId(req, res); - if (sessionId === null) return; - try { - res.status(200).json(await bridge.getSessionStatsStatus(sessionId)); - } catch (err) { - sendBridgeError(res, err, { - route: 'GET /session/:id/stats', - sessionId, - }); - } - }); + app.get( + '/session/:id/stats', + withOwnerReadSession( + 'GET /session/:id/stats', + async (_req, res, sessionId, runtime) => { + res + .status(200) + .json(await runtime.bridge.getSessionStatsStatus(sessionId)); + }, + ), + ); - app.get('/session/:id/supported-commands', async (req, res) => { - const sessionId = requireSessionId(req, res); - if (sessionId === null) return; - try { - res - .status(200) - .json(await bridge.getSessionSupportedCommandsStatus(sessionId)); - } catch (err) { - sendBridgeError(res, err, { - route: 'GET /session/:id/supported-commands', - sessionId, - }); - } - }); + app.get( + '/session/:id/supported-commands', + withOwnerReadSession( + 'GET /session/:id/supported-commands', + async (_req, res, sessionId, runtime) => { + res + .status(200) + .json( + await runtime.bridge.getSessionSupportedCommandsStatus(sessionId), + ); + }, + ), + ); - app.get('/session/:id/tasks', async (req, res) => { - const sessionId = requireSessionId(req, res); - if (sessionId === null) return; - try { - res.status(200).json(await bridge.getSessionTasksStatus(sessionId)); - } catch (err) { - sendBridgeError(res, err, { - route: 'GET /session/:id/tasks', - sessionId, - }); - } - }); + app.get( + '/session/:id/tasks', + withOwnerReadSession( + 'GET /session/:id/tasks', + async (_req, res, sessionId, runtime) => { + res + .status(200) + .json(await runtime.bridge.getSessionTasksStatus(sessionId)); + }, + ), + ); - app.get('/session/:id/lsp', async (req, res) => { - const sessionId = requireSessionId(req, res); - if (sessionId === null) return; - try { - res.status(200).json(await bridge.getSessionLspStatus(sessionId)); - } catch (err) { - sendBridgeError(res, err, { - route: 'GET /session/:id/lsp', - sessionId, - }); - } - }); + app.get( + '/session/:id/lsp', + withOwnerReadSession( + 'GET /session/:id/lsp', + async (_req, res, sessionId, runtime) => { + res + .status(200) + .json(await runtime.bridge.getSessionLspStatus(sessionId)); + }, + ), + ); // GET /session/:id/hooks — read-only session-scoped hook status. - app.get('/session/:id/hooks', async (req, res) => { - const sessionId = requireSessionId(req, res); - if (sessionId === null) return; - try { - res.status(200).json(await bridge.getSessionHooksStatus(sessionId)); - } catch (err) { - sendBridgeError(res, err, { route: 'GET /session/:id/hooks', sessionId }); - } - }); + app.get( + '/session/:id/hooks', + withOwnerReadSession( + 'GET /session/:id/hooks', + async (_req, res, sessionId, runtime) => { + res + .status(200) + .json(await runtime.bridge.getSessionHooksStatus(sessionId)); + }, + ), + ); - app.get('/session/:id/artifacts', async (req, res) => { - const sessionId = requireSessionId(req, res); - if (sessionId === null) return; - try { - res.status(200).json(await bridge.getSessionArtifacts(sessionId)); - } catch (err) { - sendBridgeError(res, err, { - route: 'GET /session/:id/artifacts', - sessionId, - }); - } - }); + app.get( + '/session/:id/artifacts', + withOwnerReadSession( + 'GET /session/:id/artifacts', + async (_req, res, sessionId, runtime) => { + res + .status(200) + .json(await runtime.bridge.getSessionArtifacts(sessionId)); + }, + ), + ); app.post( '/session/:id/artifacts', @@ -1516,99 +1682,128 @@ export function registerSessionRoutes( }, ); - app.get('/workspace/:id/sessions', async (req, res) => { - // Express decodes URL-encoded path params automatically; clients pass - // the absolute workspace cwd encoded (e.g. - // GET /workspace/%2Fwork%2Fa/sessions). - const runtime = resolveRuntimeFromWorkspaceParam(req, res); - if (runtime === null) return; - const key = runtime.workspaceCwd; - try { - const cursor = - typeof req.query['cursor'] === 'string' - ? req.query['cursor'] - : undefined; - const size = parseSessionPageSizeQuery(req.query['size']); - const rawView = req.query['view']; - let view: 'organized' | undefined; - if (rawView !== undefined) { - if (rawView !== 'organized') { + const listWorkspaceSessionsHandler = + (paramName: string): RequestHandler => + async (req, res) => { + const route = + paramName === 'workspace' + ? 'GET /workspaces/:workspace/sessions' + : 'GET /workspace/:id/sessions'; + // Express decodes URL-encoded path params automatically; clients pass + // the absolute workspace cwd encoded (e.g. + // GET /workspace/%2Fwork%2Fa/sessions). + const runtime = resolveRuntimeFromWorkspaceParam(req, res, paramName); + if (runtime === null) return; + if (!runtime.primary && !runtime.trusted) { + logSessionRoutingFailure(route, 'untrusted_workspace', { + workspaceId: runtime.workspaceId, + workspaceCwd: runtime.workspaceCwd, + }); + res.status(403).json({ + error: `Workspace "${runtime.workspaceCwd}" is not trusted.`, + code: 'untrusted_workspace', + workspaceCwd: runtime.workspaceCwd, + workspaceId: runtime.workspaceId, + }); + return; + } + const key = runtime.workspaceCwd; + try { + const cursor = + typeof req.query['cursor'] === 'string' + ? req.query['cursor'] + : undefined; + const size = parseSessionPageSizeQuery(req.query['size']); + const rawView = req.query['view']; + let view: 'organized' | undefined; + if (rawView !== undefined) { + if (rawView !== 'organized') { + res.status(400).json({ + error: '`view` must be "organized"', + code: 'invalid_session_view', + }); + return; + } + view = 'organized'; + } + const group = + typeof req.query['group'] === 'string' + ? req.query['group'] + : undefined; + if (group !== undefined && view !== 'organized') { res.status(400).json({ - error: '`view` must be "organized"', - code: 'invalid_session_view', + error: '`group` requires `view=organized`', + code: 'invalid_session_group_filter', }); return; } - view = 'organized'; - } - const group = - typeof req.query['group'] === 'string' ? req.query['group'] : undefined; - if (group !== undefined && view !== 'organized') { - res.status(400).json({ - error: '`group` requires `view=organized`', - code: 'invalid_session_group_filter', - }); - return; - } - const rawArchiveState = req.query['archiveState']; - let archiveState: SessionArchiveState | undefined; - if (rawArchiveState !== undefined) { - if ( - typeof rawArchiveState !== 'string' || - (rawArchiveState !== 'active' && rawArchiveState !== 'archived') - ) { + const rawArchiveState = req.query['archiveState']; + let archiveState: SessionArchiveState | undefined; + if (rawArchiveState !== undefined) { + if ( + typeof rawArchiveState !== 'string' || + (rawArchiveState !== 'active' && rawArchiveState !== 'archived') + ) { + res.status(400).json({ + error: '`archiveState` must be "active" or "archived"', + code: 'invalid_archive_state', + }); + return; + } + archiveState = rawArchiveState; + } + if (!runtime.primary && (archiveState === 'archived' || view)) { res.status(400).json({ - error: '`archiveState` must be "active" or "archived"', - code: 'invalid_archive_state', + error: + 'Non-primary workspace session listing is live-only in Phase 2a.', + code: 'non_primary_live_sessions_only', }); return; } - archiveState = rawArchiveState; - } - if (!runtime.primary && (archiveState === 'archived' || view)) { - res.status(400).json({ - error: - 'Non-primary workspace session listing is live-only in Phase 2a.', - code: 'non_primary_live_sessions_only', + const options = { + ...(cursor !== undefined ? { cursor } : {}), + ...(size !== undefined ? { size } : {}), + ...(archiveState !== undefined ? { archiveState } : {}), + ...(view !== undefined ? { view } : {}), + ...(group !== undefined ? { group } : {}), + }; + const result = runtime.primary + ? await listWorkspaceSessionsForResponse(runtime.bridge, key, options) + : listLiveWorkspaceSessionsForResponse(runtime.bridge, key, options); + res.status(200).json({ + sessions: result.sessions, + ...(result.nextCursor != null + ? { nextCursor: result.nextCursor } + : {}), + ...(result.liveMergeFailed ? { liveMergeFailed: true } : {}), + ...(result.truncated ? { truncated: true } : {}), }); - return; - } - const options = { - ...(cursor !== undefined ? { cursor } : {}), - ...(size !== undefined ? { size } : {}), - ...(archiveState !== undefined ? { archiveState } : {}), - ...(view !== undefined ? { view } : {}), - ...(group !== undefined ? { group } : {}), - }; - const result = runtime.primary - ? await listWorkspaceSessionsForResponse(runtime.bridge, key, options) - : listLiveWorkspaceSessionsForResponse(runtime.bridge, key, options); - res.status(200).json({ - sessions: result.sessions, - ...(result.nextCursor != null ? { nextCursor: result.nextCursor } : {}), - ...(result.liveMergeFailed ? { liveMergeFailed: true } : {}), - ...(result.truncated ? { truncated: true } : {}), - }); - } catch (err) { - if (err instanceof InvalidCursorError) { - res.status(400).json({ - error: err.message, - code: 'invalid_cursor', + } catch (err) { + if (err instanceof InvalidCursorError) { + res.status(400).json({ + error: err.message, + code: 'invalid_cursor', + }); + return; + } + if (sendSessionOrganizationError(res, err)) return; + writeStderrLine( + `qwen serve: failed to list sessions for workspace ${safeLogValue( + key, + )}: ${safeLogValue(err instanceof Error ? err.message : String(err))}`, + ); + res.status(500).json({ + error: 'Failed to list sessions', + code: 'session_list_failed', }); - return; } - if (sendSessionOrganizationError(res, err)) return; - writeStderrLine( - `qwen serve: failed to list sessions for workspace ${safeLogValue( - key, - )}: ${safeLogValue(err instanceof Error ? err.message : String(err))}`, - ); - res.status(500).json({ - error: 'Failed to list sessions', - code: 'session_list_failed', - }); - } - }); + }; + + app.get('/workspace/:id/sessions', listWorkspaceSessionsHandler('id')); + app.get( + '/workspaces/:workspace/sessions', + listWorkspaceSessionsHandler('workspace'), + ); app.post( '/session/:id/model', diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 4cabb08585..84e82ba187 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/packages/cli/src/serve/run-qwen-serve.ts @@ -799,6 +799,8 @@ async function loadServeRuntimeModules() { createTotalSessionAdmissionController: totalSessionAdmissionModule.createTotalSessionAdmissionController, createWorkspaceRegistry: workspaceRegistryModule.createWorkspaceRegistry, + createWorkspaceSessionOwnerIndex: + workspaceRegistryModule.createWorkspaceSessionOwnerIndex, }; } @@ -2456,6 +2458,7 @@ export async function runQwenServe( : [], }, ); + const sessionOwnerIndex = runtime.createWorkspaceSessionOwnerIndex(); const persistDisabledToolsFn = ( workspace: string, toolName: string, @@ -2536,6 +2539,7 @@ export async function runQwenServe( clientMcpSender: clientMcpSenderRegistry.lookup, maxSessions: opts.maxSessions, freshSessionAdmission: totalSessionAdmission.admit, + sessionLifecycle: sessionOwnerIndex.handleBridgeSessionLifecycle, ...(opts.maxPendingPromptsPerSession !== undefined ? { maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession } : {}), @@ -2803,6 +2807,7 @@ export async function runQwenServe( clientMcpSender: secondaryClientMcpSenderRegistry.lookup, maxSessions: opts.maxSessions, freshSessionAdmission: totalSessionAdmission.admit, + sessionLifecycle: sessionOwnerIndex.handleBridgeSessionLifecycle, ...(opts.maxPendingPromptsPerSession !== undefined ? { maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession } : {}), @@ -2933,7 +2938,9 @@ export async function runQwenServe( } const workspaceRegistry: WorkspaceRegistry = - runtime.createWorkspaceRegistry(workspaceRuntimes); + runtime.createWorkspaceRegistry(workspaceRuntimes, { + sessionOwnerIndex, + }); core.registerDaemonGaugeCallbacks({ sessionCount: () => diff --git a/packages/cli/src/serve/server-default-bridge-wiring.test.ts b/packages/cli/src/serve/server-default-bridge-wiring.test.ts new file mode 100644 index 0000000000..af442ba19f --- /dev/null +++ b/packages/cli/src/serve/server-default-bridge-wiring.test.ts @@ -0,0 +1,153 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + SessionNotFoundError, + type AcpSessionBridge, + type BridgeFreshSessionAdmission, + type BridgeOptions, + type BridgeSessionSummary, +} from './acp-session-bridge.js'; +import type { WorkspaceRegistry } from './workspace-registry.js'; + +const WS_BOUND = '/work/bound'; + +function makeBridge( + sessionCount = 0, + liveSessionIds?: ReadonlySet, +): AcpSessionBridge { + const getSessionSummary = (sessionId: string): BridgeSessionSummary => { + if (liveSessionIds && !liveSessionIds.has(sessionId)) { + throw new SessionNotFoundError(sessionId); + } + return { + sessionId, + workspaceCwd: WS_BOUND, + createdAt: '2026-05-17T12:00:00.000Z', + clientCount: 1, + hasActivePrompt: false, + }; + }; + + return { + get sessionCount() { + return sessionCount; + }, + getSessionSummary, + async shutdown() {}, + killAllSync() {}, + } as unknown as AcpSessionBridge; +} + +describe('createServeApp default bridge wiring', () => { + afterEach(() => { + vi.doUnmock('./acp-session-bridge.js'); + vi.resetModules(); + vi.restoreAllMocks(); + }); + + it('wires the internally-created bridge lifecycle into the workspace registry', async () => { + let sessionLifecycle: BridgeOptions['sessionLifecycle']; + const liveSessionIds = new Set(); + const bridge = makeBridge(0, liveSessionIds); + vi.doMock('./acp-session-bridge.js', async () => { + const actual = await vi.importActual< + typeof import('./acp-session-bridge.js') + >('./acp-session-bridge.js'); + return { + ...actual, + createAcpSessionBridge: vi.fn((opts: BridgeOptions) => { + sessionLifecycle = opts.sessionLifecycle; + return bridge; + }), + }; + }); + + const { createServeApp } = await import('./server.js'); + const app = createServeApp( + { + port: 0, + hostname: '127.0.0.1', + workspace: WS_BOUND, + } as Parameters[0], + () => 0, + ); + const locals = app.locals as { workspaceRegistry?: WorkspaceRegistry }; + + expect(sessionLifecycle).toBeDefined(); + liveSessionIds.add('session-indexed'); + sessionLifecycle!({ + type: 'registered', + sessionId: 'session-indexed', + workspaceCwd: WS_BOUND, + reason: 'spawn', + }); + expect( + locals.workspaceRegistry!.resolveLiveSessionOwner('session-indexed'), + ).toEqual({ + kind: 'found', + runtime: locals.workspaceRegistry!.primary, + }); + liveSessionIds.delete('session-indexed'); + sessionLifecycle!({ + type: 'removed', + sessionId: 'session-indexed', + workspaceCwd: WS_BOUND, + reason: 'client_close', + }); + expect( + locals.workspaceRegistry!.resolveLiveSessionOwner('session-indexed'), + ).toEqual({ + kind: 'not_found', + }); + }, 15_000); + + it('wires total admission into the internally-created bridge', async () => { + let freshSessionAdmission: BridgeFreshSessionAdmission | undefined; + vi.doMock('./acp-session-bridge.js', async () => { + const actual = await vi.importActual< + typeof import('./acp-session-bridge.js') + >('./acp-session-bridge.js'); + return { + ...actual, + createAcpSessionBridge: vi.fn((opts: BridgeOptions) => { + freshSessionAdmission = opts.freshSessionAdmission; + return makeBridge(1); + }), + }; + }); + + const { createServeApp } = await import('./server.js'); + createServeApp( + { + port: 0, + hostname: '127.0.0.1', + workspace: WS_BOUND, + maxTotalSessions: 1, + } as Parameters[0], + () => 0, + ); + + expect(freshSessionAdmission).toBeDefined(); + let rejection: unknown; + try { + freshSessionAdmission!({ + operation: 'spawn', + workspaceCwd: WS_BOUND, + }); + } catch (err) { + rejection = err; + } + expect(rejection).toMatchObject({ + name: 'TotalSessionLimitExceededError', + limit: 1, + scope: 'total', + operation: 'spawn', + workspaceCwd: WS_BOUND, + }); + }); +}); diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 79b91e1afb..04da1c4e6f 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -184,6 +184,20 @@ function restoreEnv(key: string, value: string | undefined): void { } } +function deferred(): { + promise: Promise; + resolve: (value: T | PromiseLike) => void; + reject: (reason?: unknown) => void; +} { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + // Workspace fixtures must round-trip through `path.resolve` so the // expected values match the canonicalized form the route produces on // every platform. On Windows `path.resolve('/work/bound')` returns @@ -1801,6 +1815,26 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { }; } +function makeWorkspaceRuntimeForTest(input: { + workspaceId: string; + workspaceCwd: string; + primary: boolean; + bridge: AcpSessionBridge; + trusted?: boolean; +}): WorkspaceRuntime { + return { + workspaceId: input.workspaceId, + workspaceCwd: input.workspaceCwd, + primary: input.primary, + trusted: input.trusted ?? true, + env: { mode: 'parent-process', overlayKeys: [] }, + bridge: input.bridge, + workspaceService: {} as DaemonWorkspaceService, + routeFileSystemFactory: {} as WorkspaceFileSystemFactory, + clientMcpSenderRegistry: new ClientMcpSenderRegistry(), + }; +} + /** * Wenshao review #4335 / 3272581557 — detectFromLoopback tests. */ @@ -4912,6 +4946,93 @@ describe('createServeApp', () => { expect(bridge.sessionLspCalls).toEqual(['s-1']); }); + it('dispatches read-only session snapshots through the live owner runtime', async () => { + const primaryBridge = fakeBridge(); + const secondaryBridge = fakeBridge({ + summaryImpl: (sessionId) => ({ + sessionId, + workspaceCwd: WS_DIFFERENT, + createdAt: '2026-05-17T12:00:00.000Z', + clientCount: 1, + hasActivePrompt: false, + }), + sessionContextImpl: async (sessionId) => ({ + v: 1, + sessionId, + workspaceCwd: WS_DIFFERENT, + state: { owner: 'secondary' }, + }), + }); + const registry = createWorkspaceRegistry([ + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-primary', + workspaceCwd: WS_BOUND, + primary: true, + bridge: primaryBridge, + }), + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-secondary', + workspaceCwd: WS_DIFFERENT, + primary: false, + bridge: secondaryBridge, + }), + ]); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { workspaceRegistry: registry }, + ); + + const res = await request(app) + .get('/session/s-secondary/context') + .set('Host', `127.0.0.1:${baseOpts.port}`); + + expect(res.status).toBe(200); + expect(res.body).toMatchObject({ + sessionId: 's-secondary', + workspaceCwd: WS_DIFFERENT, + state: { owner: 'secondary' }, + }); + expect(primaryBridge.sessionContextCalls).toEqual([]); + expect(secondaryBridge.sessionContextCalls).toEqual(['s-secondary']); + }); + + it('surfaces live owner scan failures as structured bridge errors on owner-routed reads', async () => { + const primaryBridge = fakeBridge(); + const secondaryBridge = fakeBridge({ + summaryImpl: () => { + throw new Error('summary exploded'); + }, + }); + const registry = createWorkspaceRegistry([ + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-primary', + workspaceCwd: WS_BOUND, + primary: true, + bridge: primaryBridge, + }), + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-secondary', + workspaceCwd: WS_DIFFERENT, + primary: false, + bridge: secondaryBridge, + }), + ]); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { workspaceRegistry: registry }, + ); + + const res = await request(app) + .get('/session/s-secondary/context') + .set('Host', `127.0.0.1:${baseOpts.port}`); + + expect(res.status).toBe(500); + expect(res.body.error).toContain('summary exploded'); + expect(secondaryBridge.sessionContextCalls).toEqual([]); + }); + it('returns session context-usage from the bridge', async () => { const usage: ServeSessionContextUsageStatus = { v: 1, @@ -6040,6 +6161,27 @@ describe('createServeApp', () => { } }); + it('400s unknown explicit cwd before touching the bridge', async () => { + const missingCwd = path.join( + os.tmpdir(), + `qwen-missing-workspace-${Date.now()}`, + ); + 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-unknown/${action}`) + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: missingCwd }); + + expect(res.status).toBe(400); + expect(res.body.code).toBe('workspace_mismatch'); + expect(res.body.requestedWorkspace).toBe(missingCwd); + 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`, @@ -6148,6 +6290,231 @@ describe('createServeApp', () => { expect(bridge.loadCalls).toHaveLength(0); }); + it('loads an explicit trusted non-primary cwd through that workspace runtime', async () => { + const primaryBridge = fakeBridge(); + const secondaryBridge = fakeBridge({ + loadImpl: async (req) => ({ + sessionId: req.sessionId, + workspaceCwd: req.workspaceCwd, + attached: false, + clientId: 'client-secondary-load', + state: { workspace: 'secondary' }, + hasActivePrompt: false, + }), + }); + const registry = createWorkspaceRegistry([ + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-primary', + workspaceCwd: WS_BOUND, + primary: true, + bridge: primaryBridge, + }), + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-secondary', + workspaceCwd: WS_DIFFERENT, + primary: false, + bridge: secondaryBridge, + }), + ]); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { workspaceRegistry: registry }, + ); + + const res = await request(app) + .post('/session/persisted-secondary/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: WS_DIFFERENT }); + + expect(res.status).toBe(200); + expect(res.body).toMatchObject({ + sessionId: 'persisted-secondary', + workspaceCwd: WS_DIFFERENT, + clientId: 'client-secondary-load', + state: { workspace: 'secondary' }, + }); + expect(primaryBridge.loadCalls).toHaveLength(0); + expect(secondaryBridge.loadCalls).toEqual([ + { + sessionId: 'persisted-secondary', + workspaceCwd: WS_DIFFERENT, + historyReplay: 'response', + }, + ]); + }); + + it('rejects restore when the session id is already live in another runtime', async () => { + const primaryBridge = fakeBridge({ + summaryImpl: (sessionId) => ({ + sessionId, + workspaceCwd: WS_BOUND, + createdAt: '2026-05-17T12:00:00.000Z', + clientCount: 1, + hasActivePrompt: false, + }), + }); + const secondaryBridge = fakeBridge(); + const registry = createWorkspaceRegistry([ + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-primary', + workspaceCwd: WS_BOUND, + primary: true, + bridge: primaryBridge, + }), + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-secondary', + workspaceCwd: WS_DIFFERENT, + primary: false, + bridge: secondaryBridge, + }), + ]); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { workspaceRegistry: registry }, + ); + + const res = await request(app) + .post('/session/live-primary/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: WS_DIFFERENT }); + + expect(res.status).toBe(409); + expect(res.body).toMatchObject({ + code: 'session_workspace_conflict', + sessionId: 'live-primary', + workspaceCwd: WS_DIFFERENT, + liveWorkspaceCwd: WS_BOUND, + }); + expect(secondaryBridge.loadCalls).toHaveLength(0); + }); + + it('rejects concurrent restore into a different runtime before either bridge owns the session', async () => { + const secondaryStarted = deferred(); + const releaseSecondary = deferred(); + const primaryBridge = fakeBridge(); + const secondaryBridge = fakeBridge({ + loadImpl: async (req) => { + secondaryStarted.resolve(undefined); + await releaseSecondary.promise; + return { + sessionId: req.sessionId, + workspaceCwd: req.workspaceCwd, + attached: false, + clientId: 'client-secondary-load', + state: { workspace: 'secondary' }, + hasActivePrompt: false, + }; + }, + }); + const registry = createWorkspaceRegistry([ + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-primary', + workspaceCwd: WS_BOUND, + primary: true, + bridge: primaryBridge, + }), + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-secondary', + workspaceCwd: WS_DIFFERENT, + primary: false, + bridge: secondaryBridge, + }), + ]); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { workspaceRegistry: registry }, + ); + + const secondaryRequest = request(app) + .post('/session/concurrent-restore/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: WS_DIFFERENT }) + .then((res) => res); + await secondaryStarted.promise; + + const primaryRes = await (async () => { + try { + return await request(app) + .post('/session/concurrent-restore/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: WS_BOUND }); + } finally { + releaseSecondary.resolve(undefined); + } + })(); + + expect(primaryRes.status).toBe(409); + expect(primaryRes.body).toMatchObject({ + code: 'session_workspace_conflict', + sessionId: 'concurrent-restore', + workspaceCwd: WS_BOUND, + liveWorkspaceCwd: WS_DIFFERENT, + }); + expect(primaryBridge.loadCalls).toHaveLength(0); + + const secondaryRes = await secondaryRequest; + + expect(secondaryRes.status).toBe(200); + expect(secondaryRes.body).toMatchObject({ + sessionId: 'concurrent-restore', + workspaceCwd: WS_DIFFERENT, + }); + expect(secondaryBridge.loadCalls).toEqual([ + { + sessionId: 'concurrent-restore', + workspaceCwd: WS_DIFFERENT, + historyReplay: 'response', + }, + ]); + }); + + it('surfaces live owner scan failures before restoring into a workspace runtime', async () => { + const root = await fsp.mkdtemp(path.join(os.tmpdir(), 'qws-owner-scan-')); + const primaryDir = path.join(root, 'primary'); + const secondaryDir = path.join(root, 'secondary'); + await fsp.mkdir(primaryDir); + await fsp.mkdir(secondaryDir); + const primaryCwd = realpathSync.native(primaryDir); + const secondaryCwd = realpathSync.native(secondaryDir); + const primaryBridge = fakeBridge({ + summaryImpl: () => { + throw new Error('summary exploded'); + }, + }); + const secondaryBridge = fakeBridge(); + const registry = createWorkspaceRegistry([ + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-primary', + workspaceCwd: primaryCwd, + primary: true, + bridge: primaryBridge, + }), + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-secondary', + workspaceCwd: secondaryCwd, + primary: false, + bridge: secondaryBridge, + }), + ]); + const app = createServeApp( + { ...baseOpts, workspace: primaryCwd }, + undefined, + { workspaceRegistry: registry }, + ); + + const res = await request(app) + .post('/session/persisted-secondary/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ cwd: secondaryCwd }); + + expect(res.status).toBe(500); + expect(res.body.error).toContain('summary exploded'); + expect(secondaryBridge.loadCalls).toHaveLength(0); + }); + it('503 + Retry-After: 5 when the bridge throws SessionLimitExceededError', async () => { const bridge = fakeBridge({ resumeImpl: async () => { @@ -6513,6 +6880,54 @@ describe('createServeApp', () => { expect(bridge.listCalls).toEqual([WS_BOUND]); }); + it('supports plural /workspaces/:workspace/sessions for non-primary live sessions', async () => { + const primaryBridge = fakeBridge(); + const secondaryBridge = fakeBridge({ + listImpl: () => [ + { + sessionId: 's-secondary', + workspaceCwd: WS_DIFFERENT, + createdAt: '2026-05-17T12:00:00.000Z', + clientCount: 1, + hasActivePrompt: false, + }, + ], + }); + const registry = createWorkspaceRegistry([ + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-primary', + workspaceCwd: WS_BOUND, + primary: true, + bridge: primaryBridge, + }), + makeWorkspaceRuntimeForTest({ + workspaceId: 'ws-secondary', + workspaceCwd: WS_DIFFERENT, + primary: false, + bridge: secondaryBridge, + }), + ]); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { workspaceRegistry: registry }, + ); + + const res = await request(app) + .get('/workspaces/ws-secondary/sessions') + .set('Host', `127.0.0.1:${baseOpts.port}`); + + expect(res.status).toBe(200); + expect(res.body.sessions).toEqual([ + expect.objectContaining({ + sessionId: 's-secondary', + workspaceCwd: WS_DIFFERENT, + }), + ]); + expect(primaryBridge.listCalls).toEqual([]); + expect(secondaryBridge.listCalls).toEqual([WS_DIFFERENT]); + }); + it('includes persisted sessions from the CLI session store', async () => { const storedOnlyId = '550e8400-e29b-41d4-a716-446655440000'; const liveAndStoredId = '6ba7b810-9dad-11d1-80b4-00c04fd430c8'; @@ -14255,72 +14670,6 @@ describe('createServeApp ServeAppDeps.fsFactory wiring (#4175 PR 18)', () => { expect(locals.workspaceRegistry!.primary.env).toBe(primaryRuntimeEnv); }); - it('wires total admission into the internally-created bridge', async () => { - type AdmissionContext = { - operation: 'spawn'; - workspaceCwd: string; - }; - let freshSessionAdmission: - | ((context: AdmissionContext) => { release(): void }) - | undefined; - vi.resetModules(); - vi.doMock('./acp-session-bridge.js', async () => { - const actual = await vi.importActual< - typeof import('./acp-session-bridge.js') - >('./acp-session-bridge.js'); - return { - ...actual, - createAcpSessionBridge: vi.fn((opts: unknown) => { - freshSessionAdmission = ( - opts as { - freshSessionAdmission?: typeof freshSessionAdmission; - } - ).freshSessionAdmission; - return fakeBridge(); - }), - }; - }); - - try { - const { createServeApp } = await import('./server.js'); - createServeApp( - { - port: 0, - hostname: '127.0.0.1', - workspace: WS_BOUND, - maxTotalSessions: 1, - } as Parameters[0], - () => 0, - ); - - expect(freshSessionAdmission).toBeDefined(); - const first = freshSessionAdmission!({ - operation: 'spawn', - workspaceCwd: WS_BOUND, - }); - let rejection: unknown; - try { - freshSessionAdmission!({ - operation: 'spawn', - workspaceCwd: WS_BOUND, - }); - } catch (err) { - rejection = err; - } - expect(rejection).toMatchObject({ - name: 'TotalSessionLimitExceededError', - limit: 1, - scope: 'total', - operation: 'spawn', - workspaceCwd: WS_BOUND, - }); - first.release(); - } finally { - vi.doUnmock('./acp-session-bridge.js'); - vi.resetModules(); - } - }); - it('uses an injected workspace registry as the primary runtime source', async () => { const { createServeApp } = await import('./server.js'); const runtime = makeInjectedWorkspaceRuntime(); diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 252828b8ba..90393a1a87 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -125,6 +125,7 @@ import { SessionArchiveCoordinator } from './server/session-archive.js'; import { installSelfOriginStripMiddleware } from './server/self-origin.js'; import { createSingleWorkspaceRegistry, + createWorkspaceSessionOwnerIndex, type WorkspaceRegistry, type WorkspaceRuntimeEnvMetadata, } from './workspace-registry.js'; @@ -520,6 +521,9 @@ export function createServeApp( defaultBridgeForAdmission ? [defaultBridgeForAdmission] : [], }) : undefined; + const defaultSessionOwnerIndex = !injectedWorkspaceRegistry + ? createWorkspaceSessionOwnerIndex() + : undefined; const bridge = injectedWorkspaceRegistry?.primary.bridge ?? deps.bridge ?? @@ -528,6 +532,12 @@ export function createServeApp( ...(totalSessionAdmission ? { freshSessionAdmission: totalSessionAdmission.admit } : {}), + ...(defaultSessionOwnerIndex + ? { + sessionLifecycle: + defaultSessionOwnerIndex.handleBridgeSessionLifecycle, + } + : {}), maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession, eventRingSize: opts.eventRingSize, compactedReplayMaxBytes: opts.compactedReplayMaxBytes, @@ -619,20 +629,25 @@ export function createServeApp( }); const workspaceRegistry = injectedWorkspaceRegistry ?? - createSingleWorkspaceRegistry({ - workspaceId: hashDaemonWorkspace(boundWorkspace), - workspaceCwd: boundWorkspace, - primary: true, - trusted: deps.primaryWorkspaceTrusted ?? false, - env: primaryRuntimeEnvMetadata ?? { - mode: 'parent-process', - overlayKeys: [], + createSingleWorkspaceRegistry( + { + workspaceId: hashDaemonWorkspace(boundWorkspace), + workspaceCwd: boundWorkspace, + primary: true, + trusted: deps.primaryWorkspaceTrusted ?? false, + env: primaryRuntimeEnvMetadata ?? { + mode: 'parent-process', + overlayKeys: [], + }, + bridge, + workspaceService: workspace, + routeFileSystemFactory: fsFactory, + clientMcpSenderRegistry, }, - bridge, - workspaceService: workspace, - routeFileSystemFactory: fsFactory, - clientMcpSenderRegistry, - }); + defaultSessionOwnerIndex + ? { sessionOwnerIndex: defaultSessionOwnerIndex } + : {}, + ); (app.locals as { workspaceRegistry?: WorkspaceRegistry }).workspaceRegistry = workspaceRegistry; const primaryRuntime = workspaceRegistry.primary; diff --git a/packages/cli/src/serve/server/telemetry.test.ts b/packages/cli/src/serve/server/telemetry.test.ts index cf35e5d65a..dd48aaaef5 100644 --- a/packages/cli/src/serve/server/telemetry.test.ts +++ b/packages/cli/src/serve/server/telemetry.test.ts @@ -96,6 +96,28 @@ describe('daemonTelemetryMiddleware — recordRequest seam', () => { expect(recordRequest).not.toHaveBeenCalled(); }); + it('maps plural workspace session listing to the existing route label', () => { + const recordRequest = vi.fn(); + const mw = daemonTelemetryMiddleware(() => '/ws', recordRequest); + const res = mockRes(200); + + mw( + mockReq('GET', '/workspaces/ws-secondary/sessions'), + res, + vi.fn() as unknown as NextFunction, + ); + res.emit('finish'); + + expect(recordRequest).toHaveBeenCalledTimes(1); + expect(coreMocks.withDaemonRequestSpan).toHaveBeenCalledWith( + expect.objectContaining({ + method: 'GET', + route: 'GET /workspace/:id/sessions', + }), + expect.any(Function), + ); + }); + it('excludes the dashboard status poll (GET /daemon/status) from recordRequest', () => { const recordRequest = vi.fn(); const mw = daemonTelemetryMiddleware(() => '/ws', recordRequest); diff --git a/packages/cli/src/serve/server/telemetry.ts b/packages/cli/src/serve/server/telemetry.ts index 276e9b28f2..d400474a2b 100644 --- a/packages/cli/src/serve/server/telemetry.ts +++ b/packages/cli/src/serve/server/telemetry.ts @@ -114,6 +114,9 @@ export function resolveDaemonTelemetryRoute( if (req.method === 'GET' && /^\/workspace\/[^/]+\/sessions$/.test(path)) { return { route: 'GET /workspace/:id/sessions' }; } + if (req.method === 'GET' && /^\/workspaces\/[^/]+\/sessions$/.test(path)) { + return { route: 'GET /workspace/:id/sessions' }; + } if (req.method === 'POST' && path === '/workspace/init') { return { route: 'POST /workspace/init' }; } diff --git a/packages/cli/src/serve/workspace-registry.test.ts b/packages/cli/src/serve/workspace-registry.test.ts index b45bc6bf78..0b054f6aea 100644 --- a/packages/cli/src/serve/workspace-registry.test.ts +++ b/packages/cli/src/serve/workspace-registry.test.ts @@ -7,6 +7,7 @@ import { describe, expect, it } from 'vitest'; import { SessionNotFoundError } from './acp-session-bridge.js'; import { + createWorkspaceSessionOwnerIndex, createWorkspaceRegistry, createSingleWorkspaceRegistry, type WorkspaceRuntime, @@ -162,6 +163,77 @@ describe('createWorkspaceRegistry', () => { }); }); + it('uses the session owner index before scanning runtime bridges', () => { + const primarySummary = vi.fn(() => { + throw new SessionNotFoundError('sess-secondary'); + }); + const secondarySummary = vi.fn((sessionId: string) => ({ + sessionId, + workspaceCwd: '/work/secondary', + })); + const primary = makeRuntime('/work/primary', { + workspaceId: 'ws-primary', + primary: true, + bridge: bridgeWithSummary(primarySummary), + }); + const secondary = makeRuntime('/work/secondary', { + workspaceId: 'ws-secondary', + bridge: bridgeWithSummary(secondarySummary), + }); + const sessionOwnerIndex = createWorkspaceSessionOwnerIndex(); + sessionOwnerIndex.register('sess-secondary', '/work/secondary'); + + const registry = createWorkspaceRegistry([primary, secondary], { + sessionOwnerIndex, + }); + + expect(registry.resolveLiveSessionOwner('sess-secondary')).toEqual({ + kind: 'found', + runtime: secondary, + }); + expect(primarySummary).not.toHaveBeenCalled(); + expect(secondarySummary).toHaveBeenCalledWith('sess-secondary'); + }); + + it('drops stale indexed owners and caches the fallback scan result', () => { + const primarySummary = vi.fn((sessionId: string) => ({ + sessionId, + workspaceCwd: '/work/primary', + })); + const secondarySummary = vi.fn(() => { + throw new SessionNotFoundError('stale'); + }); + const primary = makeRuntime('/work/primary', { + workspaceId: 'ws-primary', + primary: true, + bridge: bridgeWithSummary(primarySummary), + }); + const secondary = makeRuntime('/work/secondary', { + workspaceId: 'ws-secondary', + bridge: bridgeWithSummary(secondarySummary), + }); + const sessionOwnerIndex = createWorkspaceSessionOwnerIndex(); + sessionOwnerIndex.register('stale', '/work/secondary'); + + const registry = createWorkspaceRegistry([primary, secondary], { + sessionOwnerIndex, + }); + + expect(registry.resolveLiveSessionOwner('stale')).toEqual({ + kind: 'found', + runtime: primary, + }); + expect(primarySummary).toHaveBeenCalledTimes(1); + expect(secondarySummary).toHaveBeenCalledTimes(2); + + expect(registry.resolveLiveSessionOwner('stale')).toEqual({ + kind: 'found', + runtime: primary, + }); + expect(primarySummary).toHaveBeenCalledTimes(2); + expect(secondarySummary).toHaveBeenCalledTimes(2); + }); + it('fails closed when live session owner resolution is ambiguous', () => { const first = makeRuntime('/work/primary', { workspaceId: 'ws-primary', @@ -201,4 +273,35 @@ describe('createWorkspaceRegistry', () => { expect(() => registry.resolveLiveSessionOwner('sess')).toThrow(lookupError); }); + + it('does not cache partial scan results when live owner scan fails', () => { + const lookupError = new Error('bridge unavailable'); + const primarySummary = vi.fn((sessionId: string) => ({ + sessionId, + workspaceCwd: '/work/primary', + })); + const secondarySummary = vi.fn(() => { + throw lookupError; + }); + const primary = makeRuntime('/work/primary', { + workspaceId: 'ws-primary', + primary: true, + bridge: bridgeWithSummary(primarySummary), + }); + const secondary = makeRuntime('/work/secondary', { + workspaceId: 'ws-secondary', + bridge: bridgeWithSummary(secondarySummary), + }); + const sessionOwnerIndex = createWorkspaceSessionOwnerIndex(); + const registry = createWorkspaceRegistry([primary, secondary], { + sessionOwnerIndex, + }); + + expect(() => registry.resolveLiveSessionOwner('sess')).toThrow(lookupError); + expect(sessionOwnerIndex.getWorkspaceCwds('sess')).toEqual([]); + + expect(() => registry.resolveLiveSessionOwner('sess')).toThrow(lookupError); + expect(primarySummary).toHaveBeenCalledTimes(2); + expect(secondarySummary).toHaveBeenCalledTimes(2); + }); }); diff --git a/packages/cli/src/serve/workspace-registry.ts b/packages/cli/src/serve/workspace-registry.ts index 177827787d..9c580be459 100644 --- a/packages/cli/src/serve/workspace-registry.ts +++ b/packages/cli/src/serve/workspace-registry.ts @@ -45,6 +45,25 @@ export type WorkspaceSessionOwnerResolution = readonly runtimes: readonly WorkspaceRuntime[]; }; +export type WorkspaceSessionLifecycleEvent = + | { + readonly type: 'registered'; + readonly sessionId: string; + readonly workspaceCwd: string; + } + | { + readonly type: 'removed'; + readonly sessionId: string; + readonly workspaceCwd: string; + }; + +export interface WorkspaceSessionOwnerIndex { + register(sessionId: string, workspaceCwd: string): void; + remove(sessionId: string, workspaceCwd?: string): void; + getWorkspaceCwds(sessionId: string): readonly string[]; + handleBridgeSessionLifecycle(event: WorkspaceSessionLifecycleEvent): void; +} + export interface WorkspaceRegistry { readonly primary: WorkspaceRuntime; list(): readonly WorkspaceRuntime[]; @@ -56,8 +75,52 @@ export interface WorkspaceRegistry { resolveLiveSessionOwner(sessionId: string): WorkspaceSessionOwnerResolution; } +export interface WorkspaceRegistryOptions { + readonly sessionOwnerIndex?: WorkspaceSessionOwnerIndex; +} + +export function createWorkspaceSessionOwnerIndex(): WorkspaceSessionOwnerIndex { + const bySessionId = new Map>(); + + const register = (sessionId: string, workspaceCwd: string): void => { + let owners = bySessionId.get(sessionId); + if (!owners) { + owners = new Set(); + bySessionId.set(sessionId, owners); + } + owners.add(workspaceCwd); + }; + + const remove = (sessionId: string, workspaceCwd?: string): void => { + if (workspaceCwd === undefined) { + bySessionId.delete(sessionId); + return; + } + const owners = bySessionId.get(sessionId); + if (!owners) return; + owners.delete(workspaceCwd); + if (owners.size === 0) { + bySessionId.delete(sessionId); + } + }; + + return { + register, + remove, + getWorkspaceCwds: (sessionId) => [...(bySessionId.get(sessionId) ?? [])], + handleBridgeSessionLifecycle: (event) => { + if (event.type === 'registered') { + register(event.sessionId, event.workspaceCwd); + } else { + remove(event.sessionId, event.workspaceCwd); + } + }, + }; +} + export function createWorkspaceRegistry( inputRuntimes: readonly WorkspaceRuntime[], + options: WorkspaceRegistryOptions = {}, ): WorkspaceRegistry { if (inputRuntimes.length === 0) { throw new Error( @@ -96,6 +159,30 @@ export function createWorkspaceRegistry( const runtimes = Object.freeze([...inputRuntimes]); const primary = primaryRuntimes[0]!; + const sessionOwnerIndex = options.sessionOwnerIndex; + const scanLiveOwners = ( + sessionId: string, + ): WorkspaceSessionOwnerResolution => { + const matches: WorkspaceRuntime[] = []; + for (const runtime of runtimes) { + try { + runtime.bridge.getSessionSummary(sessionId); + matches.push(runtime); + } catch (err) { + if (err instanceof SessionNotFoundError) continue; + throw err; + } + } + for (const match of matches) { + sessionOwnerIndex?.register(sessionId, match.workspaceCwd); + } + if (matches.length === 0) return { kind: 'not_found' }; + if (matches.length === 1) { + return { kind: 'found', runtime: matches[0]! }; + } + return { kind: 'ambiguous', runtimes: matches }; + }; + return { primary, list: () => runtimes, @@ -104,27 +191,41 @@ export function createWorkspaceRegistry( resolveWorkspaceCwd: (workspaceCwd) => workspaceCwd === undefined ? primary : byCwd.get(workspaceCwd), resolveLiveSessionOwner: (sessionId) => { - const matches: WorkspaceRuntime[] = []; - for (const runtime of runtimes) { - try { - runtime.bridge.getSessionSummary(sessionId); - matches.push(runtime); - } catch (err) { - if (err instanceof SessionNotFoundError) continue; - throw err; + const indexedCwds = sessionOwnerIndex?.getWorkspaceCwds(sessionId) ?? []; + if (indexedCwds.length > 0) { + const matches: WorkspaceRuntime[] = []; + for (const workspaceCwd of indexedCwds) { + const runtime = byCwd.get(workspaceCwd); + if (!runtime) { + sessionOwnerIndex?.remove(sessionId, workspaceCwd); + continue; + } + try { + runtime.bridge.getSessionSummary(sessionId); + matches.push(runtime); + } catch (err) { + if (err instanceof SessionNotFoundError) { + sessionOwnerIndex?.remove(sessionId, workspaceCwd); + continue; + } + throw err; + } + } + if (matches.length === 1) { + return { kind: 'found', runtime: matches[0]! }; + } + if (matches.length > 1) { + return { kind: 'ambiguous', runtimes: matches }; } } - if (matches.length === 0) return { kind: 'not_found' }; - if (matches.length === 1) { - return { kind: 'found', runtime: matches[0]! }; - } - return { kind: 'ambiguous', runtimes: matches }; + return scanLiveOwners(sessionId); }, }; } export function createSingleWorkspaceRegistry( runtime: WorkspaceRuntime, + options: WorkspaceRegistryOptions = {}, ): WorkspaceRegistry { - return createWorkspaceRegistry([runtime]); + return createWorkspaceRegistry([runtime], options); }