vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-07-10 01:29:17 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 12

feat(cli): Add session owner index for workspace runtimes

Browse source 
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 <qwen-coder@alibabacloud.com>
This commit is contained in:
doudouOUC 2026-07-08 20:32:44 +08:00
parent 955ad27fc7
commit a7f29a7cde
15 changed files with 1445 additions and 337 deletions
Download patch file Download diff file Expand all files Collapse all files

84
docs/design/daemon-multi-workspace-phase2a-sessions.md
View file

@ -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

37
docs/developers/qwen-serve-protocol.md
View file

@ -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 \"<sid>\" is already live or restoring in another workspace runtime.",
"code": "session_workspace_conflict",
"sessionId": "<sid>",
"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,
@ -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. `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/<workspace-id>/sessions
```
Query parameters:

75
packages/acp-bridge/src/bridge.test.ts
View file

@ -77,6 +77,81 @@ function deferred<T>(): {
}
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

67
packages/acp-bridge/src/bridge.ts
View file

@ -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',

25
packages/acp-bridge/src/bridgeOptions.ts
View file

@ -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

2
packages/cli/src/serve/acp-session-bridge.ts
View file

@ -58,6 +58,8 @@ export type {
BridgeFreshSessionAdmission,
BridgeFreshSessionAdmissionContext,
BridgeFreshSessionReservation,
BridgeSessionLifecycle,
BridgeSessionLifecycleEvent,
BridgeOptions,
DaemonStatusProvider,
} from '@qwen-code/acp-bridge/bridgeOptions';

56
packages/cli/src/serve/multi-workspace-sessions.test.ts
View file

@ -630,7 +630,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 +639,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 () => {

23
packages/cli/src/serve/rate-limit.test.ts
View file

@ -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++) {

608
packages/cli/src/serve/routes/session.ts
View file

@ -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<string, unknown>,
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 (workspaceRegistry.list().length > 1 && '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 = (
@ -332,6 +462,28 @@ export function registerSessionRoutes(
}
};
const withOwnerReadSession =
(
route: string,
handler: (
req: Request,
res: Response,
sessionId: string,
runtime: WorkspaceRuntime,
) => Promise<void> | void,
): RequestHandler =>
async (req, res) => {
const sessionId = requireSessionId(req, res);
if (sessionId === null) return;
const runtime = resolveLiveSessionRuntime(sessionId, res, route);
if (!runtime) return;
try {
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,111 @@ 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) => {
// 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;
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',

9
packages/cli/src/serve/run-qwen-serve.ts
View file

@ -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: () =>

130
packages/cli/src/serve/server-default-bridge-wiring.test.ts Normal file
View file

@ -0,0 +1,130 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { afterEach, describe, expect, it, vi } from 'vitest';
import type {
AcpSessionBridge,
BridgeFreshSessionAdmission,
BridgeOptions,
BridgeSessionSummary,
} from './acp-session-bridge.js';
import type { WorkspaceRegistry } from './workspace-registry.js';
const WS_BOUND = '/work/bound';
function makeBridge(sessionCount = 0): AcpSessionBridge {
const getSessionSummary = (sessionId: string): BridgeSessionSummary => ({
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 bridge = makeBridge();
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<typeof createServeApp>[0],
() => 0,
);
const locals = app.locals as { workspaceRegistry?: WorkspaceRegistry };
expect(sessionLifecycle).toBeDefined();
sessionLifecycle!({
type: 'registered',
sessionId: 'session-indexed',
workspaceCwd: WS_BOUND,
reason: 'spawn',
});
expect(
locals.workspaceRegistry!.resolveLiveSessionOwner('session-indexed'),
).toEqual({
kind: 'found',
runtime: locals.workspaceRegistry!.primary,
});
});
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<typeof createServeApp>[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,
});
});
});

424
packages/cli/src/serve/server.test.ts
View file

@ -184,6 +184,20 @@ function restoreEnv(key: string, value: string | undefined): void {
}
}
function deferred<T = void>(): {
promise: Promise<T>;
resolve: (value: T | PromiseLike<T>) => void;
reject: (reason?: unknown) => void;
} {
let resolve!: (value: T | PromiseLike<T>) => void;
let reject!: (reason?: unknown) => void;
const promise = new Promise<T>((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,57 @@ 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('returns session context-usage from the bridge', async () => {
const usage: ServeSessionContextUsageStatus = {
v: 1,
@ -6148,6 +6233,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 +6823,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 +14613,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<typeof createServeApp>[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();

41
packages/cli/src/serve/server.ts
View file

@ -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;

72
packages/cli/src/serve/workspace-registry.test.ts
View file

@ -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',

129
packages/cli/src/serve/workspace-registry.ts
View file

@ -45,6 +45,27 @@ export type WorkspaceSessionOwnerResolution =
readonly runtimes: readonly WorkspaceRuntime[];
};
export type WorkspaceSessionLifecycleEvent =
| {
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 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 +77,52 @@ export interface WorkspaceRegistry {
resolveLiveSessionOwner(sessionId: string): WorkspaceSessionOwnerResolution;
}
export interface WorkspaceRegistryOptions {
readonly sessionOwnerIndex?: WorkspaceSessionOwnerIndex;
}
export function createWorkspaceSessionOwnerIndex(): WorkspaceSessionOwnerIndex {
const bySessionId = new Map<string, Set<string>>();
const register = (sessionId: string, workspaceCwd: string): void => {
let owners = bySessionId.get(sessionId);
if (!owners) {
owners = new Set<string>();
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 +161,28 @@ 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);
sessionOwnerIndex?.register(sessionId, runtime.workspaceCwd);
} catch (err) {
if (err instanceof SessionNotFoundError) continue;
throw err;
}
}
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);
}