QwenLM/qwen-code
1
0 Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-11-07 06:19:43 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

feat(serve): support multi-workspace rewind and shell (#6826)

Browse Source 
* feat(serve): support multi-workspace rewind and shell

Co-authored-by: Qwen-Coder <[email protected]>

* codex: address PR review feedback (#6826)

Co-authored-by: Qwen-Coder <[email protected]>

* codex: address PR review feedback (#6826)

Co-authored-by: Qwen-Coder <[email protected]>

---------

Co-authored-by: Qwen-Coder <[email protected]>
This commit is contained in:
jinyeandQwen-Coder authored and GitHub committed 2026-07-13 15:32:44 +00:00
1 parent 71c7e448f0
commit b933b90172
27 files changed
+1211 -71
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

AGENTS.md
+3
View File
@@ -181,6 +181,9 @@ npm run preflight # Full check: clean → install → format → lint → build
pass suffices for a trivial change.
5. **Code review** — run `/review` when available. Triage each comment:
valid / false positive / overthinking. Fixes go back through steps 3-4.
Here, `/review` means the Codex code-review workflow, not Qwen Review or
the `qwen-review` plugin. Do not invoke Qwen Review unless the user
explicitly requests it by name.
### Feature development
docs/design/daemon-multi-workspace-phase2a-sessions.md
+5
View File
@@ -1,5 +1,10 @@
# Phase 2a Multi-Workspace Sessions Foundation
> **Historical status:** The live-session rewind snapshots, rewind, and shell
> limitations recorded in this document are superseded by
> [`daemon-multi-workspace-session-file-ops.md`](./daemon-multi-workspace-session-file-ops.md).
> The remaining Phase 2a scope statements are unchanged.
## Summary
This document records the multi-workspace sessions contract for issue #6378
docs/design/daemon-multi-workspace-session-file-ops.md
+89
View File
@@ -0,0 +1,89 @@
# Daemon Multi-Workspace Session Rewind and Shell
## Status
Final implementation design. This document supersedes the Phase 2a
primary-only statement for live-session rewind snapshots, rewind, and shell.
## Problem
The daemon exposes singular session APIs, while a multi-workspace daemon owns
one bridge per workspace runtime. Most live session routes already resolve the
session owner, but rewind snapshots, rewind, and shell were still bound to the
primary bridge or rejected a secondary owner. That made a valid live secondary
session indistinguishable from an unsupported route to clients.
## Decision
Keep the singular REST API and resolve the owning live runtime on every request:
- `GET /session/:id/rewind/snapshots` uses owner-aware read routing.
- `POST /session/:id/rewind` and `POST /session/:id/shell` use owner-aware
mutable routing and the shared session archive coordinator.
- SDK rewind calls always select direct REST, even when the client is configured
with ACP transport. This preserves the strict REST mutation gate.
- SDK shell keeps its configured transport. The default REST transport gains
owner routing; a workspace-qualified ACP client keeps `_qwen/session/shell`.
- No workspace-qualified session REST API, ACP rewind method, core change, ACP
child change, or FileHistory migration is introduced.
## Ownership and authorization
The workspace registry searches all live bridge summaries for the session id.
Exactly one trusted owner dispatches to that runtime. No owner returns
`404 session_not_found`; an untrusted owner returns `403 untrusted_workspace`;
multiple owners return `500 ambiguous_session_owner`. All three outcomes occur
before the target bridge operation runs. Persisted sessions must first be loaded
or resumed into a runtime.
Rewind and shell retain `mutate({ strict: true })`. Shell additionally requires
effective shell enablement, a valid session-bound client id, and a non-empty
command. Rewind forwards an optional client id and accepts `rewindFiles` only
when omitted or boolean. Omitted means `true`; any other JSON type returns
`400 invalid_rewind_files_flag`.
## Behavior boundaries
Shell starts in the owning session workspace cwd and is not a filesystem path
sandbox. Rewind restores only snapshots recorded for `edit` and `write_file`.
It does not undo shell, Git, script, or manual changes. File restore is
best-effort: the conversation may already be rewound when the response reports
`rewound: false` with `filesFailed[]`. Active prompts retain `409 session_busy`
and `Retry-After: 5`; invalid targets retain `400 invalid_rewind_target`.
Web Shell continues to request `rewindFiles: false`.
The existing `~/.qwen/file-history/<sessionId>` layout is unchanged. A live UUID
collision therefore fails closed through owner ambiguity rather than selecting
the primary runtime.
## Capabilities
`multi_workspace_session_rewind` is advertised only while more than one runtime
exists. `multi_workspace_session_shell` additionally requires effective session
shell enablement, which means both the enable flag and a configured token.
Client preflight is additive:
- Primary rewind: `session_rewind`.
- Secondary rewind: `session_rewind` and
`multi_workspace_session_rewind`.
- Primary shell: `session_shell_command`.
- Secondary shell: `session_shell_command` and
`multi_workspace_session_shell`.
ACP-native clients use initialize `_qwen.methods`; the daemon does not advertise
an ACP rewind vendor method.
## Verification
Unit coverage pins owner dispatch, zero calls to non-owning bridges, trust and
ambiguity failures, strict validation order, `rewindFiles` semantics, SDK REST
fallback, unchanged shell transport, conditional capability advertising, and
the absence of ACP rewind mappings. ACP workspace tests retain the invariant
that an A connection cannot operate a B session while a workspace-qualified B
shell succeeds.
The E2E scenario creates a session and tracked edits in workspace B, verifies
snapshots and shell cwd are B-scoped, checks both rewind file modes, proves a
shell-created file survives rewind, and records busy, partial restore, and
untrusted-secondary outcomes.
docs/developers/daemon/02-serve-runtime.md
+1 -1
View File
@@ -20,7 +20,7 @@
**App factory**: `createServeApp(opts, getPort, deps)` in `packages/cli/src/serve/server.ts`. Builds the Express `Application`. Direct embedders and tests call it without the bootstrap wrapper.
**Capability registry**: `SERVE_CAPABILITY_REGISTRY` in `packages/cli/src/serve/capabilities.ts`. Each tag has a `since` version and optional `modes`. Ten conditional tags (`require_auth`, `mcp_workspace_pool`, `mcp_pool_restart`, `allow_origin`, `prompt_absolute_deadline`, `writer_idle_timeout`, `workspace_settings`, `session_shell_command`, `rate_limit`, `workspace_reload`) are omitted when their corresponding toggle is off. See [`11-capabilities-versioning.md`](./11-capabilities-versioning.md).
**Capability registry**: `SERVE_CAPABILITY_REGISTRY` in `packages/cli/src/serve/capabilities.ts`. Each tag has a `since` version and optional `modes`. Conditional tags are omitted when their deployment or runtime predicate is false; the registry and predicate map are the source of truth. See [`11-capabilities-versioning.md`](./11-capabilities-versioning.md).
**Middleware** (`packages/cli/src/serve/auth.ts` and `server.ts`):
docs/developers/daemon/08-session-lifecycle.md
+17 -1
View File
@@ -194,7 +194,23 @@ cache path for a single-turn, no-tool LLM call and returns
without routing through the LLM. It streams output on the session SSE bus via
`user_shell_command` / `user_shell_result` events and injects the command plus
result into the LLM conversation history. The response is
`{ exitCode, output, aborted }`.
`{ exitCode, output, aborted }`. For a live secondary-workspace session, the
singular REST route resolves the session owner and executes on that runtime's
bridge, so the command starts in the owning workspace cwd. The route does not
provide a path sandbox. Workspace-qualified ACP clients may continue to use
`_qwen/session/shell` on the owning workspace connection.
### Session Rewind
`GET /session/:id/rewind/snapshots` and `POST /session/:id/rewind` resolve the
owning live workspace runtime. Persisted sessions must be loaded or resumed
before rewind. Rewind truncates conversation history and optionally restores
files tracked by `edit` and `write_file`; it does not undo shell commands, Git,
scripts, or manual changes. File restoration is best-effort, so a response may
report `rewound: false` and `filesFailed[]` after the conversation history has
already moved. SDK rewind calls always use owner-aware REST, including when the
client otherwise uses ACP transport, because the mutation must retain strict
REST authentication.
### Session Detach
docs/developers/daemon/11-capabilities-versioning.md
+36 -14
View File
@@ -81,6 +81,16 @@ export const CONDITIONAL_SERVE_FEATURES: ReadonlyMap<
(t) => t.voiceTranscriptionAvailable === true,
],
['session_shell_command', (t) => t.sessionShellCommandEnabled === true],
[
'multi_workspace_session_rewind',
(t) => t.multiWorkspaceSessionsEnabled === true,
],
[
'multi_workspace_session_shell',
(t) =>
t.multiWorkspaceSessionsEnabled === true &&
t.sessionShellCommandEnabled === true,
],
['rate_limit', (t) => t.rateLimit === true],
['workspace_reload', (t) => t.reloadAvailable === true],
['voice_transcribe', (t) => t.voiceWsAvailable !== false],
@@ -94,7 +104,7 @@ The `Map` stores membership and predicate together. Adding a new conditional tag
Baseline tags are not present in the `Map` and are advertised unconditionally. This is intentionally represented by absence rather than by a separate Set.
### 75 tags (v1, grouped by domain)
### v1 tags grouped by domain
Foundation: `health`, `daemon_status`, `capabilities`.
@@ -120,6 +130,15 @@ Voice: **`workspace_voice`** (conditional), **`workspace_voice_transcription`**
Rate limiting: **`rate_limit`** (conditional).
Multi-workspace session routing: **`multi_workspace_sessions`** (conditional),
**`multi_workspace_session_rewind`** (conditional), and
**`multi_workspace_session_shell`** (conditional). A client may use rewind for
a primary session with `session_rewind`; a secondary live session additionally
requires `multi_workspace_session_rewind`. Shell uses the equivalent
`session_shell_command` plus `multi_workspace_session_shell` pairing for a
secondary session. ACP-native clients continue to use the `_qwen.methods`
returned by initialize; no ACP rewind vendor method is advertised.
Bold tags have `modes` or are conditional.
## Flow
@@ -167,23 +186,26 @@ sequenceDiagram
## Dependencies
- Read by `packages/cli/src/serve/server.ts` when building `/capabilities` responses.
- Toggle input comes from `runQwenServe` / `createServeApp`: `{ requireAuth, mcpPoolActive, allowOriginActive, promptDeadlineMs, writerIdleTimeoutMs, persistSettingAvailable, sessionShellCommandEnabled, rateLimit, reloadAvailable }`.
- Toggle input comes from `runQwenServe` / `createServeApp`, including
authentication, MCP, origin, prompt, settings, shell, rate-limit, reload, and
live workspace-runtime-count state.
- The active `permission` policy in the envelope comes from `BridgeOptions.permissionPolicy`, which itself reads `settings.json` `policy.permissionStrategy`.
## Configuration
| Source | Knob | Effect on capabilities |
| -------------------------- | --------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- |
| CLI flag | `--require-auth` | Advertises `require_auth`. |
| Env | `QWEN_SERVE_NO_MCP_POOL=1` | Stops advertising `mcp_workspace_pool` and `mcp_pool_restart`; MCP events no longer stamp `scope: 'workspace'`. |
| CLI flag | `--mcp-client-budget=N`, `--mcp-budget-mode={off,warn,enforce}` | Does not change the tag set (`mcp_guardrails` is always advertised), but changes per-server reservation and refusal behavior. |
| CLI flag / env | `--rate-limit` / `QWEN_SERVE_RATE_LIMIT=1` | Advertises `rate_limit`. |
| Embedded option | `persistSettingAvailable` | Advertises `workspace_settings` and `workspace_voice`. |
| Embedded option | `voiceTranscriptionAvailable` | Advertises `workspace_voice_transcription`. |
| CLI flag / embedded option | `--enable-session-shell` / `sessionShellCommandEnabled` | Advertises `session_shell_command`. |
| Embedded option | `reloadAvailable` | Advertises `workspace_reload`. |
| Embedded option | `voiceWsAvailable` | Advertises `voice_transcribe`. |
| `settings.json` | `policy.permissionStrategy` | Sets envelope `policy.permission`. |
| Source | Knob | Effect on capabilities |
| -------------------------- | --------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| CLI flag | `--require-auth` | Advertises `require_auth`. |
| Env | `QWEN_SERVE_NO_MCP_POOL=1` | Stops advertising `mcp_workspace_pool` and `mcp_pool_restart`; MCP events no longer stamp `scope: 'workspace'`. |
| CLI flag | `--mcp-client-budget=N`, `--mcp-budget-mode={off,warn,enforce}` | Does not change the tag set (`mcp_guardrails` is always advertised), but changes per-server reservation and refusal behavior. |
| CLI flag / env | `--rate-limit` / `QWEN_SERVE_RATE_LIMIT=1` | Advertises `rate_limit`. |
| Embedded option | `persistSettingAvailable` | Advertises `workspace_settings` and `workspace_voice`. |
| Embedded option | `voiceTranscriptionAvailable` | Advertises `workspace_voice_transcription`. |
| CLI flag / embedded option | `--enable-session-shell` / `sessionShellCommandEnabled` | Advertises `session_shell_command`. |
| Runtime state | More than one registered workspace runtime | Advertises `multi_workspace_sessions` and `multi_workspace_session_rewind`; also advertises `multi_workspace_session_shell` when session shell is effectively enabled. |
| Embedded option | `reloadAvailable` | Advertises `workspace_reload`. |
| Embedded option | `voiceWsAvailable` | Advertises `voice_transcribe`. |
| `settings.json` | `policy.permissionStrategy` | Sets envelope `policy.permission`. |
## Caveats and known limits
docs/developers/daemon/12-auth-security.md
+9 -1
View File
@@ -138,7 +138,15 @@ The `code: 'token_required'` shape is distinct from `bearerAuth`'s plain `Unauth
`/workspace/tools/:name/enable`, `/workspace/mcp/:server/restart`,
`/workspace/mcp/:server/{enable,disable,authenticate,clear-auth}`,
`/workspace/mcp/servers` (POST/DELETE), `/workspace/auth/device-flow`,
`/workspace/init`, `/session/:id/approval-mode`.
`/workspace/init`, `/session/:id/approval-mode`, `/session/:id/rewind`, and
`/session/:id/shell`.
Rewind remains REST-only in the TypeScript SDK even when an ACP transport is
configured. This preserves the strict mutation gate and bearer/client identity
headers; the ACP route table intentionally has no rewind mapping. Owner routing
also rechecks workspace trust before either rewind or shell reaches a secondary
runtime bridge. Duplicate live session ids fail closed as
`ambiguous_session_owner` instead of falling back to the primary runtime.
### `/health` exemption
docs/developers/qwen-serve-protocol.md
+4 -1
View File
@@ -187,7 +187,8 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design
'non_blocking_prompt', 'session_language', 'session_rewind',
'workspace_hooks', 'session_hooks', 'workspace_extensions',
'session_branch', 'rate_limit', 'workspace_reload',
'multi_workspace_sessions', 'persistent_workspace_registration',
'multi_workspace_sessions', 'multi_workspace_session_rewind',
'multi_workspace_session_shell', 'persistent_workspace_registration',
'workspace_qualified_rest_core', 'workspace_persisted_transcript',
'client_mcp_over_ws', 'cdp_tunnel_over_ws', 'browser_automation_mcp']
```
@@ -256,6 +257,8 @@ operator diagnostic snapshot documented below.
| `writer_idle_timeout` | `--writer-idle-timeout-ms` / `QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS` / `ServeOptions.writerIdleTimeoutMs` is set to a positive integer. |
| `workspace_settings` | the daemon was created with settings persistence available. |
| `session_shell_command` | session shell execution is explicitly enabled. |
| `multi_workspace_session_rewind` | more than one workspace runtime is registered; singular live-session rewind routes resolve the owning runtime. |
| `multi_workspace_session_shell` | more than one workspace runtime is registered and session shell execution is explicitly enabled; singular REST shell resolves the owning runtime. |
| `rate_limit` | `--rate-limit` / `QWEN_SERVE_RATE_LIMIT=1` / `ServeOptions.rateLimit` is enabled. |
| `workspace_reload` | workspace reload support is available in the embedded route configuration. |
| `persistent_workspace_registration` | a workspace registration store is wired into the daemon. Production `runQwenServe` supplies the user-level store automatically; direct `createServeApp` embeds must inject one explicitly and own startup restoration of their workspace registry. |
packages/cli/src/serve/acp-http/transport.test.ts
+3
View File
@@ -1169,6 +1169,9 @@ describe('ACP Streamable HTTP transport (over the wire)', () => {
expect(result.agentCapabilities._meta.qwen.methods).toContain(
'_qwen/session/shell',
);
expect(result.agentCapabilities._meta.qwen.methods).not.toContain(
'_qwen/session/rewind',
);
});
it('initialize advertises _qwen/session/lsp', async () => {
packages/cli/src/serve/acp-http/workspace-qualified-acp.test.ts
+89
View File
@@ -38,7 +38,20 @@ const PARENT_ENV: WorkspaceRuntimeEnvMetadata = {
function makeBridge(): HttpAcpBridge {
return {
spawnOrAttach: vi.fn(async (req: { workspaceCwd: string }) => ({
sessionId:
req.workspaceCwd === '/ws-b' ? 'secondary-session' : 'primary-session',
workspaceCwd: req.workspaceCwd,
attached: false,
clientId:
req.workspaceCwd === '/ws-b' ? 'secondary-client' : 'primary-client',
})),
detachClient: vi.fn(async () => {}),
executeShellCommand: vi.fn(async () => ({
exitCode: 0,
output: 'ok',
aborted: false,
})),
isWorkspaceMemoryRememberAvailable: vi.fn(async () => true),
runWorkspaceMemoryRemember: vi.fn(async () => ({
filesTouched: [],
@@ -172,6 +185,7 @@ describe('workspace-qualified ACP (/workspaces/:workspace/acp)', () => {
cdpTunnelOverWs: true,
cdpTunnelRegistry: cdpRegistry,
checkRate,
sessionShellCommandEnabled: true,
workspaceRememberLane: new WorkspaceRememberTaskLane(primaryBridge),
});
@@ -251,6 +265,44 @@ describe('workspace-qualified ACP (/workspaces/:workspace/acp)', () => {
});
}
async function sendWsRequests(
pathname: string,
requests: Array<Record<string, unknown>>,
): Promise<Array<Record<string, unknown>>> {
return new Promise((resolve, reject) => {
const responses: Array<Record<string, unknown>> = [];
const ws = new WebSocket(`ws://127.0.0.1:${port}${pathname}`, {
handshakeTimeout: 2000,
});
ws.on('open', () => ws.send(INITIALIZE));
ws.on('message', (data: WebSocket.RawData) => {
try {
const message = JSON.parse(data.toString()) as Record<
string,
unknown
>;
if (message['id'] === 1) {
ws.send(JSON.stringify(requests[0]));
return;
}
const request = requests[responses.length];
if (message['id'] !== request?.['id']) return;
responses.push(message);
if (responses.length === requests.length) {
ws.close();
resolve(responses);
return;
}
ws.send(JSON.stringify(requests[responses.length]));
} catch (err) {
ws.terminate();
reject(err as Error);
}
});
ws.on('error', reject);
});
}
async function initializeWs(pathname: string): Promise<{
result?: {
protocolVersion?: number;
@@ -369,6 +421,43 @@ describe('workspace-qualified ACP (/workspaces/:workspace/acp)', () => {
expect(primary.status).toBe(404);
});
it('keeps ACP shell scoped to the workspace-qualified connection', async () => {
const secondary = await sendWsRequests('/workspaces/secondary-id/acp', [
{
jsonrpc: '2.0',
id: 2,
method: 'session/new',
params: { workspaceCwd: '/ws-b' },
},
{
jsonrpc: '2.0',
id: 3,
method: '_qwen/session/shell',
params: { sessionId: 'secondary-session', command: 'pwd' },
},
]);
expect(secondary[1]?.['result']).toMatchObject({
exitCode: 0,
output: 'ok',
});
expect(secondaryBridge.executeShellCommand).toHaveBeenCalledWith(
'secondary-session',
'pwd',
expect.any(AbortSignal),
expect.objectContaining({ clientId: 'secondary-client' }),
);
expect(primaryBridge.executeShellCommand).not.toHaveBeenCalled();
const primary = await sendWsRequest('/acp', {
jsonrpc: '2.0',
id: 4,
method: '_qwen/session/shell',
params: { sessionId: 'secondary-session', command: 'pwd' },
});
expect(primary['error']).toMatchObject({ code: -32602 });
expect(primaryBridge.executeShellCommand).not.toHaveBeenCalled();
});
it('rejects a body workspaceCwd that differs from the selected mount', async () => {
const response = await sendWsRequest('/workspaces/secondary-id/acp', {
jsonrpc: '2.0',
packages/cli/src/serve/capabilities.ts
+14
View File
@@ -273,6 +273,10 @@ export const SERVE_CAPABILITY_REGISTRY = {
// Multi-workspace sessions closed loop (issue #6378 Phase 2a). Advertised
// only when one daemon hosts more than one registered workspace runtime.
multi_workspace_sessions: { since: 'v1' },
// Singular session rewind routes resolve the owning live workspace runtime.
multi_workspace_session_rewind: { since: 'v1' },
// Singular session shell routes resolve the owning live workspace runtime.
multi_workspace_session_shell: { since: 'v1' },
persistent_workspace_registration: { since: 'v1' },
// Workspace-qualified core REST routes under `/workspaces/:workspace/...`.
// Covers core file/status/permissions/trust/lifecycle/MCP/tool, memory,
@@ -450,6 +454,16 @@ export const CONDITIONAL_SERVE_FEATURES: ReadonlyMap<
'multi_workspace_sessions',
(toggles) => toggles.multiWorkspaceSessionsEnabled === true,
],
[
'multi_workspace_session_rewind',
(toggles) => toggles.multiWorkspaceSessionsEnabled === true,
],
[
'multi_workspace_session_shell',
(toggles) =>
toggles.multiWorkspaceSessionsEnabled === true &&
toggles.sessionShellCommandEnabled === true,
],
[
'persistent_workspace_registration',
(toggles) => toggles.persistentWorkspaceRegistrationAvailable === true,
packages/cli/src/serve/multi-workspace-sessions.test.ts
+493 -3
View File
@@ -16,6 +16,8 @@ import {
setDebugLogSession,
} from '@qwen-code/qwen-code-core';
import {
InvalidRewindTargetError,
SessionBusyError,
SessionNotFoundError,
type AcpSessionBridge,
type BridgeClientRequestContext,
@@ -117,6 +119,18 @@ interface FakeBridge extends AcpSessionBridge {
taskKind: 'agent' | 'shell' | 'monitor';
}>;
readonly goalClearCalls: string[];
readonly rewindSnapshotCalls: string[];
readonly rewindCalls: Array<{
sessionId: string;
req: Parameters<AcpSessionBridge['rewindSession']>[1];
context?: BridgeClientRequestContext;
}>;
readonly shellCalls: Array<{
sessionId: string;
command: string;
signal?: AbortSignal;
context?: BridgeClientRequestContext;
}>;
}
function makeSummary(
@@ -212,7 +226,11 @@ async function withRuntimeDir<T>(fn: () => Promise<T>): Promise<T> {
function makeBridge(
workspaceCwd: string,
summaries: BridgeSessionSummary[] = [],
options: { channelLive?: boolean } = {},
options: {
channelLive?: boolean;
rewindImpl?: AcpSessionBridge['rewindSession'];
shellImpl?: AcpSessionBridge['executeShellCommand'];
} = {},
): FakeBridge {
const live = new Map(
summaries.map((summary) => [summary.sessionId, summary]),
@@ -238,6 +256,9 @@ function makeBridge(
const midTurnMessageCalls: FakeBridge['midTurnMessageCalls'] = [];
const taskCancelCalls: FakeBridge['taskCancelCalls'] = [];
const goalClearCalls: string[] = [];
const rewindSnapshotCalls: string[] = [];
const rewindCalls: FakeBridge['rewindCalls'] = [];
const shellCalls: FakeBridge['shellCalls'] = [];
const bridge = {
permissionPolicy: 'first-responder' as const,
spawnCalls,
@@ -261,6 +282,9 @@ function makeBridge(
midTurnMessageCalls,
taskCancelCalls,
goalClearCalls,
rewindSnapshotCalls,
rewindCalls,
shellCalls,
get sessionCount() {
return live.size;
},
@@ -445,6 +469,60 @@ function makeBridge(
condition: workspaceCwd,
};
},
async getRewindSnapshots(sessionId: string) {
rewindSnapshotCalls.push(sessionId);
return {
snapshots: [
{
promptId: `${workspaceCwd}-prompt`,
turnIndex: 0,
timestamp: '2026-07-08T00:00:00.000Z',
diffStats: {
filesChanged: 1,
insertions: 1,
deletions: 0,
},
},
],
};
},
async rewindSession(
sessionId: string,
req: Parameters<AcpSessionBridge['rewindSession']>[1],
context?: BridgeClientRequestContext,
) {
rewindCalls.push({
sessionId,
req,
...(context ? { context } : {}),
});
if (options.rewindImpl) {
return options.rewindImpl(sessionId, req, context);
}
return {
rewound: true,
targetTurnIndex: 0,
filesChanged: req.rewindFiles === false ? [] : ['tracked.txt'],
filesFailed: [],
};
},
async executeShellCommand(
sessionId: string,
command: string,
signal?: AbortSignal,
context?: BridgeClientRequestContext,
) {
shellCalls.push({
sessionId,
command,
...(signal ? { signal } : {}),
...(context ? { context } : {}),
});
if (options.shellImpl) {
return options.shellImpl(sessionId, command, signal, context);
}
return { exitCode: 0, output: workspaceCwd, aborted: false };
},
async cancelSession(sessionId: string) {
cancelCalls.push(sessionId);
},
@@ -542,12 +620,16 @@ function makeHarness(opts?: {
secondaryTrusted?: boolean;
secondaryChannelLive?: boolean;
daemonLog?: DaemonLogger;
primarySummaries?: BridgeSessionSummary[];
secondarySummaries?: BridgeSessionSummary[];
token?: string;
secondaryRewindImpl?: AcpSessionBridge['rewindSession'];
secondaryShellImpl?: AcpSessionBridge['executeShellCommand'];
serveOptions?: Partial<ServeOptions>;
}) {
const primaryBridge = makeBridge(
PRIMARY_CWD,
[makeSummary('primary-session', PRIMARY_CWD)],
opts?.primarySummaries ?? [makeSummary('primary-session', PRIMARY_CWD)],
{ channelLive: true },
);
const secondaryBridge = makeBridge(
@@ -555,7 +637,15 @@ function makeHarness(opts?: {
opts?.secondarySummaries ?? [
makeSummary('secondary-session', SECONDARY_CWD),
],
{ channelLive: opts?.secondaryChannelLive ?? true },
{
channelLive: opts?.secondaryChannelLive ?? true,
...(opts?.secondaryRewindImpl
? { rewindImpl: opts.secondaryRewindImpl }
: {}),
...(opts?.secondaryShellImpl
? { shellImpl: opts.secondaryShellImpl }
: {}),
},
);
const registry = createWorkspaceRegistry([
makeRuntime({
@@ -578,6 +668,7 @@ function makeHarness(opts?: {
...baseOpts,
workspace: PRIMARY_CWD,
...(opts?.token !== undefined ? { token: opts.token } : {}),
...opts?.serveOptions,
},
undefined,
{
@@ -600,6 +691,8 @@ describe('multi-workspace session dispatch', () => {
expect(res.status).toBe(200);
expect(res.body.workspaceCwd).toBe(PRIMARY_CWD);
expect(res.body.features).toContain('multi_workspace_sessions');
expect(res.body.features).toContain('multi_workspace_session_rewind');
expect(res.body.features).not.toContain('multi_workspace_session_shell');
expect(res.body.features).toContain('workspace_persisted_transcript');
expect(res.body.workspaces).toEqual([
{ id: 'primary-id', cwd: PRIMARY_CWD, primary: true, trusted: true },
@@ -614,6 +707,20 @@ describe('multi-workspace session dispatch', () => {
expect(res.body.limits.maxTotalSessions).toBeNull();
});
it('advertises multi-workspace shell only when effective session shell is enabled', async () => {
const { app } = makeHarness({
serveOptions: { token: 'secret', enableSessionShell: true },
});
const res = await request(app)
.get('/capabilities')
.set('Host', host())
.set('Authorization', 'Bearer secret');
expect(res.status).toBe(200);
expect(res.body.features).toContain('session_shell_command');
expect(res.body.features).toContain('multi_workspace_session_shell');
});
it('aggregates daemon status session count and exposes workspace metadata', async () => {
const { app } = makeHarness();
const res = await request(app).get('/daemon/status').set('Host', host());
@@ -788,6 +895,389 @@ describe('multi-workspace session dispatch', () => {
expect(secondaryBridge.detachCalls).toEqual(['secondary-session']);
});
it('routes secondary rewind snapshots, rewind, and shell only to the owner bridge', async () => {
const daemonLog = makeDaemonLog();
const { app, primaryBridge, secondaryBridge } = makeHarness({
daemonLog,
serveOptions: { token: 'secret', enableSessionShell: true },
});
const auth = (test: request.Test) =>
test.set('Host', host()).set('Authorization', 'Bearer secret');
const snapshots = await auth(
request(app).get('/session/secondary-session/rewind/snapshots'),
);
expect(snapshots.status).toBe(200);
expect(snapshots.body.snapshots[0].promptId).toBe(
`${SECONDARY_CWD}-prompt`,
);
const rewind = await auth(
request(app).post('/session/secondary-session/rewind'),
)
.set('X-Qwen-Client-Id', 'client-2')
.send({ promptId: 'secondary-prompt', rewindFiles: true });
expect(rewind.status).toBe(200);
expect(rewind.body.filesChanged).toEqual(['tracked.txt']);
const shell = await auth(
request(app).post('/session/secondary-session/shell'),
)
.set('X-Qwen-Client-Id', 'client-2')
.send({ command: ' pwd ' });
expect(shell.status).toBe(200);
expect(shell.body.output).toBe(SECONDARY_CWD);
expect(primaryBridge.rewindSnapshotCalls).toEqual([]);
expect(primaryBridge.rewindCalls).toEqual([]);
expect(primaryBridge.shellCalls).toEqual([]);
expect(secondaryBridge.rewindSnapshotCalls).toEqual(['secondary-session']);
expect(secondaryBridge.rewindCalls).toEqual([
{
sessionId: 'secondary-session',
req: { promptId: 'secondary-prompt', rewindFiles: true },
context: { clientId: 'client-2' },
},
]);
expect(secondaryBridge.shellCalls).toEqual([
{
sessionId: 'secondary-session',
command: 'pwd',
signal: expect.any(AbortSignal),
context: { clientId: 'client-2' },
},
]);
expect(daemonLog.info).toHaveBeenCalledWith('rewind snapshots loaded', {
sessionId: 'secondary-session',
snapshotCount: 1,
workspaceId: 'secondary-id',
workspaceCwd: SECONDARY_CWD,
});
expect(daemonLog.info).toHaveBeenCalledWith('session rewind completed', {
sessionId: 'secondary-session',
promptId: 'secondary-prompt',
rewindFiles: true,
rewound: true,
filesChangedCount: 1,
filesFailedCount: 0,
workspaceId: 'secondary-id',
workspaceCwd: SECONDARY_CWD,
});
expect(daemonLog.info).toHaveBeenCalledWith('shell command completed', {
sessionId: 'secondary-session',
clientId: 'client-2',
exitCode: 0,
workspaceId: 'secondary-id',
workspaceCwd: SECONDARY_CWD,
});
});
it('preserves rewindFiles defaults and rejects non-boolean values', async () => {
const { app, primaryBridge, secondaryBridge } = makeHarness({
serveOptions: { token: 'secret' },
});
const rewind = (body: Record<string, unknown>) =>
request(app)
.post('/session/secondary-session/rewind')
.set('Host', host())
.set('Authorization', 'Bearer secret')
.send(body);
for (const [body, expected] of [
[{ promptId: 'omitted' }, true],
[{ promptId: 'true', rewindFiles: true }, true],
[{ promptId: 'false', rewindFiles: false }, false],
] as const) {
await rewind(body).expect(200);
expect(secondaryBridge.rewindCalls.at(-1)?.req.rewindFiles).toBe(
expected,
);
}
for (const rewindFiles of ['false', 0, null]) {
const response = await rewind({ promptId: 'invalid', rewindFiles });
expect(response.status).toBe(400);
expect(response.body.code).toBe('invalid_rewind_files_flag');
}
for (const body of [{}, { promptId: '' }]) {
const response = await rewind(body);
expect(response.status).toBe(400);
expect(response.body.code).toBe('missing_prompt_id');
}
expect(primaryBridge.rewindCalls).toEqual([]);
expect(secondaryBridge.rewindCalls).toHaveLength(3);
});
it('fails closed for unknown, untrusted, and ambiguous rewind owners', async () => {
const unknown = makeHarness();
const unknownRes = await request(unknown.app)
.get('/session/missing/rewind/snapshots')
.set('Host', host());
expect(unknownRes.status).toBe(404);
expect(unknownRes.body.code).toBe('session_not_found');
const untrusted = makeHarness({ secondaryTrusted: false });
const untrustedRes = await request(untrusted.app)
.get('/session/secondary-session/rewind/snapshots')
.set('Host', host());
expect(untrustedRes.status).toBe(403);
expect(untrustedRes.body.code).toBe('untrusted_workspace');
expect(untrusted.secondaryBridge.rewindSnapshotCalls).toEqual([]);
const duplicate = makeSummary('duplicate-session', PRIMARY_CWD);
const ambiguous = makeHarness({
primarySummaries: [duplicate],
secondarySummaries: [makeSummary('duplicate-session', SECONDARY_CWD)],
});
const ambiguousRes = await request(ambiguous.app)
.get('/session/duplicate-session/rewind/snapshots')
.set('Host', host());
expect(ambiguousRes.status).toBe(500);
expect(ambiguousRes.body.code).toBe('ambiguous_session_owner');
expect(ambiguous.primaryBridge.rewindSnapshotCalls).toEqual([]);
expect(ambiguous.secondaryBridge.rewindSnapshotCalls).toEqual([]);
});
it('rejects untrusted secondary rewind and shell before bridge execution', async () => {
const { app, secondaryBridge } = makeHarness({
secondaryTrusted: false,
serveOptions: { token: 'secret', enableSessionShell: true },
});
const auth = (test: request.Test) =>
test.set('Host', host()).set('Authorization', 'Bearer secret');
const rewind = await auth(
request(app).post('/session/secondary-session/rewind'),
).send({ promptId: 'secondary-prompt' });
expect(rewind.status).toBe(403);
expect(rewind.body.code).toBe('untrusted_workspace');
const shell = await auth(
request(app).post('/session/secondary-session/shell'),
)
.set('X-Qwen-Client-Id', 'client-2')
.send({ command: 'pwd' });
expect(shell.status).toBe(403);
expect(shell.body.code).toBe('untrusted_workspace');
expect(secondaryBridge.rewindCalls).toEqual([]);
expect(secondaryBridge.shellCalls).toEqual([]);
});
it('aborts only the owning secondary shell when the HTTP client disconnects', async () => {
let markStarted: (() => void) | undefined;
const started = new Promise<void>((resolve) => {
markStarted = resolve;
});
let markAborted: (() => void) | undefined;
const aborted = new Promise<void>((resolve) => {
markAborted = resolve;
});
let secondarySignal: AbortSignal | undefined;
const { app, primaryBridge, secondaryBridge } = makeHarness({
serveOptions: { token: 'secret', enableSessionShell: true },
secondaryShellImpl: async (_sessionId, _command, signal) => {
secondarySignal = signal;
markStarted?.();
await new Promise<void>((resolve) => {
signal?.addEventListener(
'abort',
() => {
markAborted?.();
resolve();
},
{ once: true },
);
});
return { exitCode: null, output: '', aborted: true };
},
});
const pending = request(app)
.post('/session/secondary-session/shell')
.set('Host', host())
.set('Authorization', 'Bearer secret')
.set('X-Qwen-Client-Id', 'client-2')
.send({ command: 'sleep 10' });
const response = pending.then(
() => undefined,
() => undefined,
);
await started;
pending.abort();
await Promise.all([response, aborted]);
expect(secondarySignal?.aborted).toBe(true);
expect(primaryBridge.shellCalls).toEqual([]);
expect(secondaryBridge.shellCalls).toHaveLength(1);
});
it('preserves strict shell validation order for a secondary owner', async () => {
const disabled = makeHarness({ serveOptions: { token: 'secret' } });
const disabledResponse = await request(disabled.app)
.post('/session/secondary-session/shell')
.set('Host', host())
.set('Authorization', 'Bearer secret')
.send({ command: '' });
expect(disabledResponse.status).toBe(403);
expect(disabledResponse.body.code).toBe('session_shell_disabled');
const enabled = makeHarness({
serveOptions: { token: 'secret', enableSessionShell: true },
});
const tokenRequired = await request(enabled.app)
.post('/session/secondary-session/shell')
.set('Host', host())
.send({ command: 'pwd' });
expect(tokenRequired.status).toBe(401);
expect(tokenRequired.body.error).toBe('Unauthorized');
const clientRequired = await request(enabled.app)
.post('/session/secondary-session/shell')
.set('Host', host())
.set('Authorization', 'Bearer secret')
.send({ command: '' });
expect(clientRequired.status).toBe(403);
expect(clientRequired.body.code).toBe('client_id_required');
const emptyCommand = await request(enabled.app)
.post('/session/secondary-session/shell')
.set('Host', host())
.set('Authorization', 'Bearer secret')
.set('X-Qwen-Client-Id', 'client-2')
.send({ command: ' ' });
expect(emptyCommand.status).toBe(400);
expect(disabled.secondaryBridge.shellCalls).toEqual([]);
expect(enabled.primaryBridge.shellCalls).toEqual([]);
expect(enabled.secondaryBridge.shellCalls).toEqual([]);
});
it('keeps primary rewind and shell behavior when multiple runtimes exist', async () => {
const { app, primaryBridge, secondaryBridge } = makeHarness({
serveOptions: { token: 'secret', enableSessionShell: true },
});
await request(app)
.get('/session/primary-session/rewind/snapshots')
.set('Host', host())
.set('Authorization', 'Bearer secret')
.expect(200);
await request(app)
.post('/session/primary-session/rewind')
.set('Host', host())
.set('Authorization', 'Bearer secret')
.send({ promptId: 'primary-prompt', rewindFiles: false })
.expect(200);
await request(app)
.post('/session/primary-session/shell')
.set('Host', host())
.set('Authorization', 'Bearer secret')
.set('X-Qwen-Client-Id', 'client-1')
.send({ command: 'pwd' })
.expect(200);
expect(primaryBridge.rewindSnapshotCalls).toEqual(['primary-session']);
expect(primaryBridge.rewindCalls).toHaveLength(1);
expect(primaryBridge.shellCalls).toHaveLength(1);
expect(secondaryBridge.rewindSnapshotCalls).toEqual([]);
expect(secondaryBridge.rewindCalls).toEqual([]);
expect(secondaryBridge.shellCalls).toEqual([]);
});
it('keeps rewind and shell behavior in a single-workspace daemon', async () => {
const bridge = makeBridge(PRIMARY_CWD, [
makeSummary('primary-session', PRIMARY_CWD),
]);
const app = createServeApp(
{
...baseOpts,
workspace: PRIMARY_CWD,
token: 'secret',
enableSessionShell: true,
},
undefined,
{ bridge },
);
const auth = (test: request.Test) =>
test.set('Host', host()).set('Authorization', 'Bearer secret');
const capabilities = await auth(request(app).get('/capabilities'));
expect(capabilities.body.features).not.toContain(
'multi_workspace_session_rewind',
);
expect(capabilities.body.features).not.toContain(
'multi_workspace_session_shell',
);
await auth(
request(app).get('/session/primary-session/rewind/snapshots'),
).expect(200);
await auth(request(app).post('/session/primary-session/rewind'))
.send({ promptId: 'primary-prompt', rewindFiles: false })
.expect(200);
await auth(request(app).post('/session/primary-session/shell'))
.set('X-Qwen-Client-Id', 'client-1')
.send({ command: 'pwd' })
.expect(200);
expect(bridge.rewindSnapshotCalls).toEqual(['primary-session']);
expect(bridge.rewindCalls).toHaveLength(1);
expect(bridge.shellCalls).toHaveLength(1);
});
it('preserves rewind busy, invalid-target, and partial-restore contracts', async () => {
const postRewind = (
app: ReturnType<typeof createServeApp>,
promptId: string,
) =>
request(app)
.post('/session/secondary-session/rewind')
.set('Host', host())
.set('Authorization', 'Bearer secret')
.send({ promptId });
const busy = makeHarness({
serveOptions: { token: 'secret' },
secondaryRewindImpl: async (sessionId) => {
throw new SessionBusyError(sessionId);
},
});
const busyResponse = await postRewind(busy.app, 'busy');
expect(busyResponse.status).toBe(409);
expect(busyResponse.body.code).toBe('session_busy');
expect(busyResponse.headers['retry-after']).toBe('5');
const invalid = makeHarness({
serveOptions: { token: 'secret' },
secondaryRewindImpl: async (sessionId) => {
throw new InvalidRewindTargetError(sessionId);
},
});
const invalidResponse = await postRewind(invalid.app, 'invalid');
expect(invalidResponse.status).toBe(400);
expect(invalidResponse.body.code).toBe('invalid_rewind_target');
const partial = makeHarness({
serveOptions: { token: 'secret' },
secondaryRewindImpl: async () => ({
rewound: false,
targetTurnIndex: 1,
filesChanged: ['restored.txt'],
filesFailed: ['failed.txt'],
}),
});
const partialResponse = await postRewind(partial.app, 'partial');
expect(partialResponse.status).toBe(200);
expect(partialResponse.body).toEqual({
rewound: false,
targetTurnIndex: 1,
filesChanged: ['restored.txt'],
filesFailed: ['failed.txt'],
});
});
it('dispatches secondary events, permissions, pending prompts, and close', async () => {
const { app, primaryBridge, secondaryBridge } = makeHarness();
packages/cli/src/serve/routes/session.ts
+47 -24
View File
@@ -2887,9 +2887,9 @@ export function registerSessionRoutes(
app.post(
'/session/:id/shell',
mutate({ strict: true }),
withMutableSession(
withOwnerMutableSession(
'POST /session/:id/shell',
async (req, res, sessionId) => {
async (req, res, sessionId, runtime) => {
if (!sessionShellCommandEnabled) {
sendBridgeError(res, new SessionShellDisabledError(), {
route: 'POST /session/:id/shell',
@@ -2922,7 +2922,7 @@ export function registerSessionRoutes(
};
res.once('close', onResClose);
try {
const result = await bridge.executeShellCommand(
const result = await runtime.bridge.executeShellCommand(
sessionId,
command.trim(),
abort.signal,
@@ -2933,6 +2933,8 @@ export function registerSessionRoutes(
sessionId,
clientId,
exitCode: result.exitCode,
workspaceId: runtime.workspaceId,
workspaceCwd: runtime.workspaceCwd,
});
}
res.status(200).json(result);
@@ -2955,30 +2957,31 @@ export function registerSessionRoutes(
),
);
app.get('/session/:id/rewind/snapshots', async (req, res) => {
const sessionId = req.params['id'];
if (!sessionId) {
res
.status(400)
.json({ error: '`sessionId` route parameter is required' });
return;
}
try {
res.status(200).json(await bridge.getRewindSnapshots(sessionId));
} catch (err) {
sendBridgeError(res, err, {
route: 'GET /session/:id/rewind/snapshots',
sessionId,
});
}
});
app.get(
'/session/:id/rewind/snapshots',
withOwnerReadSession(
'GET /session/:id/rewind/snapshots',
async (_req, res, sessionId, runtime) => {
const response = await runtime.bridge.getRewindSnapshots(sessionId);
if (daemonLog) {
daemonLog.info('rewind snapshots loaded', {
sessionId,
snapshotCount: response.snapshots.length,
workspaceId: runtime.workspaceId,
workspaceCwd: runtime.workspaceCwd,
});
}
res.status(200).json(response);
},
),
);
app.post(
'/session/:id/rewind',
mutate({ strict: true }),
withMutableSession(
withOwnerMutableSession(
'POST /session/:id/rewind',
async (req, res, sessionId) => {
async (req, res, sessionId, runtime) => {
const body = safeBody(req);
const promptId = body['promptId'];
if (typeof promptId !== 'string' || promptId.length === 0) {
@@ -2988,13 +2991,33 @@ export function registerSessionRoutes(
});
return;
}
const rewindFiles = body['rewindFiles'];
if (rewindFiles !== undefined && typeof rewindFiles !== 'boolean') {
res.status(400).json({
error: '`rewindFiles` must be a boolean when provided',
code: 'invalid_rewind_files_flag',
});
return;
}
const clientId = parseClientIdHeader(req, res);
if (clientId === null) return;
const response = await bridge.rewindSession(
const response = await runtime.bridge.rewindSession(
sessionId,
{ promptId, rewindFiles: body['rewindFiles'] !== false },
{ promptId, rewindFiles: rewindFiles !== false },
clientId !== undefined ? { clientId } : undefined,
);
if (daemonLog) {
daemonLog.info('session rewind completed', {
sessionId,
promptId,
rewindFiles: rewindFiles !== false,
rewound: response.rewound,
filesChangedCount: response.filesChanged.length,
filesFailedCount: response.filesFailed.length,
workspaceId: runtime.workspaceId,
workspaceCwd: runtime.workspaceCwd,
});
}
res.status(200).json(response);
},
),
packages/cli/src/serve/server.test.ts
+49
View File
@@ -394,6 +394,8 @@ const EXPECTED_REGISTERED_FEATURES = [
'channel_reload',
'channel_control',
'multi_workspace_sessions',
'multi_workspace_session_rewind',
'multi_workspace_session_shell',
'persistent_workspace_registration',
'workspace_qualified_rest_core',
'workspace_persisted_transcript',
@@ -2238,6 +2240,53 @@ describe('createServeApp', () => {
);
continue;
}
if (feature === 'multi_workspace_session_rewind') {
expect(predicate({ multiWorkspaceSessionsEnabled: true })).toBe(true);
expect(predicate({ multiWorkspaceSessionsEnabled: false })).toBe(
false,
);
expect(predicate({})).toBe(false);
expect(
getAdvertisedServeFeatures(undefined, {
multiWorkspaceSessionsEnabled: true,
}),
).toContain(feature);
expect(getAdvertisedServeFeatures(undefined, {})).not.toContain(
feature,
);
continue;
}
if (feature === 'multi_workspace_session_shell') {
expect(
predicate({
multiWorkspaceSessionsEnabled: true,
sessionShellCommandEnabled: true,
}),
).toBe(true);
expect(
predicate({
multiWorkspaceSessionsEnabled: true,
sessionShellCommandEnabled: false,
}),
).toBe(false);
expect(
predicate({
multiWorkspaceSessionsEnabled: false,
sessionShellCommandEnabled: true,
}),
).toBe(false);
expect(predicate({})).toBe(false);
expect(
getAdvertisedServeFeatures(undefined, {
multiWorkspaceSessionsEnabled: true,
sessionShellCommandEnabled: true,
}),
).toContain(feature);
expect(getAdvertisedServeFeatures(undefined, {})).not.toContain(
feature,
);
continue;
}
if (feature === 'persistent_workspace_registration') {
expect(
predicate({ persistentWorkspaceRegistrationAvailable: true }),
packages/cli/src/serve/server.ts
+32 -19
View File
@@ -1000,27 +1000,40 @@ export function createServeApp(
});
app.use(
daemonTelemetryMiddleware((req) => {
const match = req.path.match(/^\/workspaces\/([^/]+)/);
const rawSelector = match?.[1];
if (rawSelector) {
try {
const selector = decodeURIComponent(rawSelector);
const byId = workspaceRegistry.getByWorkspaceId(selector);
if (byId) return byId.workspaceCwd;
if (isPortableAbsolutePath(selector)) {
const runtime = resolveRegisteredWorkspaceRuntimeByPathSelector(
workspaceRegistry,
selector,
);
if (runtime) return runtime.workspaceCwd;
daemonTelemetryMiddleware(
(req) => {
const match = req.path.match(/^\/workspaces\/([^/]+)/);
const rawSelector = match?.[1];
if (rawSelector) {
try {
const selector = decodeURIComponent(rawSelector);
const byId = workspaceRegistry.getByWorkspaceId(selector);
if (byId) return byId.workspaceCwd;
if (isPortableAbsolutePath(selector)) {
const runtime = resolveRegisteredWorkspaceRuntimeByPathSelector(
workspaceRegistry,
selector,
);
if (runtime) return runtime.workspaceCwd;
}
} catch {
return primaryBoundWorkspace;
}
} catch {
return primaryBoundWorkspace;
}
}
return primaryBoundWorkspace;
}, deps.recordDaemonRequest),
return primaryBoundWorkspace;
},
deps.recordDaemonRequest,
(sessionId) => {
try {
const owner = workspaceRegistry.resolveLiveSessionOwner(sessionId);
return owner.kind === 'found'
? owner.runtime.workspaceCwd
: undefined;
} catch {
return undefined;
}
},
),
);
const buildWorkspaceCtx = createBuildWorkspaceCtx(primaryBoundWorkspace);
packages/cli/src/serve/server/telemetry.test.ts
+87
View File
@@ -140,6 +140,93 @@ describe('daemonTelemetryMiddleware — recordRequest seam', () => {
);
});
it('attributes singular rewind and shell routes to the live session owner', () => {
const resolveSessionWorkspaceCwd = vi.fn(() => '/workspace/secondary');
const mw = daemonTelemetryMiddleware(
() => '/workspace/primary',
undefined,
resolveSessionWorkspaceCwd,
);
for (const [method, path, route] of [
[
'GET',
'/session/secondary-session/rewind/snapshots',
'GET /session/:id/rewind/snapshots',
],
['POST', '/session/secondary-session/rewind', 'POST /session/:id/rewind'],
['POST', '/session/secondary-session/shell', 'POST /session/:id/shell'],
] as const) {
const res = mockRes(200);
mw(mockReq(method, path), res, vi.fn() as unknown as NextFunction);
res.emit('finish');
expect(coreMocks.withDaemonRequestSpan).toHaveBeenLastCalledWith(
expect.objectContaining({
method,
route,
sessionId: 'secondary-session',
workspaceHash: 'hash:/workspace/secondary',
}),
expect.any(Function),
);
}
expect(resolveSessionWorkspaceCwd).toHaveBeenCalledTimes(3);
expect(resolveSessionWorkspaceCwd).toHaveBeenCalledWith(
'secondary-session',
);
});
it('decodes session ids before owner lookup and span attribution', () => {
const resolveSessionWorkspaceCwd = vi.fn(() => '/workspace/secondary');
const mw = daemonTelemetryMiddleware(
() => '/workspace/primary',
undefined,
resolveSessionWorkspaceCwd,
);
const res = mockRes(200);
mw(
mockReq('POST', '/session/secondary%2Fsession/rewind'),
res,
vi.fn() as unknown as NextFunction,
);
res.emit('finish');
expect(resolveSessionWorkspaceCwd).toHaveBeenCalledWith(
'secondary/session',
);
expect(coreMocks.withDaemonRequestSpan).toHaveBeenCalledWith(
expect.objectContaining({ sessionId: 'secondary/session' }),
expect.any(Function),
);
});
it('keeps malformed session id encodings without throwing', () => {
const resolveSessionWorkspaceCwd = vi.fn(() => undefined);
const mw = daemonTelemetryMiddleware(
() => '/workspace/primary',
undefined,
resolveSessionWorkspaceCwd,
);
const res = mockRes(200);
expect(() => {
mw(
mockReq('POST', '/session/bad%ZZ/rewind'),
res,
vi.fn() as unknown as NextFunction,
);
}).not.toThrow();
res.emit('finish');
expect(resolveSessionWorkspaceCwd).toHaveBeenCalledWith('bad%ZZ');
expect(coreMocks.withDaemonRequestSpan).toHaveBeenCalledWith(
expect.objectContaining({ sessionId: 'bad%ZZ' }),
expect.any(Function),
);
});
it('normalizes plural workspace agent routes to stable route labels', () => {
const mw = daemonTelemetryMiddleware(() => '/ws');
for (const [method, path, route] of [
packages/cli/src/serve/server/telemetry.ts
+29 -2
View File
@@ -18,6 +18,14 @@ import {
MAX_CLIENT_ID_LENGTH,
} from './request-helpers.js';
function decodePathSegment(value: string): string {
try {
return decodeURIComponent(value);
} catch {
return value;
}
}
// Route handlers are split across `routes/*.ts`; any added or renamed route
// that needs daemon telemetry must keep these patterns in sync.
export function resolveDaemonTelemetryRoute(
@@ -35,6 +43,13 @@ export function resolveDaemonTelemetryRoute(
if (req.method === 'GET' && path === '/daemon/status') {
return { route: 'GET /daemon/status' };
}
const rewindSnapshots = path.match(/^\/session\/([^/]+)\/rewind\/snapshots$/);
if (rewindSnapshots?.[1] && req.method === 'GET') {
return {
route: 'GET /session/:id/rewind/snapshots',
sessionId: rewindSnapshots[1],
};
}
const sessionAction = path.match(
/^\/session\/([^/]+)\/(load|resume|prompt|cancel|recap|btw|mid-turn-message|model|shell|detach|rewind|approval-mode|language|a2ui-action)$/,
);
@@ -295,6 +310,7 @@ export function daemonTelemetryMiddleware(
// the OTel counter's scope, so the "requests" line reflects daemon API
// traffic rather than static-asset or unrouted noise.
recordRequest?: (durationMs: number, statusCode: number) => void,
resolveSessionWorkspaceCwd?: (sessionId: string) => string | undefined,
): (req: Request, res: Response, next: NextFunction) => void {
const workspaceHashByCwd = new Map<string, string>();
const resolveWorkspaceHash = (workspaceCwd: string): string => {
@@ -311,7 +327,18 @@ export function daemonTelemetryMiddleware(
next();
return;
}
const workspaceHash = resolveWorkspaceHash(resolveWorkspaceCwd(req));
const resolveOwnerWorkspace =
route.route === 'GET /session/:id/rewind/snapshots' ||
route.route === 'POST /session/:id/rewind' ||
route.route === 'POST /session/:id/shell';
const sessionId = route.sessionId
? decodePathSegment(route.sessionId)
: undefined;
const workspaceCwd =
(resolveOwnerWorkspace && sessionId
? resolveSessionWorkspaceCwd?.(sessionId)
: undefined) ?? resolveWorkspaceCwd(req);
const workspaceHash = resolveWorkspaceHash(workspaceCwd);
const rawClientId = req.get(CLIENT_ID_HEADER);
const clientId =
rawClientId !== undefined &&
@@ -326,7 +353,7 @@ export function daemonTelemetryMiddleware(
method: req.method,
route: route.route,
workspaceHash,
...(route.sessionId ? { sessionId: route.sessionId } : {}),
...(sessionId ? { sessionId } : {}),
...(route.permissionRequestId
? { permissionRequestId: route.permissionRequestId }
: {}),
packages/sdk-typescript/src/daemon/AcpHttpTransport.ts
+2
View File
@@ -199,6 +199,7 @@ export class AcpHttpTransport implements DaemonTransport {
readonly type = 'acp-http' as const;
readonly supportsReplay = true;
readonly restFetch: typeof globalThis.fetch;
constructor(
baseUrl: string,
@@ -208,6 +209,7 @@ export class AcpHttpTransport implements DaemonTransport {
this.baseUrl = baseUrl;
this.token = token;
this._fetch = fetchFn;
this.restFetch = fetchFn;
}
get connected(): boolean {
packages/sdk-typescript/src/daemon/AcpWsTransport.ts
+7 -1
View File
@@ -109,10 +109,16 @@ export class AcpWsTransport implements DaemonTransport {
readonly type = 'acp-ws' as const;
readonly supportsReplay = false;
readonly restFetch: typeof globalThis.fetch | undefined;
constructor(wsUrl: string, token?: string) {
constructor(
wsUrl: string,
token?: string,
restFetch?: typeof globalThis.fetch,
) {
this.wsUrl = wsUrl;
this.token = token;
this.restFetch = restFetch;
}
get connected(): boolean {
packages/sdk-typescript/src/daemon/AutoReconnectTransport.ts
+6 -1
View File
@@ -52,6 +52,7 @@ export class AutoReconnectTransport implements DaemonTransport {
private reconnecting: Promise<void> | undefined;
readonly supportsReplay: boolean;
readonly restFetch: typeof globalThis.fetch;
constructor(opts: {
baseUrl: string;
@@ -63,7 +64,11 @@ export class AutoReconnectTransport implements DaemonTransport {
}) {
this.baseUrl = opts.baseUrl;
this.token = opts.token;
this.fetchFn = opts.fetch ?? globalThis.fetch.bind(globalThis);
this.fetchFn =
opts.fetch ??
opts.initial?.restFetch ??
globalThis.fetch.bind(globalThis);
this.restFetch = this.fetchFn;
this.preferredType = opts.preferredType ?? 'rest';
this.factory = opts.factory;
this.inner =
packages/sdk-typescript/src/daemon/DaemonClient.ts
+11 -2
View File
@@ -205,7 +205,9 @@ export interface DaemonClientOptions {
* Pluggable transport. When omitted, a `RestSseTransport` is created
* automatically — this preserves the existing REST+SSE behavior with
* zero caller-side changes. Pass an `AcpWsTransport` or
* `AcpHttpTransport` to use JSON-RPC over WebSocket or HTTP.
* `AcpHttpTransport` to use JSON-RPC over WebSocket or HTTP. Rewind APIs
* intentionally use direct REST even when an ACP transport is configured so
* owner routing and strict mutation authentication remain authoritative.
*/
transport?: DaemonTransport;
}
@@ -460,7 +462,10 @@ export class DaemonClient {
// thread the value through every construction. See
// `readTokenFromEnv` above for browser-safety + trim semantics.
this.token = opts.token ?? readTokenFromEnv();
this._fetch = opts.fetch ?? globalThis.fetch.bind(globalThis);
this._fetch =
opts.fetch ??
opts.transport?.restFetch ??
globalThis.fetch.bind(globalThis);
// Coerce non-positive / non-finite to 0 (= disabled). Without this
// a caller passing `-1` or `NaN` would slip past the
// `Number.isFinite` check inside `fetchWithTimeout` (NaN fails
@@ -2043,6 +2048,8 @@ export class DaemonClient {
}
return (await res.json()) as { snapshots: DaemonRewindSnapshotInfo[] };
},
undefined,
'rest',
);
}
@@ -2072,6 +2079,8 @@ export class DaemonClient {
}
return (await res.json()) as DaemonRewindResult;
},
undefined,
'rest',
);
}
packages/sdk-typescript/src/daemon/DaemonTransport.ts
+3
View File
@@ -53,6 +53,9 @@ export type DaemonTransportType = 'rest' | 'acp-http' | 'acp-ws';
* `DaemonClient` needs no control-flow changes.
*/
export interface DaemonTransport {
/** Native HTTP fetch used by REST-only daemon endpoints. */
readonly restFetch?: typeof globalThis.fetch;
/**
* Issue an HTTP-shaped request. REST transports delegate to the
* underlying `fetch`; ACP transports translate the URL + body into
packages/sdk-typescript/src/daemon/RestSseTransport.ts
+2
View File
@@ -30,6 +30,7 @@ export class RestSseTransport implements DaemonTransport {
readonly type = 'rest' as const;
readonly supportsReplay = true;
readonly restFetch: typeof globalThis.fetch;
constructor(
baseUrl: string,
@@ -39,6 +40,7 @@ export class RestSseTransport implements DaemonTransport {
this.baseUrl = baseUrl;
this.token = token;
this._fetch = fetchFn;
this.restFetch = fetchFn;
}
get connected(): boolean {
packages/sdk-typescript/src/daemon/negotiateTransport.ts
+1 -1
View File
@@ -92,7 +92,7 @@ export async function negotiateTransport(
const { AcpWsTransport } = await import('./AcpWsTransport.js');
// Convert http(s) → ws(s) for the WS URL.
const wsUrl = baseUrl.replace(/^https:/, 'wss:').replace(/^http:/, 'ws:');
const transport = new AcpWsTransport(wsUrl + '/acp', token);
const transport = new AcpWsTransport(wsUrl + '/acp', token, fetchFn);
// Probe: try to connect with a timeout.
const probeTimer = setTimeout(() => {
/* timeout — handled by race */
packages/sdk-typescript/test/unit/AutoReconnectTransport.test.ts
+12
View File
@@ -58,6 +58,18 @@ describe('AutoReconnectTransport', () => {
// ---- Delegation to inner transport ------------------------------------
describe('delegation', () => {
it('preserves the initial transport native fetch for REST-only calls', () => {
const restFetch = vi.fn() as unknown as typeof globalThis.fetch;
const inner = createMockTransport({ restFetch });
const transport = new AutoReconnectTransport({
baseUrl: 'http://d',
initial: inner,
});
expect(transport.restFetch).toBe(restFetch);
transport.dispose();
});
it('type delegates to inner transport', () => {
const inner = createMockTransport({ type: 'acp-ws' });
const transport = new AutoReconnectTransport({
packages/sdk-typescript/test/unit/DaemonClient.test.ts
+155
View File
@@ -14,6 +14,7 @@ import {
normalizePendingPromptLimit,
} from '../../src/daemon/DaemonClient.js';
import type { DaemonTransport } from '../../src/daemon/DaemonTransport.js';
import { negotiateTransport } from '../../src/daemon/negotiateTransport.js';
import {
DaemonCapabilityMissingError,
isDaemonContentHash,
@@ -705,6 +706,9 @@ describe('DaemonClient', () => {
type: 'acp-http',
supportsReplay: true,
connected: true,
restFetch: vi.fn(async () => {
throw new Error('transport REST fetch must not override opts.fetch');
}) as unknown as typeof globalThis.fetch,
fetch: transportFetch,
async *subscribeEvents() {},
dispose() {},
@@ -1175,6 +1179,157 @@ describe('DaemonClient', () => {
});
});
describe('session rewind transport', () => {
it('reuses the negotiated native fetch for REST-only rewind calls', async () => {
const negotiatedFetch = vi.fn(async (input: string | URL | Request) => {
const url = String(input);
if (url.endsWith('/capabilities')) {
return jsonResponse(200, { transports: ['acp-http'] });
}
if (url.endsWith('/rewind/snapshots')) {
return jsonResponse(200, { snapshots: [] });
}
return jsonResponse(200, {
rewound: true,
targetTurnIndex: 0,
filesChanged: [],
filesFailed: [],
});
}) as unknown as typeof globalThis.fetch;
const globalFetch = vi
.spyOn(globalThis, 'fetch')
.mockRejectedValue(new Error('global fetch must not handle rewind'));
try {
const transport = await negotiateTransport('http://daemon', 'secret', {
fetchFn: negotiatedFetch,
});
const client = new DaemonClient({
baseUrl: 'http://daemon',
token: 'secret',
transport,
});
await expect(client.getRewindSnapshots('s-1')).resolves.toEqual({
snapshots: [],
});
await expect(
client.rewindSession('s-1', 'prompt-1'),
).resolves.toMatchObject({ rewound: true });
expect(negotiatedFetch).toHaveBeenCalledTimes(3);
expect(globalFetch).not.toHaveBeenCalled();
} finally {
globalFetch.mockRestore();
}
});
it('forces owner-aware REST with auth, client id, timeout, and boolean body', async () => {
const { fetch, calls } = recordingFetch((req) => {
if (req.url.endsWith('/rewind/snapshots')) {
return jsonResponse(200, { snapshots: [] });
}
return jsonResponse(200, {
rewound: true,
targetTurnIndex: 0,
filesChanged: [],
filesFailed: [],
});
});
const transportFetch = vi.fn(async () => {
throw new Error('ACP transport must not handle rewind');
});
const transport: DaemonTransport = {
type: 'acp-http',
supportsReplay: true,
connected: true,
fetch: transportFetch,
async *subscribeEvents() {},
dispose() {},
};
const client = new DaemonClient({
baseUrl: 'http://daemon',
token: 'secret',
fetch,
transport,
fetchTimeoutMs: 1234,
});
await expect(client.getRewindSnapshots('with/slash')).resolves.toEqual({
snapshots: [],
});
await expect(
client.rewindSession('with/slash', 'prompt-1', {
clientId: 'client-1',
rewindFiles: false,
}),
).resolves.toMatchObject({ rewound: true });
expect(transportFetch).not.toHaveBeenCalled();
expect(calls).toHaveLength(2);
expect(calls[0]).toMatchObject({
url: 'http://daemon/session/with%2Fslash/rewind/snapshots',
method: 'GET',
headers: { authorization: 'Bearer secret' },
signal: expect.any(AbortSignal),
});
expect(calls[1]).toMatchObject({
url: 'http://daemon/session/with%2Fslash/rewind',
method: 'POST',
headers: {
authorization: 'Bearer secret',
'content-type': 'application/json',
'x-qwen-client-id': 'client-1',
},
signal: expect.any(AbortSignal),
});
expect(JSON.parse(calls[1]!.body!)).toEqual({
promptId: 'prompt-1',
rewindFiles: false,
});
});
it('keeps shell on the configured transport', async () => {
const nativeFetch = vi.fn(async () => {
throw new Error('native REST fetch must not handle ACP shell');
}) as unknown as typeof globalThis.fetch;
const transportFetch = vi.fn(async () =>
jsonResponse(200, { exitCode: 0, output: '/work/b', aborted: false }),
);
const transport: DaemonTransport = {
type: 'acp-http',
supportsReplay: true,
connected: true,
fetch: transportFetch,
async *subscribeEvents() {},
dispose() {},
};
const client = new DaemonClient({
baseUrl: 'http://daemon',
token: 'secret',
fetch: nativeFetch,
transport,
});
await expect(
client.shellCommand('session-b', 'pwd', { clientId: 'client-b' }),
).resolves.toMatchObject({ output: '/work/b' });
expect(nativeFetch).not.toHaveBeenCalled();
expect(transportFetch).toHaveBeenCalledOnce();
expect(transportFetch).toHaveBeenCalledWith(
'http://daemon/session/session-b/shell',
expect.objectContaining({
method: 'POST',
body: JSON.stringify({ command: 'pwd' }),
headers: expect.objectContaining({
Authorization: 'Bearer secret',
'X-Qwen-Client-Id': 'client-b',
}),
}),
);
});
});
describe('bearer auth', () => {
it('attaches Authorization: Bearer when token is set', async () => {
const { fetch, calls } = recordingFetch(() =>
packages/sdk-typescript/test/unit/acpRouteTable.test.ts
+5
View File
@@ -900,6 +900,11 @@ describe('acpRouteTable – matchRoute', () => {
expect(matchRoute('/session/s12/approval-mode', 'POST')).toBeNull();
});
it('keeps rewind routes off ACP so strict REST auth cannot be bypassed', () => {
expect(matchRoute('/session/s12/rewind/snapshots', 'GET')).toBeNull();
expect(matchRoute('/session/s12/rewind', 'POST')).toBeNull();
});
// ---- Unknown/unmatched routes ---------------------------------------
it('returns null for unknown path', () => {