diff --git a/docs/design/ctrl-o-detail-expand/design.md b/docs/design/ctrl-o-detail-expand/design.md index d929ba9e69f..4d986bce521 100644 --- a/docs/design/ctrl-o-detail-expand/design.md +++ b/docs/design/ctrl-o-detail-expand/design.md @@ -411,7 +411,7 @@ claude code 的机制是"**存储层保留完整、显示层按 `verbose` 截断 > - **新增** core helper `packages/core/src/utils/generateContentResponseUtilities.ts` 的 `getToolResponseDisplayText(parts)`（读 `functionResponse.response.output` + 遍历 nested `functionResponse.parts` 媒体占位、空/缺失返回 `undefined`；**不二次截断**；规则见 §4.9 改动点 1）； > - **改** `packages/cli/src/ui/types.ts`：`IndividualToolCallDisplay` 加 `detailedDisplay?: string`（派生、不持久化）； > - **改** `packages/cli/src/ui/hooks/useReactToolScheduler.ts`（live 提取，`success` 分支派生 `detailedDisplay`）、`packages/cli/src/ui/utils/resumeHistoryUtils.ts`（resume 提取，`tool_result` 分支从 `responseParts ?? message.parts` 派生）、`packages/cli/src/ui/components/messages/ToolMessage.tsx` + `ToolGroupMessage.tsx`（渲染拆分：`ToolGroupMessage` 下传 `fullDetail`，`ToolMessage` 仅 `fullDetail && isCollapsibleTool && detailedDisplay` 切数据源）； -> - **不改** `packages/cli/src/acp-integration/session/HistoryReplayer.ts` / `emitters/ToolCallEmitter.ts`——ACP `content[]` 已含完整 `output`（见上表），TUI transcript 不经此路，无需改动； +> - **不改** `packages/cli/src/acp-integration/session/history-replayer.ts` / `emitters/tool-call-emitter.ts`——ACP `content[]` 已含完整 `output`（见上表），TUI transcript 不经此路，无需改动； > - **不改** 持久化 schema（`serializeToolResponse` / `chatRecordingService` / ACP 协议字段）——完整明细已天然存于 `responseParts`，新字段为派生值。 ### C. 新增 diff --git a/docs/design/daemon-session-artifacts/session-artifacts-daemon-api-implementation-design.md b/docs/design/daemon-session-artifacts/session-artifacts-daemon-api-implementation-design.md index c1223724b4e..8af13d34719 100644 --- a/docs/design/daemon-session-artifacts/session-artifacts-daemon-api-implementation-design.md +++ b/docs/design/daemon-session-artifacts/session-artifacts-daemon-api-implementation-design.md @@ -141,7 +141,7 @@ Client 点击时打开 URL；Daemon 不读取、不验证、不预渲染该 URL - `packages/core/src/tools/tool-names.ts` - `packages/core/src/tools/artifact/artifact-tool.ts` - `packages/cli/src/acp-integration/session/Session.ts` -- `packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.ts` +- `packages/cli/src/acp-integration/session/emitters/tool-call-emitter.ts` 现状： @@ -1005,7 +1005,7 @@ Phase A 先接入 `ToolResult.artifacts` 和 `ArtifactTool`；`record_artifact` - `packages/cli/src/acp-integration/session/types.ts` - `ToolCallResultParams.artifacts?` -- `packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.ts` +- `packages/cli/src/acp-integration/session/emitters/tool-call-emitter.ts` - `_meta.artifacts = params.artifacts` - `packages/cli/src/acp-integration/session/Session.ts` - 工具成功后收集 `toolResult.artifacts`。 @@ -1338,7 +1338,7 @@ cd packages/core && npx vitest run src/tools/artifact/artifact-tool.test.ts 命令： ```bash -cd packages/cli && npx vitest run src/acp-integration/session/emitters/ToolCallEmitter.test.ts +cd packages/cli && npx vitest run src/acp-integration/session/emitters/tool-call-emitter.test.ts cd packages/cli && npx vitest run src/acp-integration/session/Session.test.ts ``` diff --git a/docs/design/untrusted-persisted-transcript.md b/docs/design/untrusted-persisted-transcript.md new file mode 100644 index 00000000000..8085b391cb3 --- /dev/null +++ b/docs/design/untrusted-persisted-transcript.md @@ -0,0 +1,47 @@ +# Untrusted persisted transcript reader + +## Context + +The daemon already exposes persisted-only session and group catalogs for registered untrusted secondary workspaces. Transcript paging remains unavailable because the legacy `GET /session/:id/transcript` route resolves a session owner through live bridge state and delegates replay to an ACP child. That path can start a process, load workspace settings and project-defined capabilities, and create or repair a persisted cursor signing key. + +This design adds a separate workspace-qualified REST contract that reads only an active persisted transcript. The legacy route and its ACP-backed behavior remain unchanged. + +## Contract + +The daemon advertises the unconditional `workspace_persisted_transcript` capability and serves: + +`GET /workspaces/:workspace/session/:id/transcript?cursor=&limit=<1..500>` + +The workspace selector resolves an exact registered workspace id first, then a URL-decoded portable absolute cwd. Unknown or unregistered selectors fail closed with `workspace_mismatch`. Trusted primary and secondary workspaces and untrusted secondary workspaces may read. An untrusted primary remains rejected to preserve the existing plural-route safe-mode boundary. + +Only active persisted JSONL is read. Archived sessions return `session_archived`; active/archive conflicts return `session_conflict`; missing sessions return 404. The response reuses `DaemonSessionTranscriptPage` and contains id-less `session_update` frames produced by the existing visible history projection. Raw JSONL and hidden system records are never returned. + +## Security boundary + +The route must not call a bridge method, start or preheat ACP, load settings, parse agents or skills, discover tools, run external commands, create a persisted cursor key, or write a route-specific daemon/debug log. Existing daemon-wide HTTP access logging and telemetry remain outside this route-specific no-write guarantee. + +Direct persisted transcript reads run with debug session logging suppressed. The archive coordinator holds a shared session lock across location checks, index construction, record reads, session-id validation and replay. Every returned record must carry the requested session id; a mismatch fails closed as a snapshot conflict. + +The direct replay context contains only a session id, an update sink, optional message rewriting, and optional cumulative usage. Tool display metadata falls back to persisted tool name and description when no full `Config` is present. No project-controlled tool registry is consulted. + +## Cursor lifecycle + +The core reader accepts an explicit cursor codec. Existing callers continue to use the file-backed codec. The daemon creates one random master key per process, derives a separate 32-byte HMAC key from workspace id and canonical cwd, and caches an in-memory codec per registered runtime. Cursors from the new route therefore expire on daemon restart and cannot be replayed across workspaces. + +The cursor freezes file identity, byte size, active leaf, replay position and replay state. Appends after page one do not change the snapshot. Delete, archive, truncate, replacement, leaf change, or session-id mismatch returns `transcript_snapshot_unavailable`. + +## Failure and race handling + +For the first page, the route checks active/archive location but still attempts the active reader when location is unknown so malformed active storage is not silently reported as missing. For cursor pages it validates only the frozen active snapshot. On `ENOENT`, a cursor page returns snapshot unavailable; a first page rechecks location to detect an archive race before returning missing. + +Replay is page-transactional at the protocol boundary: updates emitted before a replay conversion failure are returned with `partial: true`, a generic replay error and no next cursor. Pending tool calls and cumulative usage are carried in the signed cursor only after successful replay. + +## Compatibility and limits + +The old singular route, persisted cursor key and ACP error mapping remain intact. The new route keeps the existing default page size of 100, maximum of 500, 256 MiB snapshot cap, 32-entry/64 MiB index cache and five-minute cache lifetime. The first index scan remains linear in the frozen snapshot size. + +The TypeScript SDK exposes the method on `WorkspaceDaemonClient`. It forces native REST transport and has no ACP route mapping. Older daemons can be detected through the new capability and continue returning 404 for the route. + +## Deferred work + +Archived transcript reading, reverse/tail pagination, live follow, automatic full-history loading, worker-thread indexing, cross-restart cursors, and the Web Shell viewer are separate follow-ups. diff --git a/docs/developers/daemon/00-index.md b/docs/developers/daemon/00-index.md index 4c206b56401..0b45935df3a 100644 --- a/docs/developers/daemon/00-index.md +++ b/docs/developers/daemon/00-index.md @@ -84,21 +84,21 @@ Pick the path that matches your goal: Use these anchors when moving from the docs into the latest `main` code: -| Surface | Implementation anchors | Primary docs | -| ----------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- | -| Bootstrap and HTTP assembly | `packages/cli/src/serve/run-qwen-serve.ts`, `packages/cli/src/serve/server.ts`, `packages/cli/src/serve/routes/health-demo.ts`, `/demo` | [`02`](./02-serve-runtime.md), [`20`](./20-quickstart-operations.md) | -| ACP bridge and session multiplexing | `packages/acp-bridge/src/bridge.ts`, `packages/acp-bridge/src/bridgeTypes.ts`, `@qwen-code/acp-bridge` | [`03`](./03-acp-bridge.md), [`08`](./08-session-lifecycle.md) | -| Permission mediation | `packages/acp-bridge/src/permissionMediator.ts`, `fromLoopback: boolean`, `policy.*` | [`04`](./04-permission-mediation.md), [`12`](./12-auth-security.md) | -| MCP transport pool | `packages/core/src/tools/mcp-transport-pool.ts`, `mcp-pool-key.ts`, `pid-descendants.ts`, `session-mcp-view.ts`, `/mcp refresh`, `MCPCallInterruptedError` | [`05`](./05-mcp-transport-pool.md), [`06`](./06-mcp-budget-guardrails.md) | -| MCP budget guardrails | `packages/core/src/tools/mcp-workspace-budget.ts`, `ServeMcpBudgetStatusCell.scope`, `budgets[]` | [`06`](./06-mcp-budget-guardrails.md) | -| Workspace filesystem | `packages/cli/src/serve/fs/`, `assertTrustedForIntent(trusted, intent)`, `meta.matchedIgnore`, `includeIgnored` | [`07`](./07-workspace-filesystem.md) | -| Event schema and SSE writer | `packages/sdk-typescript/src/daemon/events.ts`, `packages/cli/src/serve/routes/sse-events.ts`, `formatSseFrame`, `packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.ts`, `ToolCallEmitter.resolveToolProvenance`, `tool_call.provenance`, `serverId` | [`09`](./09-event-schema.md), [`10`](./10-event-bus.md) | -| Event resync | `state_resync_required`, `awaitingResync`, `RESYNC_PASSTHROUGH_TYPES`, `asKnownDaemonEvent`, `unrecognizedKnownEventCount` | [`09`](./09-event-schema.md), [`10`](./10-event-bus.md) | -| Capabilities | `packages/cli/src/serve/capabilities.ts`, `mcp_server_restart_refused.reason`, `MCP_RESTART_REFUSED_REASONS.has` | [`11`](./11-capabilities-versioning.md) | -| Auth and device flow | `packages/cli/src/serve/auth.ts`, `packages/cli/src/serve/auth/device-flow.ts` | [`12`](./12-auth-security.md) | -| TypeScript SDK daemon client | `packages/sdk-typescript/src/daemon/{DaemonClient,DaemonSessionClient,DaemonAuthFlow,sse,events,types}.ts`, `MCP_RESTART_DEFAULT_TIMEOUT_MS` | [`13`](./13-sdk-daemon-client.md) | -| Shared UI transcript layer | `DaemonUiEventType`, `DaemonSessionProvider`, `packages/webui/src/daemon/` | [`13`](./13-sdk-daemon-client.md), [`14`](./14-cli-tui-adapter.md), [`../daemon-ui/README.md`](../daemon-ui/README.md) | -| Channels and IDE adapters | `packages/channels/`, `packages/vscode-ide-companion/src/services/daemonIdeConnection.ts` | [`15`](./15-channel-adapters.md), [`16`](./16-vscode-ide-adapter.md) | +| Surface | Implementation anchors | Primary docs | +| ----------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- | +| Bootstrap and HTTP assembly | `packages/cli/src/serve/run-qwen-serve.ts`, `packages/cli/src/serve/server.ts`, `packages/cli/src/serve/routes/health-demo.ts`, `/demo` | [`02`](./02-serve-runtime.md), [`20`](./20-quickstart-operations.md) | +| ACP bridge and session multiplexing | `packages/acp-bridge/src/bridge.ts`, `packages/acp-bridge/src/bridgeTypes.ts`, `@qwen-code/acp-bridge` | [`03`](./03-acp-bridge.md), [`08`](./08-session-lifecycle.md) | +| Permission mediation | `packages/acp-bridge/src/permissionMediator.ts`, `fromLoopback: boolean`, `policy.*` | [`04`](./04-permission-mediation.md), [`12`](./12-auth-security.md) | +| MCP transport pool | `packages/core/src/tools/mcp-transport-pool.ts`, `mcp-pool-key.ts`, `pid-descendants.ts`, `session-mcp-view.ts`, `/mcp refresh`, `MCPCallInterruptedError` | [`05`](./05-mcp-transport-pool.md), [`06`](./06-mcp-budget-guardrails.md) | +| MCP budget guardrails | `packages/core/src/tools/mcp-workspace-budget.ts`, `ServeMcpBudgetStatusCell.scope`, `budgets[]` | [`06`](./06-mcp-budget-guardrails.md) | +| Workspace filesystem | `packages/cli/src/serve/fs/`, `assertTrustedForIntent(trusted, intent)`, `meta.matchedIgnore`, `includeIgnored` | [`07`](./07-workspace-filesystem.md) | +| Event schema and SSE writer | `packages/sdk-typescript/src/daemon/events.ts`, `packages/cli/src/serve/routes/sse-events.ts`, `formatSseFrame`, `packages/cli/src/acp-integration/session/emitters/tool-call-emitter.ts`, `ToolCallEmitter.resolveToolProvenance`, `tool_call.provenance`, `serverId` | [`09`](./09-event-schema.md), [`10`](./10-event-bus.md) | +| Event resync | `state_resync_required`, `awaitingResync`, `RESYNC_PASSTHROUGH_TYPES`, `asKnownDaemonEvent`, `unrecognizedKnownEventCount` | [`09`](./09-event-schema.md), [`10`](./10-event-bus.md) | +| Capabilities | `packages/cli/src/serve/capabilities.ts`, `mcp_server_restart_refused.reason`, `MCP_RESTART_REFUSED_REASONS.has` | [`11`](./11-capabilities-versioning.md) | +| Auth and device flow | `packages/cli/src/serve/auth.ts`, `packages/cli/src/serve/auth/device-flow.ts` | [`12`](./12-auth-security.md) | +| TypeScript SDK daemon client | `packages/sdk-typescript/src/daemon/{DaemonClient,DaemonSessionClient,DaemonAuthFlow,sse,events,types}.ts`, `MCP_RESTART_DEFAULT_TIMEOUT_MS` | [`13`](./13-sdk-daemon-client.md) | +| Shared UI transcript layer | `DaemonUiEventType`, `DaemonSessionProvider`, `packages/webui/src/daemon/` | [`13`](./13-sdk-daemon-client.md), [`14`](./14-cli-tui-adapter.md), [`../daemon-ui/README.md`](../daemon-ui/README.md) | +| Channels and IDE adapters | `packages/channels/`, `packages/vscode-ide-companion/src/services/daemonIdeConnection.ts` | [`15`](./15-channel-adapters.md), [`16`](./16-vscode-ide-adapter.md) | ## What is intentionally out of scope diff --git a/docs/developers/daemon/09-event-schema.md b/docs/developers/daemon/09-event-schema.md index 3955e99117d..5dea9af2939 100644 --- a/docs/developers/daemon/09-event-schema.md +++ b/docs/developers/daemon/09-event-schema.md @@ -244,7 +244,7 @@ Events triggered by a request that carried a registered `X-Qwen-Client-Id` may s ## Tool-call `_meta` (provenance / serverId) -This is separate from envelope `_meta`: ACP `session/update` payloads can carry their own `_meta` in `event.data._meta`. `ToolCallEmitter` (`packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.ts`) stamps two fields on `emitStart`, `emitResult`, and `emitError`: +This is separate from envelope `_meta`: ACP `session/update` payloads can carry their own `_meta` in `event.data._meta`. `ToolCallEmitter` (`packages/cli/src/acp-integration/session/emitters/tool-call-emitter.ts`) stamps two fields on `emitStart`, `emitResult`, and `emitError`: | Field | Type | Resolution rule | | ------------ | ----------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | diff --git a/docs/developers/daemon/13-sdk-daemon-client.md b/docs/developers/daemon/13-sdk-daemon-client.md index 265856e6db7..8f4897214aa 100644 --- a/docs/developers/daemon/13-sdk-daemon-client.md +++ b/docs/developers/daemon/13-sdk-daemon-client.md @@ -344,6 +344,8 @@ On reconnect the daemon replays events with `id > lastSeenEventId` from its boun `history_truncated.fullTranscriptAvailable` is a boolean capability flag. When it is `true`, callers can page the full active persisted replay with `DaemonClient.getSessionTranscriptPage(sessionId, { cursor, limit })`; when it is `false`, clients should keep rendering the bounded replay normally. +When `workspace_persisted_transcript` is advertised, `client.workspaceById(workspaceId).getSessionTranscriptPage(sessionId, { cursor, limit })` reads the selected registered workspace without attaching to ACP. The workspace-qualified method always uses native REST even if the client has a replaceable transport; its cursor expires when the daemon restarts. + ### Seeding `lastEventId` at Construction Callers that persist the cursor across process restarts can seed it: diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 40f01cecdea..44964bd54cc 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -188,7 +188,7 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design 'workspace_hooks', 'session_hooks', 'workspace_extensions', 'session_branch', 'rate_limit', 'workspace_reload', 'multi_workspace_sessions', 'persistent_workspace_registration', - 'workspace_qualified_rest_core', + 'workspace_qualified_rest_core', 'workspace_persisted_transcript', 'client_mcp_over_ws', 'cdp_tunnel_over_ws', 'browser_automation_mcp'] ``` @@ -202,6 +202,8 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design `session_transcript` advertises `GET /session/:id/transcript`, a read-only paged replay view over the persisted active-session JSONL. It is separate from `/load`: it does not attach a client, seed the live EventBus, create a live session, or change the live replay window. Clients should use it when they need the complete on-disk transcript for a long session, and continue using `/load` only for bounded live replay during cold UI restore. +`workspace_persisted_transcript` advertises `GET /workspaces/:workspace/session/:id/transcript`, a daemon-local persisted-only pager that does not start ACP, query live bridge state, load settings, discover project capabilities, or create the legacy persisted cursor key. The tag is unconditional because trusted single-workspace primaries can use the plural route; per-workspace trust authorization is still evaluated on every request. Registered untrusted secondary workspaces may read, while an untrusted primary remains rejected. + `slow_client_warning` covers SSE backpressure behavior: (a) the daemon emits a `slow_client_warning` synthetic event-stream frame when a subscriber's live frame backlog or live serialized-byte backlog crosses 75% full, once per overflow episode (rearmed after both measurements drain below 37.5%); (b) `GET /session/:id/events` accepts a `?maxQueued=N` query param (range `[16, 2048]`) to pre-size the per-subscriber frame backlog for cold reconnects against a large replay ring. The serialized-byte cap is daemon-owned (default **2 MiB** per subscriber), live-only, and intentionally has no query parameter. The daemon-wide ring size is controlled by `--event-ring-size` (default **8000**, per #3803 §02). Old daemons silently lack the warning/query behavior — pre-flight this tag before opting in. `typed_event_schema` advertises daemon event payloads that match the SDK's `KnownDaemonEvent` schema. Older daemons may still stream compatible frames, but SDK clients should pre-flight this tag before assuming typed event coverage. @@ -214,7 +216,7 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design `session_archive` advertises the v1 directory-state archive API: `POST /sessions/archive`, `POST /sessions/unarchive`, and `GET /workspace/:id/sessions?archiveState=active|archived`. Archived sessions cannot be loaded or resumed until they are unarchived. -`workspace_qualified_rest_core` advertises plural core REST routes under `/workspaces/:workspace/...`. The selector resolves as exact workspace id first, then as a URL-encoded absolute cwd after canonicalization. On single-workspace daemons, `workspaces[]` is absent unless `multi_workspace_sessions` is also advertised, so clients use `capabilities.workspaceCwd` as the cwd selector. Trust status and trust request routes are available for registered untrusted workspaces; file read routes follow the existing filesystem read policy. Registered untrusted secondary workspaces also expose persisted-only session and session-group catalogs: these reads do not attach to a session, start ACP, or merge live bridge state. File writes, catalog mutations, and every other plural core route require a trusted workspace and return `403 { code: "untrusted_workspace" }` when the selected runtime is untrusted. An untrusted primary continues to receive that `403` from the plural catalog routes; legacy singular primary routes keep their existing compatibility behavior. This tag covers the core file, status, settings, permissions, trust, lifecycle, MCP control, tool toggle, memory, workspace agent CRUD, and session storage surfaces. It does not cover auth, voice, extensions, ACP/WebSocket transport, or channel-worker routing. Workspace trust is not an ACL: a client holding the daemon token can read every registered workspace surface allowed by this policy. +`workspace_qualified_rest_core` advertises plural core REST routes under `/workspaces/:workspace/...`. The selector resolves as exact workspace id first, then as a URL-encoded absolute cwd after canonicalization. On single-workspace daemons, `workspaces[]` is absent unless `multi_workspace_sessions` is also advertised, so clients use `capabilities.workspaceCwd` as the cwd selector. Trust status and trust request routes are available for registered untrusted workspaces; file read routes follow the existing filesystem read policy. Registered untrusted secondary workspaces also expose persisted-only session and session-group catalogs: these reads do not attach to a session, start ACP, or merge live bridge state. File writes, catalog mutations, and other plural core routes require a trusted workspace unless a separate capability explicitly defines a narrower read-only policy, such as `workspace_persisted_transcript`. An untrusted primary continues to receive `403 { code: "untrusted_workspace" }` from the plural catalog and transcript routes; legacy singular primary routes keep their existing compatibility behavior. This tag covers the core file, status, settings, permissions, trust, lifecycle, MCP control, tool toggle, memory, workspace agent CRUD, and session storage surfaces. It does not cover auth, voice, extensions, ACP/WebSocket transport, or channel-worker routing. Workspace trust is not an ACL: a client holding the daemon token can read every registered workspace surface allowed by this policy. `session_lsp` advertises `GET /session/:id/lsp`, the read-only structured LSP status snapshot for daemon clients. Older daemons return `404`; pre-flight this tag before exposing remote LSP status. @@ -585,6 +587,7 @@ Capability tags: - `session_tasks` → `GET /session/:id/tasks` - `session_status` → `GET /session/:id/status` - `session_transcript` → `GET /session/:id/transcript` +- `workspace_persisted_transcript` → `GET /workspaces/:workspace/session/:id/transcript` Common status cell: @@ -1449,7 +1452,7 @@ The first page freezes the current JSONL snapshot size. Later pages read only th To protect daemon memory and latency, snapshots above the transcript indexing cap fail before the daemon scans the JSONL. Clients receive `413 transcript_too_large` and should fall back to export/offline processing or ask the user to shorten/archive older history. -`partial: true` and `replayError` may appear if replay conversion fails after producing some frames. When the response also includes `nextCursor`, clients should continue paging with that cursor; it skips records remaining after the failure point on the partial page, so the assembled replay has a hole that `partial: true` marks. +`partial: true` and `replayError` may appear if replay conversion fails after producing some frames. Partial responses never include `nextCursor`, so clients cannot silently paginate past records that were not converted. **Errors:** @@ -1459,6 +1462,14 @@ To protect daemon memory and latency, snapshots above the transcript indexing ca - `409` — transcript snapshot is unavailable because the file was deleted, truncated, replaced, or archived after the cursor was issued; this also applies when preflight can no longer find the active file for a cursor request. - `413` — `transcript_too_large` when the frozen transcript snapshot exceeds the daemon indexing cap. +### `GET /workspaces/:workspace/session/:id/transcript` + +Return the same `DaemonSessionTranscriptPage` projection as the singular route from the selected registered workspace's active persisted JSONL. Pre-flight `workspace_persisted_transcript`; this capability is independent of `multi_workspace_sessions` and works for a trusted single-workspace primary selected by id or cwd. + +The selector and query parameters follow the existing plural workspace and transcript rules. Trusted primary and secondary runtimes and untrusted secondary runtimes may read. An untrusted primary returns `403 untrusted_workspace`. Archived content is not returned. + +Unlike the legacy singular route, this path is implemented entirely inside the daemon process. It does not call the workspace bridge, start ACP, load settings, parse project-defined agents or skills, or create/repair `session-transcript-cursor-key`. Tool frames use persisted tool names and descriptions without consulting the runtime tool registry. Its HMAC cursor key exists only in daemon memory, is isolated per workspace, and rotates on restart; a cursor from a previous daemon process returns `400 invalid_transcript_cursor`. + ### `POST /session/:id/resume` Restore a persisted ACP session by id WITHOUT replaying history through SSE. The model context is restored internally on the agent side (via `geminiClient.initialize` reading `config.getResumedSessionData`); the SSE stream stays clean for clients that already have history rendered. Pre-flight `caps.features.session_resume`; `unstable_session_resume` remains a deprecated compatibility alias for older clients. diff --git a/docs/users/qwen-serve.md b/docs/users/qwen-serve.md index ce8afc5d30d..496224311df 100644 --- a/docs/users/qwen-serve.md +++ b/docs/users/qwen-serve.md @@ -417,7 +417,7 @@ Both flags accept a positive integer in milliseconds; `0`, `NaN`, non-integer, o Pass `--workspace` more than once to register several non-overlapping workspaces in one `qwen serve` process. The first path is primary. Each registered workspace owns an isolated runtime boundary, while the daemon-wide listener, authentication policy, and total-session limit are shared. Trusted workspaces start their own `qwen --acp` child on demand. Requests may select a registered workspace by canonical `cwd`; requests that omit `cwd` use the primary workspace. Use one daemon per user or security principal; workspace trust is an execution gate, not an ACL. -An untrusted secondary workspace is visible in Web Shell as `untrusted` and `read-only`. It can be expanded to inspect the persisted session catalog, but it cannot be selected, opened, resumed, or used to create sessions. The REST API also exposes its persisted session-group catalog. Neither catalog includes transcript content or live runtime state, and catalog reads do not start an ACP child. Trust the workspace and restart the daemon before using execution or mutation features. An untrusted primary remains disabled in Web Shell. +An untrusted secondary workspace is visible in Web Shell as `untrusted` and `read-only`. It can be expanded to inspect the persisted session catalog, but it cannot yet be selected or opened in Web Shell, resumed, or used to create sessions. The REST API exposes its persisted session-group catalog and, when `workspace_persisted_transcript` is advertised, its active persisted transcript through the workspace-qualified pager. These reads do not include live runtime state or start an ACP child. Trust the workspace and restart the daemon before using execution or mutation features. An untrusted primary remains disabled in Web Shell. Use separate daemon processes when you need a smaller fault or security boundary, independent bearer tokens, quotas, audit boundaries, operating-system isolation, or independent resource supervision. Multi-workspace mode is intended for one operator hosting several repos; it is not a multi-tenant isolation boundary. A single daemon token authorizes every route the daemon exposes, including the allowed read-only catalog for all registered workspaces. @@ -429,11 +429,12 @@ To handle multiple **users** (each with their own quota, audit log, sandbox) or The daemon exposes ACP's `session/load` and resume flow over HTTP, plus a separate read-only transcript pager: -| Route | Use when | -| ----------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `POST /session/:id/load` | The client has **no** useful local history rendered (cold reconnect, picker-then-open). For a live session, the daemon returns and injects the current bounded replay snapshot window; if older replay was dropped, the snapshot begins with `history_truncated`. Capability tag: `session_load`. | -| `POST /session/:id/resume` | The client already has the turns on screen and only needs the daemon-side handle back. Model context is restored on the agent side without UI replay — the SSE stream stays clean. Capability tag: `session_resume` (`unstable_session_resume` remains a deprecated alias for older clients). | -| `GET /session/:id/transcript` | The client needs the complete active persisted transcript. It returns id-less replay frames in cursor pages and does not call `/load`, attach a client, seed the live EventBus, create a live session, or change the live replay window. Capability tag: `session_transcript`. | +| Route | Use when | +| --------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `POST /session/:id/load` | The client has **no** useful local history rendered (cold reconnect, picker-then-open). For a live session, the daemon returns and injects the current bounded replay snapshot window; if older replay was dropped, the snapshot begins with `history_truncated`. Capability tag: `session_load`. | +| `POST /session/:id/resume` | The client already has the turns on screen and only needs the daemon-side handle back. Model context is restored on the agent side without UI replay — the SSE stream stays clean. Capability tag: `session_resume` (`unstable_session_resume` remains a deprecated alias for older clients). | +| `GET /session/:id/transcript` | The client needs the complete active persisted transcript. It returns id-less replay frames in cursor pages and does not call `/load`, attach a client, seed the live EventBus, create a live session, or change the live replay window. Capability tag: `session_transcript`. | +| `GET /workspaces/:workspace/session/:id/transcript` | The client needs an active persisted transcript from a selected workspace without starting ACP or loading workspace settings. Registered untrusted secondary workspaces may use this read-only path. Capability tag: `workspace_persisted_transcript`. | For load and resume, the TypeScript SDK exposes static factories on `DaemonSessionClient`: @@ -463,9 +464,11 @@ For full persisted replay, page with `DaemonClient.getSessionTranscriptPage(sess curl "http://127.0.0.1:4170/session/$SESSION_ID/transcript?limit=100" ``` +For a registered workspace, use `client.workspaceById(workspaceId).getSessionTranscriptPage(sessionId, { cursor, limit })` or `/workspaces/:workspace/session/:id/transcript`. The workspace-qualified method always uses native REST even when the SDK client has a replaceable ACP transport. Its cursors are daemon-lifetime-only and must be restarted from page one after a daemon restart. + `limit` counts active chat records, not emitted replay frames; one record can produce several `session_update` events. The first response freezes the JSONL snapshot size and returns `nextCursor` while `hasMore` is true. Later pages ignore appends after page 1, but return `409` if the file is deleted, truncated, replaced, archived, or otherwise conflicts with the frozen cursor. Very large snapshots return `413 transcript_too_large` before indexing so the daemon does not scan unbounded transcript files on the request path. -For repeated cold-session transcript paging, set `--channel-idle-timeout-ms` to a positive value. With the default `0`, an idle workspace's ACP child — and the in-process transcript index cache it holds — is reaped after every page, so each page re-spawns the child and rebuilds the index by re-scanning the whole frozen prefix (`O(snapshotSize)` per page). A positive timeout keeps the child alive across the cursor walk so it reuses its cached transcript index and replay config. +For repeated paging through the legacy singular route, set `--channel-idle-timeout-ms` to a positive value. With the default `0`, an idle workspace's ACP child — and the in-process transcript index cache it holds — is reaped after every page, so each page re-spawns the child and rebuilds the index by re-scanning the whole frozen prefix (`O(snapshotSize)` per page). A positive timeout keeps the child alive across the cursor walk so it reuses its cached transcript index and replay config. The workspace-qualified persisted route never starts an ACP child and is unaffected by this timeout. Note: live-session history replay is bounded twice: by the SSE ring for `Last-Event-ID` reconnects and by `--compacted-replay-max-bytes` for the snapshot returned by `POST /session/:id/load`. Long histories with chatty turns can exceed either bound. The daemon surfaces snapshot truncation with `history_truncated`; use `/transcript` when you need the complete active persisted history. diff --git a/eslint.legacy-filenames.mjs b/eslint.legacy-filenames.mjs index 3480630a120..afce5ce5b37 100644 --- a/eslint.legacy-filenames.mjs +++ b/eslint.legacy-filenames.mjs @@ -42,7 +42,6 @@ export const legacyFilenames = [ 'backgroundWorkUtils', 'bareMode', 'baseController', - 'BaseEmitter', 'BaseJsonOutputAdapter', 'baseLlmClient', 'branchCommand', @@ -158,7 +157,6 @@ export const legacyFilenames = [ 'headlessSafetyWarnings', 'helpCommand', 'historyMapping', - 'HistoryReplayer', 'historyUtils', 'hookAggregator', 'hookController', @@ -399,7 +397,6 @@ export const legacyFilenames = [ 'todoWrite', 'tokenEstimation', 'tokenLimits', - 'ToolCallEmitter', 'toolHookTriggers', 'toolsCommand', 'toolUseSummary', diff --git a/integration-tests/cli/qwen-serve-routes.test.ts b/integration-tests/cli/qwen-serve-routes.test.ts index f5f269ab0b6..e3dbe24a98a 100644 --- a/integration-tests/cli/qwen-serve-routes.test.ts +++ b/integration-tests/cli/qwen-serve-routes.test.ts @@ -373,6 +373,7 @@ describe('qwen serve — capabilities envelope', () => { 'session_branch', 'workspace_reload', 'workspace_qualified_rest_core', + 'workspace_persisted_transcript', 'voice_transcribe', ]); }); diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index 7f7296a53fc..e70533c0dc8 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -537,7 +537,7 @@ type MockPendingToolCall = { const { mockHistoryPendingToolCalls } = vi.hoisted(() => ({ mockHistoryPendingToolCalls: vi.fn((): MockPendingToolCall[] => []), })); -vi.mock('./session/HistoryReplayer.js', () => ({ +vi.mock('./session/history-replayer.js', () => ({ HistoryReplayer: vi.fn().mockImplementation((context: unknown) => ({ replay: (messages: unknown, gaps: unknown) => mockHistoryReplay(context, messages, gaps), @@ -6995,7 +6995,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); - it('qwen/status/session/transcript preserves hasMore on replay errors', async () => { + it('qwen/status/session/transcript terminates pagination on replay errors', async () => { const settings = makeCoreSettings(); vi.mocked(loadCliConfig).mockResolvedValue({ ...makeInnerConfig(), @@ -7048,7 +7048,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { replayError?: string; }; - expect(result.hasMore).toBe(true); + expect(result.hasMore).toBe(false); // On a replay error the page is partial and must NOT hand back a cursor: // continuing would drop the un-replayed records and carry corrupted // pendingToolCalls forward into the next page. diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index a4d9a1de458..006345fd131 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -61,7 +61,6 @@ import { SessionTranscriptSnapshotUnavailableError, SessionTranscriptTooLargeError, encodeSessionTranscriptCursor, - type SessionTranscriptCursorState, subagentGenerator, redactUrlCredentials, computeUniqueBranchTitle, @@ -80,13 +79,11 @@ import { refreshMemoryInstruction, type AgentParams, type ApprovalMode, - type ChatRecord, type Config, type ConfigInitializeOptions, type DeviceAuthorizationData, type DiscoveredMCPPrompt, type DiscoveredMCPResource, - type HistoryGap, type HookConfig, type McpBudgetEvent, type McpBudgetMode, @@ -169,11 +166,7 @@ import { type PermissionRuleSet, } from '../config/permission-settings.js'; import { createLoadedSettingsAdapter } from '../config/loadedSettingsAdapter.js'; -import type { - ApprovalModeValue, - CumulativeUsage, - SessionContext, -} from './session/types.js'; +import type { ApprovalModeValue } from './session/types.js'; import { z } from 'zod'; import type { CliArgs } from '../config/config.js'; import { @@ -195,9 +188,11 @@ import { import { Session, buildAvailableCommandsSnapshot } from './session/Session.js'; import { buildSessionTasksStatus } from './session/tasksSnapshot.js'; import { - HistoryReplayer, - type PendingReplayToolCall, -} from './session/HistoryReplayer.js'; + collectHistoryReplayUpdates, + copyCumulativeUsage, + createReplayCumulativeUsage, + replayTranscriptRecordPage, +} from './session/history-replay-page.js'; import { formatAcpModelId, parseAcpBaseModelId, @@ -411,190 +406,6 @@ function isBulkLoadReplayRequest(params: LoadSessionRequest): boolean { return meta?.[LOAD_REPLAY_MODE_META_KEY] === LOAD_REPLAY_BULK_MODE; } -function createReplayCumulativeUsage(): CumulativeUsage { - return { - promptTokens: 0, - cachedTokens: 0, - candidateTokens: 0, - apiTimeMs: 0, - }; -} - -function copyCumulativeUsage( - target: CumulativeUsage, - source: CumulativeUsage, -): void { - target.promptTokens = source.promptTokens; - target.cachedTokens = source.cachedTokens; - target.candidateTokens = source.candidateTokens; - target.apiTimeMs = source.apiTimeMs; -} - -function isCumulativeUsage(value: unknown): value is CumulativeUsage { - if (!isObjectRecord(value)) return false; - return ( - typeof value['promptTokens'] === 'number' && - Number.isFinite(value['promptTokens']) && - typeof value['cachedTokens'] === 'number' && - Number.isFinite(value['cachedTokens']) && - typeof value['candidateTokens'] === 'number' && - Number.isFinite(value['candidateTokens']) && - typeof value['apiTimeMs'] === 'number' && - Number.isFinite(value['apiTimeMs']) - ); -} - -function isPendingReplayToolCall( - value: unknown, -): value is PendingReplayToolCall { - if (!isObjectRecord(value)) return false; - return ( - typeof value['callId'] === 'string' && - typeof value['toolName'] === 'string' && - (value['timestamp'] === undefined || - typeof value['timestamp'] === 'string') && - typeof value['recordId'] === 'string' - ); -} - -function parseTranscriptReplayState(replay: unknown): { - pendingToolCalls: PendingReplayToolCall[]; - cumulativeUsage: CumulativeUsage; -} { - if (!isObjectRecord(replay)) { - return { - pendingToolCalls: [], - cumulativeUsage: createReplayCumulativeUsage(), - }; - } - const rawPending = replay['pendingToolCalls']; - const pendingToolCalls = Array.isArray(rawPending) - ? rawPending.filter(isPendingReplayToolCall) - : []; - if ( - Array.isArray(rawPending) && - pendingToolCalls.length !== rawPending.length - ) { - // A cursor from a newer or corrupted daemon can carry pending tool calls - // whose shape no longer matches; drop them defensively but log it so an - // operator can tell this apart from a genuine "tool never completed". - const dropped = rawPending.length - pendingToolCalls.length; - debugLogger.warn( - `[transcript] replay state dropped ${dropped} of ${rawPending.length} malformed pending tool calls`, - ); - } - const cumulativeUsage = isCumulativeUsage(replay['cumulativeUsage']) - ? { ...replay['cumulativeUsage'] } - : createReplayCumulativeUsage(); - return { pendingToolCalls, cumulativeUsage }; -} - -async function collectHistoryReplayUpdates({ - sessionId, - config, - records, - gaps, - cumulativeUsage, -}: { - sessionId: string; - config: Config; - records: ChatRecord[]; - gaps?: HistoryGap[]; - cumulativeUsage: CumulativeUsage; -}): Promise<{ updates: SessionUpdate[]; replayError?: string }> { - const updates: SessionUpdate[] = []; - const replayContext: SessionContext = { - sessionId, - config, - sendUpdate: async (update) => { - updates.push(update); - }, - cumulativeUsage, - }; - - try { - await new HistoryReplayer(replayContext).replay(records, gaps); - } catch (error) { - const replayError = error instanceof Error ? error.message : String(error); - debugLogger.warn( - '[historyReplay] History replay failed for session %s (partial updates: %d):', - sessionId, - updates.length, - error, - ); - return { updates, replayError }; - } - - return { updates }; -} - -async function collectHistoryReplayUpdatesPage({ - sessionId, - config, - records, - gaps, - cumulativeUsage, - pendingToolCalls, - finalizeDangling, -}: { - sessionId: string; - config: Config; - records: ChatRecord[]; - gaps?: HistoryGap[]; - cumulativeUsage: CumulativeUsage; - pendingToolCalls: PendingReplayToolCall[]; - finalizeDangling: boolean; -}): Promise<{ - updates: SessionUpdate[]; - pendingToolCalls: PendingReplayToolCall[]; - replayError?: string; -}> { - const updates: SessionUpdate[] = []; - const replayContext: SessionContext = { - sessionId, - config, - sendUpdate: async (update) => { - updates.push(update); - }, - cumulativeUsage, - }; - const replayer = new HistoryReplayer(replayContext); - - try { - const state = await replayer.replayPage(records, { - pendingToolCalls, - finalizeDangling, - gaps, - }); - return { updates, pendingToolCalls: state.pendingToolCalls }; - } catch (error) { - debugLogger.warn( - '[historyReplay] Paged history replay failed for session %s (partial updates: %d):', - sessionId, - updates.length, - error, - ); - return { - updates, - pendingToolCalls: replayer.getPendingToolCalls(), - replayError: 'Replay conversion failed for this page', - }; - } -} - -function liftSessionUpdateTimestamps( - updates: SessionUpdate[], -): SessionUpdate[] { - return updates.map((update) => { - const record = update as Record; - const meta = record['_meta']; - const timestamp = isObjectRecord(meta) ? meta['timestamp'] : undefined; - return typeof timestamp === 'number' || typeof timestamp === 'string' - ? ({ ...record, timestamp } as unknown as SessionUpdate) - : update; - }); -} - function createHiddenWorkspaceMemoryConfig(config: Config): Config { return new Proxy(config, { get(target, prop) { @@ -3368,8 +3179,9 @@ class QwenAgent implements Agent { records, gaps: sessionData?.historyGaps, cumulativeUsage: replayUsage, + logger: debugLogger, }); - replayUpdates = liftSessionUpdateTimestamps(replay.updates); + replayUpdates = replay.updates; copyCumulativeUsage(session.cumulativeUsage, replayUsage); if (replay.replayError !== undefined) { replayEnvelope = { @@ -6027,48 +5839,28 @@ class QwenAgent implements Agent { ...(typeof rawLimit === 'number' ? { limit: rawLimit } : {}), }); const config = await this.getTranscriptReplayConfig(cwd, settings); - const replayState = parseTranscriptReplayState(page.replay); - const replay = await collectHistoryReplayUpdatesPage({ + const replay = await replayTranscriptRecordPage({ sessionId, + page, config, - records: page.records, - gaps: page.gaps, - cumulativeUsage: replayState.cumulativeUsage, - pendingToolCalls: replayState.pendingToolCalls, - finalizeDangling: !page.hasMore, + encodeCursor: (state) => + encodeSessionTranscriptCursor(state, cwd), + logger: debugLogger, }); - const updates = liftSessionUpdateTimestamps(replay.updates); - let nextCursor: string | undefined; - // On a mid-page replay error the page is partial: records after the - // failed record are dropped and pendingToolCalls reflect partial - // state. Withhold nextCursor so the client cannot paginate forward - // past the dropped records with a corrupted cursor — the page is - // already flagged partial + replayError below. - if (page.nextCursorState && replay.replayError === undefined) { - const nextCursorState: SessionTranscriptCursorState = { - ...page.nextCursorState, - replay: { - pendingToolCalls: replay.pendingToolCalls, - // Preserved for future replay emitters that need cumulative - // usage across page boundaries; current frames expose per-record - // usage metadata only. - cumulativeUsage: replayState.cumulativeUsage, - }, - }; - nextCursor = encodeSessionTranscriptCursor(nextCursorState, cwd); - } return { v: 1, sessionId, - events: updates.map((update) => ({ + events: replay.updates.map((update) => ({ v: 1, type: 'session_update', data: update, })), - ...(nextCursor !== undefined ? { nextCursor } : {}), - hasMore: page.hasMore, - startTime: page.startTime, - lastUpdated: page.lastUpdated, + ...(replay.nextCursor !== undefined + ? { nextCursor: replay.nextCursor } + : {}), + hasMore: replay.hasMore, + startTime: replay.startTime, + lastUpdated: replay.lastUpdated, ...(replay.replayError !== undefined ? { partial: true, replayError: replay.replayError } : {}), @@ -8029,10 +7821,11 @@ class QwenAgent implements Agent { records: sessionData.conversation.messages, gaps: sessionData.historyGaps, cumulativeUsage: createReplayCumulativeUsage(), + logger: debugLogger, }); return { - updates: liftSessionUpdateTimestamps(replay.updates), + updates: replay.updates, startTime: sessionData.conversation.startTime, lastUpdated: sessionData.conversation.lastUpdated, // Signal to the client that replay aborted partway so it doesn't diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index d3d04800bf4..6025c9f7381 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -207,8 +207,8 @@ import type { SessionContext, ToolCallStartParams, } from './types.js'; -import { HistoryReplayer } from './HistoryReplayer.js'; -import { ToolCallEmitter } from './emitters/ToolCallEmitter.js'; +import { HistoryReplayer } from './history-replayer.js'; +import { ToolCallEmitter } from './emitters/tool-call-emitter.js'; import { PlanEmitter } from './emitters/PlanEmitter.js'; import { MessageEmitter } from './emitters/MessageEmitter.js'; import { SubAgentTracker } from './SubAgentTracker.js'; diff --git a/packages/cli/src/acp-integration/session/SubAgentTracker.ts b/packages/cli/src/acp-integration/session/SubAgentTracker.ts index 380e2d554e7..7d4564d510c 100644 --- a/packages/cli/src/acp-integration/session/SubAgentTracker.ts +++ b/packages/cli/src/acp-integration/session/SubAgentTracker.ts @@ -22,7 +22,7 @@ import { } from '@qwen-code/qwen-code-core'; import { z } from 'zod'; import type { SessionContext } from './types.js'; -import { ToolCallEmitter } from './emitters/ToolCallEmitter.js'; +import { ToolCallEmitter } from './emitters/tool-call-emitter.js'; import { MessageEmitter } from './emitters/MessageEmitter.js'; import type { AgentSideConnection, diff --git a/packages/cli/src/acp-integration/session/emitters/MessageEmitter.ts b/packages/cli/src/acp-integration/session/emitters/MessageEmitter.ts index 43da4b4061d..dee1d87d67a 100644 --- a/packages/cli/src/acp-integration/session/emitters/MessageEmitter.ts +++ b/packages/cli/src/acp-integration/session/emitters/MessageEmitter.ts @@ -11,7 +11,7 @@ import { getActiveGoal, type GoalTerminalEvent, } from '@qwen-code/qwen-code-core'; -import { BaseEmitter } from './BaseEmitter.js'; +import { BaseEmitter } from './base-emitter.js'; import type { HistoryItemGoalStatus } from '../../../ui/types.js'; /** diff --git a/packages/cli/src/acp-integration/session/emitters/PlanEmitter.ts b/packages/cli/src/acp-integration/session/emitters/PlanEmitter.ts index 540203f7369..6777e651880 100644 --- a/packages/cli/src/acp-integration/session/emitters/PlanEmitter.ts +++ b/packages/cli/src/acp-integration/session/emitters/PlanEmitter.ts @@ -4,7 +4,7 @@ * SPDX-License-Identifier: Apache-2.0 */ -import { BaseEmitter } from './BaseEmitter.js'; +import { BaseEmitter } from './base-emitter.js'; import type { TodoItem } from '../types.js'; import type { PlanEntry } from '@agentclientprotocol/sdk'; diff --git a/packages/cli/src/acp-integration/session/emitters/BaseEmitter.ts b/packages/cli/src/acp-integration/session/emitters/base-emitter.ts similarity index 85% rename from packages/cli/src/acp-integration/session/emitters/BaseEmitter.ts rename to packages/cli/src/acp-integration/session/emitters/base-emitter.ts index fbd4d274c2e..1180ce1f1ad 100644 --- a/packages/cli/src/acp-integration/session/emitters/BaseEmitter.ts +++ b/packages/cli/src/acp-integration/session/emitters/base-emitter.ts @@ -4,7 +4,7 @@ * SPDX-License-Identifier: Apache-2.0 */ -import type { SessionContext } from '../types.js'; +import type { SessionEmitterContext } from '../types.js'; import type { SessionUpdate } from '@agentclientprotocol/sdk'; /** @@ -12,7 +12,7 @@ import type { SessionUpdate } from '@agentclientprotocol/sdk'; * Provides common functionality and access to session context. */ export abstract class BaseEmitter { - constructor(protected readonly ctx: SessionContext) {} + constructor(protected readonly ctx: SessionEmitterContext) {} /** * Converts an ISO timestamp string or epoch ms to epoch ms number. @@ -41,13 +41,6 @@ export abstract class BaseEmitter { return this.ctx.sendUpdate(update); } - /** - * Gets the session configuration. - */ - protected get config() { - return this.ctx.config; - } - /** * Gets the session ID. */ diff --git a/packages/cli/src/acp-integration/session/emitters/index.ts b/packages/cli/src/acp-integration/session/emitters/index.ts index f99a6dc7bf4..d47ec4690a2 100644 --- a/packages/cli/src/acp-integration/session/emitters/index.ts +++ b/packages/cli/src/acp-integration/session/emitters/index.ts @@ -4,7 +4,7 @@ * SPDX-License-Identifier: Apache-2.0 */ -export { BaseEmitter } from './BaseEmitter.js'; +export { BaseEmitter } from './base-emitter.js'; export { MessageEmitter } from './MessageEmitter.js'; export { PlanEmitter } from './PlanEmitter.js'; -export { ToolCallEmitter } from './ToolCallEmitter.js'; +export { ToolCallEmitter } from './tool-call-emitter.js'; diff --git a/packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.test.ts b/packages/cli/src/acp-integration/session/emitters/tool-call-emitter.test.ts similarity index 97% rename from packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.test.ts rename to packages/cli/src/acp-integration/session/emitters/tool-call-emitter.test.ts index 6306116673e..829f1e3d08a 100644 --- a/packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.test.ts +++ b/packages/cli/src/acp-integration/session/emitters/tool-call-emitter.test.ts @@ -5,8 +5,12 @@ */ import { describe, it, expect, vi, beforeEach } from 'vitest'; -import { ToolCallEmitter } from './ToolCallEmitter.js'; -import type { SessionContext, SubagentMeta } from '../types.js'; +import { ToolCallEmitter } from './tool-call-emitter.js'; +import type { + SessionContext, + SessionEmitterContext, + SubagentMeta, +} from '../types.js'; import type { Config, ToolRegistry, @@ -543,6 +547,25 @@ describe('ToolCallEmitter', () => { }); describe('resolveToolMetadata', () => { + it('uses persisted metadata without a Config in direct replay', () => { + const context: SessionEmitterContext = { + sessionId: 'persisted-session', + sendUpdate: vi.fn().mockResolvedValue(undefined), + }; + const configless = new ToolCallEmitter(context); + + expect( + configless.resolveToolMetadata('read_file', { + description: 'Read the persisted file', + path: '/untrusted/project/file.ts', + }), + ).toEqual({ + title: 'read_file: Read the persisted file', + locations: [], + kind: 'other', + }); + }); + it('should return defaults when tool not found', () => { const metadata = emitter.resolveToolMetadata('unknown_tool', { arg: 'value', diff --git a/packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.ts b/packages/cli/src/acp-integration/session/emitters/tool-call-emitter.ts similarity index 96% rename from packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.ts rename to packages/cli/src/acp-integration/session/emitters/tool-call-emitter.ts index f1c0a148631..8a0588fd630 100644 --- a/packages/cli/src/acp-integration/session/emitters/ToolCallEmitter.ts +++ b/packages/cli/src/acp-integration/session/emitters/tool-call-emitter.ts @@ -4,15 +4,16 @@ * SPDX-License-Identifier: Apache-2.0 */ -import { BaseEmitter } from './BaseEmitter.js'; +import { BaseEmitter } from './base-emitter.js'; import { PlanEmitter } from './PlanEmitter.js'; import type { - SessionContext, + SessionEmitterContext, ToolCallStartParams, ToolCallResultParams, ResolvedToolMetadata, SubagentMeta, } from '../types.js'; +import { hasFullSessionContext } from '../types.js'; import type { ToolCallContent, ToolCallLocation, @@ -56,7 +57,7 @@ const KIND_MAP: Record = { export class ToolCallEmitter extends BaseEmitter { private readonly planEmitter: PlanEmitter; - constructor(ctx: SessionContext) { + constructor(ctx: SessionEmitterContext) { super(ctx); this.planEmitter = new PlanEmitter(ctx); } @@ -297,7 +298,18 @@ export class ToolCallEmitter extends BaseEmitter { toolName: string, args?: Record, ): ResolvedToolMetadata { - const toolRegistry = this.config.getToolRegistry(); + if (!hasFullSessionContext(this.ctx)) { + const description = + typeof args?.['description'] === 'string' + ? args['description'].trim() + : ''; + return { + title: description ? `${toolName}: ${description}` : toolName, + locations: [], + kind: 'other', + }; + } + const toolRegistry = this.ctx.config.getToolRegistry(); const tool = toolRegistry.getTool(toolName); let title = tool?.displayName ?? toolName; diff --git a/packages/cli/src/acp-integration/session/history-replay-page.test.ts b/packages/cli/src/acp-integration/session/history-replay-page.test.ts new file mode 100644 index 00000000000..90fff918988 --- /dev/null +++ b/packages/cli/src/acp-integration/session/history-replay-page.test.ts @@ -0,0 +1,172 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { + ChatRecord, + SessionTranscriptCursorState, + SessionTranscriptRecordPage, +} from '@qwen-code/qwen-code-core'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { HistoryReplayer } from './history-replayer.js'; +import { + collectHistoryReplayUpdates, + createReplayCumulativeUsage, + replayTranscriptRecordPage, +} from './history-replay-page.js'; + +const SESSION_ID = '550e8400-e29b-41d4-a716-446655440000'; +const TIMESTAMP = '2026-07-12T00:00:00.000Z'; + +function userRecord(): ChatRecord { + return { + uuid: 'user-record', + parentUuid: null, + sessionId: SESSION_ID, + timestamp: TIMESTAMP, + type: 'user', + cwd: '/workspace', + version: '1.0.0', + message: { + role: 'user', + parts: [{ text: 'hello' }], + }, + }; +} + +function cursorState(): SessionTranscriptCursorState { + return { + v: 1, + sessionId: SESSION_ID, + fileIdentity: { dev: 1, ino: 2 }, + snapshotSize: 100, + position: 1, + leafUuid: 'next-record', + startTime: TIMESTAMP, + lastUpdated: TIMESTAMP, + }; +} + +function recordPage( + overrides: Partial = {}, +): SessionTranscriptRecordPage { + return { + sessionId: SESSION_ID, + filePath: '/workspace/chats/session.jsonl', + records: [], + gaps: [], + hasMore: false, + startTime: TIMESTAMP, + lastUpdated: TIMESTAMP, + ...overrides, + }; +} + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('history replay page', () => { + it('lifts record timestamps for bulk replay callers', async () => { + const result = await collectHistoryReplayUpdates({ + sessionId: SESSION_ID, + records: [userRecord()], + cumulativeUsage: createReplayCumulativeUsage(), + }); + + expect(result.updates).toEqual([ + expect.objectContaining({ + sessionUpdate: 'user_message_chunk', + timestamp: Date.parse(TIMESTAMP), + }), + ]); + }); + + it('filters malformed replay state before encoding the next cursor', async () => { + const logger = { warn: vi.fn() }; + const encodeCursor = vi.fn(() => 'next-cursor'); + const page = recordPage({ + hasMore: true, + nextCursorState: cursorState(), + replay: { + pendingToolCalls: [ + { + callId: 'call-1', + toolName: 'Read', + recordId: 'record-1', + }, + { callId: 1, toolName: 'invalid', recordId: 'record-2' }, + ], + cumulativeUsage: { + promptTokens: 1, + cachedTokens: 2, + candidateTokens: 3, + apiTimeMs: 4, + }, + }, + }); + + const result = await replayTranscriptRecordPage({ + sessionId: SESSION_ID, + page, + encodeCursor, + logger, + }); + + expect(result).toMatchObject({ + updates: [], + nextCursor: 'next-cursor', + hasMore: true, + }); + expect(logger.warn).toHaveBeenCalledWith( + expect.stringContaining('dropped 1 of 2 malformed pending tool calls'), + ); + expect(encodeCursor).toHaveBeenCalledWith( + expect.objectContaining({ + replay: { + pendingToolCalls: [ + { + callId: 'call-1', + toolName: 'Read', + recordId: 'record-1', + }, + ], + cumulativeUsage: { + promptTokens: 1, + cachedTokens: 2, + candidateTokens: 3, + apiTimeMs: 4, + }, + }, + }), + ); + }); + + it('terminates pagination when replay conversion fails', async () => { + vi.spyOn(HistoryReplayer.prototype, 'replayPage').mockRejectedValueOnce( + new Error('replay failed'), + ); + const encodeCursor = vi.fn(() => 'next-cursor'); + + const result = await replayTranscriptRecordPage({ + sessionId: SESSION_ID, + page: recordPage({ + records: [userRecord()], + hasMore: true, + nextCursorState: cursorState(), + }), + encodeCursor, + }); + + expect(result).toMatchObject({ + updates: [], + hasMore: false, + partial: true, + replayError: 'Replay conversion failed for this page', + }); + expect(result.nextCursor).toBeUndefined(); + expect(encodeCursor).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/cli/src/acp-integration/session/history-replay-page.ts b/packages/cli/src/acp-integration/session/history-replay-page.ts new file mode 100644 index 00000000000..5e408b793af --- /dev/null +++ b/packages/cli/src/acp-integration/session/history-replay-page.ts @@ -0,0 +1,235 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { + ChatRecord, + Config, + HistoryGap, + SessionTranscriptCursorState, + SessionTranscriptRecordPage, +} from '@qwen-code/qwen-code-core'; +import type { SessionUpdate } from '@agentclientprotocol/sdk'; +import { HistoryReplayer } from './history-replayer.js'; +import type { PendingReplayToolCall } from './history-replayer.js'; +import type { CumulativeUsage, SessionEmitterContext } from './types.js'; + +interface ReplayLogger { + warn(message: string, ...args: unknown[]): void; +} + +export function createReplayCumulativeUsage(): CumulativeUsage { + return { + promptTokens: 0, + cachedTokens: 0, + candidateTokens: 0, + apiTimeMs: 0, + }; +} + +export function copyCumulativeUsage( + target: CumulativeUsage, + source: CumulativeUsage, +): void { + target.promptTokens = source.promptTokens; + target.cachedTokens = source.cachedTokens; + target.candidateTokens = source.candidateTokens; + target.apiTimeMs = source.apiTimeMs; +} + +function isObjectRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function isCumulativeUsage(value: unknown): value is CumulativeUsage { + if (!isObjectRecord(value)) return false; + return ( + typeof value['promptTokens'] === 'number' && + Number.isFinite(value['promptTokens']) && + typeof value['cachedTokens'] === 'number' && + Number.isFinite(value['cachedTokens']) && + typeof value['candidateTokens'] === 'number' && + Number.isFinite(value['candidateTokens']) && + typeof value['apiTimeMs'] === 'number' && + Number.isFinite(value['apiTimeMs']) + ); +} + +function isPendingReplayToolCall( + value: unknown, +): value is PendingReplayToolCall { + if (!isObjectRecord(value)) return false; + return ( + typeof value['callId'] === 'string' && + typeof value['toolName'] === 'string' && + (value['timestamp'] === undefined || + typeof value['timestamp'] === 'string') && + typeof value['recordId'] === 'string' + ); +} + +function parseTranscriptReplayState( + replay: unknown, + logger?: ReplayLogger, +): { + pendingToolCalls: PendingReplayToolCall[]; + cumulativeUsage: CumulativeUsage; +} { + if (!isObjectRecord(replay)) { + return { + pendingToolCalls: [], + cumulativeUsage: createReplayCumulativeUsage(), + }; + } + const rawPending = replay['pendingToolCalls']; + const pendingToolCalls = Array.isArray(rawPending) + ? rawPending.filter(isPendingReplayToolCall) + : []; + if ( + logger && + Array.isArray(rawPending) && + pendingToolCalls.length !== rawPending.length + ) { + const dropped = rawPending.length - pendingToolCalls.length; + logger.warn( + `[transcript] replay state dropped ${dropped} of ${rawPending.length} malformed pending tool calls`, + ); + } + const cumulativeUsage = isCumulativeUsage(replay['cumulativeUsage']) + ? { ...replay['cumulativeUsage'] } + : createReplayCumulativeUsage(); + return { pendingToolCalls, cumulativeUsage }; +} + +function replayContext( + sessionId: string, + updates: SessionUpdate[], + cumulativeUsage: CumulativeUsage, + config?: Config, +): SessionEmitterContext { + return { + sessionId, + sendUpdate: async (update) => { + updates.push(update); + }, + cumulativeUsage, + ...(config ? { config } : {}), + }; +} + +export async function collectHistoryReplayUpdates({ + sessionId, + config, + records, + gaps, + cumulativeUsage, + logger, +}: { + sessionId: string; + config?: Config; + records: ChatRecord[]; + gaps?: HistoryGap[]; + cumulativeUsage: CumulativeUsage; + logger?: ReplayLogger; +}): Promise<{ updates: SessionUpdate[]; replayError?: string }> { + const updates: SessionUpdate[] = []; + try { + await new HistoryReplayer( + replayContext(sessionId, updates, cumulativeUsage, config), + ).replay(records, gaps); + } catch (error) { + const replayError = error instanceof Error ? error.message : String(error); + logger?.warn( + '[historyReplay] History replay failed for session %s (partial updates: %d):', + sessionId, + updates.length, + error, + ); + return { updates: liftSessionUpdateTimestamps(updates), replayError }; + } + return { updates: liftSessionUpdateTimestamps(updates) }; +} + +export function liftSessionUpdateTimestamps( + updates: SessionUpdate[], +): SessionUpdate[] { + return updates.map((update) => { + const record = update as Record; + const meta = record['_meta']; + const timestamp = isObjectRecord(meta) ? meta['timestamp'] : undefined; + return typeof timestamp === 'number' || typeof timestamp === 'string' + ? ({ ...record, timestamp } as unknown as SessionUpdate) + : update; + }); +} + +export interface ReplayedTranscriptPage { + updates: SessionUpdate[]; + nextCursor?: string; + hasMore: boolean; + startTime: string; + lastUpdated: string; + partial?: true; + replayError?: string; +} + +export async function replayTranscriptRecordPage({ + sessionId, + page, + config, + encodeCursor, + logger, +}: { + sessionId: string; + page: SessionTranscriptRecordPage; + config?: Config; + encodeCursor: (state: SessionTranscriptCursorState) => string; + logger?: ReplayLogger; +}): Promise { + const state = parseTranscriptReplayState(page.replay, logger); + const updates: SessionUpdate[] = []; + const replayer = new HistoryReplayer( + replayContext(sessionId, updates, state.cumulativeUsage, config), + ); + let pendingToolCalls: PendingReplayToolCall[]; + let replayError: string | undefined; + try { + const replayState = await replayer.replayPage(page.records, { + pendingToolCalls: state.pendingToolCalls, + finalizeDangling: !page.hasMore, + gaps: page.gaps, + }); + pendingToolCalls = replayState.pendingToolCalls; + } catch (error) { + logger?.warn( + '[historyReplay] Paged history replay failed for session %s (partial updates: %d):', + sessionId, + updates.length, + error, + ); + pendingToolCalls = replayer.getPendingToolCalls(); + replayError = 'Replay conversion failed for this page'; + } + + const nextCursor = + page.nextCursorState && replayError === undefined + ? encodeCursor({ + ...page.nextCursorState, + replay: { + pendingToolCalls, + cumulativeUsage: state.cumulativeUsage, + }, + }) + : undefined; + + return { + updates: liftSessionUpdateTimestamps(updates), + ...(nextCursor ? { nextCursor } : {}), + hasMore: replayError === undefined && page.hasMore, + startTime: page.startTime, + lastUpdated: page.lastUpdated, + ...(replayError ? { partial: true, replayError } : {}), + }; +} diff --git a/packages/cli/src/acp-integration/session/HistoryReplayer.test.ts b/packages/cli/src/acp-integration/session/history-replayer.test.ts similarity index 99% rename from packages/cli/src/acp-integration/session/HistoryReplayer.test.ts rename to packages/cli/src/acp-integration/session/history-replayer.test.ts index 5071c091b5e..256b1bb0a7e 100644 --- a/packages/cli/src/acp-integration/session/HistoryReplayer.test.ts +++ b/packages/cli/src/acp-integration/session/history-replayer.test.ts @@ -8,7 +8,7 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; import { HistoryReplayer, MISSING_TOOL_RESULT_MESSAGE, -} from './HistoryReplayer.js'; +} from './history-replayer.js'; import type { SessionContext } from './types.js'; import type { Config, diff --git a/packages/cli/src/acp-integration/session/HistoryReplayer.ts b/packages/cli/src/acp-integration/session/history-replayer.ts similarity index 98% rename from packages/cli/src/acp-integration/session/HistoryReplayer.ts rename to packages/cli/src/acp-integration/session/history-replayer.ts index abeeb3eac53..c982f03bf22 100644 --- a/packages/cli/src/acp-integration/session/HistoryReplayer.ts +++ b/packages/cli/src/acp-integration/session/history-replayer.ts @@ -15,9 +15,9 @@ import type { Content, GenerateContentResponseUsageMetadata, } from '@google/genai'; -import type { SessionContext } from './types.js'; +import type { SessionEmitterContext } from './types.js'; import { MessageEmitter } from './emitters/MessageEmitter.js'; -import { ToolCallEmitter } from './emitters/ToolCallEmitter.js'; +import { ToolCallEmitter } from './emitters/tool-call-emitter.js'; import { getToolResultCallId } from '../../utils/chat-record-tool-call-id.js'; import { formatHistoryGapNotice, @@ -53,7 +53,7 @@ export interface HistoryReplayPageState { * have appeared during the original session. */ export class HistoryReplayer { - private readonly ctx: SessionContext; + private readonly ctx: SessionEmitterContext; private readonly messageEmitter: MessageEmitter; private readonly toolCallEmitter: ToolCallEmitter; private readonly pendingReplayToolCalls = new Map< @@ -61,7 +61,7 @@ export class HistoryReplayer { PendingReplayToolCall >(); - constructor(ctx: SessionContext) { + constructor(ctx: SessionEmitterContext) { this.ctx = ctx; this.messageEmitter = new MessageEmitter(ctx); this.toolCallEmitter = new ToolCallEmitter(ctx); diff --git a/packages/cli/src/acp-integration/session/index.ts b/packages/cli/src/acp-integration/session/index.ts index ece0663389b..0628a4d431a 100644 --- a/packages/cli/src/acp-integration/session/index.ts +++ b/packages/cli/src/acp-integration/session/index.ts @@ -27,13 +27,13 @@ export type { } from './types.js'; // Emitters -export { BaseEmitter } from './emitters/BaseEmitter.js'; +export { BaseEmitter } from './emitters/base-emitter.js'; export { MessageEmitter } from './emitters/MessageEmitter.js'; export { PlanEmitter } from './emitters/PlanEmitter.js'; -export { ToolCallEmitter } from './emitters/ToolCallEmitter.js'; +export { ToolCallEmitter } from './emitters/tool-call-emitter.js'; // Components -export { HistoryReplayer } from './HistoryReplayer.js'; +export { HistoryReplayer } from './history-replayer.js'; export { SubAgentTracker } from './SubAgentTracker.js'; // Main Session class diff --git a/packages/cli/src/acp-integration/session/types.ts b/packages/cli/src/acp-integration/session/types.ts index 90b0878c78f..8fc17de4755 100644 --- a/packages/cli/src/acp-integration/session/types.ts +++ b/packages/cli/src/acp-integration/session/types.ts @@ -46,13 +46,8 @@ export interface CumulativeUsage { apiTimeMs: number; } -/** - * Session context shared across all emitters. - * Provides access to session state and configuration. - */ -export interface SessionContext extends SessionUpdateSender { +export interface SessionEmitterContext extends SessionUpdateSender { readonly sessionId: string; - readonly config: Config; /** Optional message rewrite middleware for ACP message transformation. * Installed after history replay to avoid rewriting historical messages. */ messageRewriter?: MessageRewriteMiddleware; @@ -65,6 +60,19 @@ export interface SessionContext extends SessionUpdateSender { readonly cumulativeUsage?: CumulativeUsage; } +/** + * Session context shared by live emitters that may resolve runtime metadata. + */ +export interface SessionContext extends SessionEmitterContext { + readonly config: Config; +} + +export function hasFullSessionContext( + context: SessionEmitterContext, +): context is SessionContext { + return 'config' in context; +} + /** * Subagent metadata for tracking parent tool call context. */ diff --git a/packages/cli/src/serve/capabilities.ts b/packages/cli/src/serve/capabilities.ts index d56c930a9e3..2fe493b866d 100644 --- a/packages/cli/src/serve/capabilities.ts +++ b/packages/cli/src/serve/capabilities.ts @@ -279,6 +279,10 @@ export const SERVE_CAPABILITY_REGISTRY = { // persistence. ACP/WebSocket, auth, voice, and extensions stay on their // existing primary-workspace routes in this phase. workspace_qualified_rest_core: { since: 'v1' }, + // Workspace-qualified, daemon-local persisted transcript paging. The tag is + // unconditional because the route also serves a trusted single-workspace + // primary; authorization is evaluated for the selected runtime per request. + workspace_persisted_transcript: { since: 'v1' }, // Workspace-qualified ACP transport (issue #6378 Phase 4): // `/workspaces/:workspace/acp` mounts a per-runtime ACP dispatcher (HTTP + // WebSocket) for each registered workspace, with per-runtime device-flow and diff --git a/packages/cli/src/serve/multi-workspace-sessions.test.ts b/packages/cli/src/serve/multi-workspace-sessions.test.ts index c2a619bca8d..245f49acc14 100644 --- a/packages/cli/src/serve/multi-workspace-sessions.test.ts +++ b/packages/cli/src/serve/multi-workspace-sessions.test.ts @@ -489,6 +489,7 @@ 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('workspace_persisted_transcript'); expect(res.body.workspaces).toEqual([ { id: 'primary-id', cwd: PRIMARY_CWD, primary: true, trusted: true }, { @@ -1458,6 +1459,310 @@ describe('multi-workspace session dispatch', () => { }); }); + it('chains untrusted transcript pages without bridge or cursor-key writes', async () => { + await withRuntimeDir(async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440270'; + await writeStoredSession({ + sessionId, + cwd: SECONDARY_CWD, + timestamp: '2026-07-08T00:00:00.000Z', + prompt: 'first page', + mtime: new Date('2026-07-08T00:00:00.000Z'), + }); + const transcriptPath = path.join( + new Storage(SECONDARY_CWD).getProjectDir(), + 'chats', + `${sessionId}.jsonl`, + ); + await fsp.appendFile( + transcriptPath, + [ + { + uuid: `${sessionId}-assistant-1`, + parentUuid: `${sessionId}-user-1`, + sessionId, + timestamp: '2026-07-08T00:01:00.000Z', + type: 'assistant', + message: { role: 'model', parts: [{ text: 'first answer' }] }, + cwd: SECONDARY_CWD, + }, + { + uuid: `${sessionId}-user-2`, + parentUuid: `${sessionId}-assistant-1`, + sessionId, + timestamp: '2026-07-08T00:02:00.000Z', + type: 'user', + message: { role: 'user', parts: [{ text: 'second question' }] }, + cwd: SECONDARY_CWD, + }, + { + uuid: `${sessionId}-assistant-2`, + parentUuid: `${sessionId}-user-2`, + sessionId, + timestamp: '2026-07-08T00:03:00.000Z', + type: 'assistant', + message: { role: 'model', parts: [{ text: 'second answer' }] }, + cwd: SECONDARY_CWD, + }, + ] + .map((record) => JSON.stringify(record)) + .join('\n') + '\n', + 'utf8', + ); + await writeStoredSession({ + sessionId, + cwd: PRIMARY_CWD, + timestamp: '2026-07-08T00:00:00.000Z', + prompt: 'same id in primary', + mtime: new Date('2026-07-08T00:00:00.000Z'), + }); + const chatsDir = path.dirname(transcriptPath); + const beforeEntries = await fsp.readdir(chatsDir); + const beforeContent = await fsp.readFile(transcriptPath); + const beforeMtimeMs = (await fsp.stat(transcriptPath)).mtimeMs; + const { app, primaryBridge, secondaryBridge } = makeHarness({ + secondaryTrusted: false, + }); + + const first = await request(app) + .get(`/workspaces/secondary-id/session/${sessionId}/transcript?limit=2`) + .set('Host', host()) + .expect(200); + expect( + first.body.events.map( + (event: { + data: { sessionUpdate: string; content?: { text?: string } }; + }) => [event.data.sessionUpdate, event.data.content?.text], + ), + ).toEqual([ + ['user_message_chunk', 'first page'], + ['agent_message_chunk', 'first answer'], + ]); + expect(first.body.hasMore).toBe(true); + expect(first.body.nextCursor).toEqual(expect.any(String)); + + const crossWorkspace = await request(app) + .get( + `/workspaces/primary-id/session/${sessionId}/transcript?cursor=${encodeURIComponent( + first.body.nextCursor as string, + )}`, + ) + .set('Host', host()); + expect(crossWorkspace.status).toBe(400); + expect(crossWorkspace.body.code).toBe('invalid_transcript_cursor'); + expect(crossWorkspace.body.sessionId).toBe(sessionId); + + const second = await request(app) + .get( + `/workspaces/${encodeURIComponent(SECONDARY_CWD)}/session/${sessionId}/transcript?limit=2&cursor=${encodeURIComponent( + first.body.nextCursor as string, + )}`, + ) + .set('Host', host()) + .expect(200); + expect( + second.body.events.map( + (event: { + data: { sessionUpdate: string; content?: { text?: string } }; + }) => [event.data.sessionUpdate, event.data.content?.text], + ), + ).toEqual([ + ['user_message_chunk', 'second question'], + ['agent_message_chunk', 'second answer'], + ]); + expect(second.body.hasMore).toBe(false); + expect(second.body.nextCursor).toBeUndefined(); + expect(primaryBridge.spawnCalls).toEqual([]); + expect(primaryBridge.restoreCalls).toEqual([]); + expect(secondaryBridge.spawnCalls).toEqual([]); + expect(secondaryBridge.restoreCalls).toEqual([]); + await expect( + fsp.stat( + path.join( + new Storage(SECONDARY_CWD).getProjectDir(), + 'session-transcript-cursor-key', + ), + ), + ).rejects.toMatchObject({ code: 'ENOENT' }); + expect(await fsp.readdir(chatsDir)).toEqual(beforeEntries); + expect(await fsp.readFile(transcriptPath)).toEqual(beforeContent); + expect((await fsp.stat(transcriptPath)).mtimeMs).toBe(beforeMtimeMs); + + const restarted = makeHarness({ secondaryTrusted: false }); + const expired = await request(restarted.app) + .get( + `/workspaces/secondary-id/session/${sessionId}/transcript?cursor=${encodeURIComponent( + first.body.nextCursor as string, + )}`, + ) + .set('Host', host()); + expect(expired.status).toBe(400); + expect(expired.body.code).toBe('invalid_transcript_cursor'); + expect(expired.body.sessionId).toBe(sessionId); + }); + }); + + it('fails closed for mismatched persisted transcript records', async () => { + await withRuntimeDir(async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440271'; + await writeStoredSession({ + sessionId, + cwd: SECONDARY_CWD, + timestamp: '2026-07-08T00:00:00.000Z', + prompt: 'wrong owner', + mtime: new Date('2026-07-08T00:00:00.000Z'), + }); + const transcriptPath = path.join( + new Storage(SECONDARY_CWD).getProjectDir(), + 'chats', + `${sessionId}.jsonl`, + ); + const content = await fsp.readFile(transcriptPath, 'utf8'); + await fsp.writeFile( + transcriptPath, + content.replace( + `"sessionId":"${sessionId}"`, + '"sessionId":"550e8400-e29b-41d4-a716-446655440999"', + ), + 'utf8', + ); + const { app } = makeHarness({ secondaryTrusted: false }); + + const res = await request(app) + .get(`/workspaces/secondary-id/session/${sessionId}/transcript`) + .set('Host', host()); + + expect(res.status).toBe(409); + expect(res.body.code).toBe('transcript_snapshot_unavailable'); + }); + }); + + it('suppresses file-backed debug logging during untrusted transcript reads', async () => { + await withRuntimeDir(async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440274'; + await writeStoredSession({ + sessionId, + cwd: SECONDARY_CWD, + timestamp: '2026-07-08T00:00:00.000Z', + prompt: 'no debug writes', + mtime: new Date('2026-07-08T00:00:00.000Z'), + }); + const previousDebugLogFile = process.env['QWEN_DEBUG_LOG_FILE']; + const debugSessionId = '550e8400-e29b-41d4-a716-446655440275'; + const debugLogPath = Storage.getDebugLogPath(debugSessionId); + process.env['QWEN_DEBUG_LOG_FILE'] = '1'; + resetDebugLoggingState(); + setDebugLogSession({ getSessionId: () => debugSessionId }); + try { + const { app } = makeHarness({ secondaryTrusted: false }); + await request(app) + .get(`/workspaces/secondary-id/session/${sessionId}/transcript`) + .set('Host', host()) + .expect(200); + await expect(fsp.stat(debugLogPath)).rejects.toMatchObject({ + code: 'ENOENT', + }); + } finally { + setDebugLogSession(null); + resetDebugLoggingState(); + if (previousDebugLogFile === undefined) { + delete process.env['QWEN_DEBUG_LOG_FILE']; + } else { + process.env['QWEN_DEBUG_LOG_FILE'] = previousDebugLogFile; + } + } + }); + }); + + it('keeps archived and untrusted-primary transcript boundaries', async () => { + await withRuntimeDir(async () => { + const sessionId = '550e8400-e29b-41d4-a716-446655440272'; + await writeStoredSession({ + sessionId, + cwd: SECONDARY_CWD, + timestamp: '2026-07-08T00:00:00.000Z', + prompt: 'archived', + mtime: new Date('2026-07-08T00:00:00.000Z'), + }); + await archiveStoredSession(SECONDARY_CWD, sessionId); + const secondary = makeHarness({ secondaryTrusted: false }); + const archived = await request(secondary.app) + .get(`/workspaces/secondary-id/session/${sessionId}/transcript`) + .set('Host', host()); + expect(archived.status).toBe(409); + expect(archived.body.code).toBe('session_archived'); + + const primary = makeHarness({ primaryTrusted: false }); + const forbidden = await request(primary.app) + .get(`/workspaces/primary-id/session/${sessionId}/transcript`) + .set('Host', host()); + expect(forbidden.status).toBe(403); + expect(forbidden.body.code).toBe('untrusted_workspace'); + }); + }); + + it('serves trusted runtimes and rejects missing or unknown transcript targets', async () => { + await withRuntimeDir(async () => { + const primarySessionId = '550e8400-e29b-41d4-a716-446655440276'; + const secondarySessionId = '550e8400-e29b-41d4-a716-446655440277'; + for (const [sessionId, cwd] of [ + [primarySessionId, PRIMARY_CWD], + [secondarySessionId, SECONDARY_CWD], + ] as const) { + await writeStoredSession({ + sessionId, + cwd, + timestamp: '2026-07-08T00:00:00.000Z', + prompt: `trusted ${sessionId}`, + mtime: new Date('2026-07-08T00:00:00.000Z'), + }); + } + const { app } = makeHarness(); + + await request(app) + .get(`/workspaces/primary-id/session/${primarySessionId}/transcript`) + .set('Host', host()) + .expect(200); + await request(app) + .get( + `/workspaces/secondary-id/session/${secondarySessionId}/transcript`, + ) + .set('Host', host()) + .expect(200); + + const missing = await request(app) + .get( + '/workspaces/secondary-id/session/550e8400-e29b-41d4-a716-446655440278/transcript', + ) + .set('Host', host()); + expect(missing.status).toBe(404); + + const unknown = await request(app) + .get( + `/workspaces/${encodeURIComponent(UNKNOWN_CWD)}/session/${secondarySessionId}/transcript`, + ) + .set('Host', host()); + expect(unknown.status).toBe(400); + expect(unknown.body.code).toBe('workspace_mismatch'); + + const unknownWithInvalidLimit = await request(app) + .get( + `/workspaces/${encodeURIComponent(UNKNOWN_CWD)}/session/${secondarySessionId}/transcript?limit=501`, + ) + .set('Host', host()); + expect(unknownWithInvalidLimit.status).toBe(400); + expect(unknownWithInvalidLimit.body.code).toBe('workspace_mismatch'); + + const invalidLimit = await request(app) + .get( + `/workspaces/secondary-id/session/${secondarySessionId}/transcript?limit=501`, + ) + .set('Host', host()); + expect(invalidLimit.status).toBe(400); + expect(invalidLimit.body.code).toBe('invalid_transcript_limit'); + }); + }); + it('returns empty untrusted catalogs without creating storage', async () => { await withRuntimeDir(async () => { const projectDir = new Storage(SECONDARY_CWD).getProjectDir(); diff --git a/packages/cli/src/serve/routes/session.ts b/packages/cli/src/serve/routes/session.ts index 8c3896d0108..f4f1822e73f 100644 --- a/packages/cli/src/serve/routes/session.ts +++ b/packages/cli/src/serve/routes/session.ts @@ -13,6 +13,9 @@ import { SessionService, SessionOrganizationError, SESSION_TRANSCRIPT_MAX_LIMIT, + SessionTranscriptCursorCodec, + SessionTranscriptReader, + SessionTranscriptSnapshotUnavailableError, addDaemonRequestAttribute, runWithoutDebugLogSession, type ApprovalMode, @@ -67,6 +70,7 @@ import { sessionExportFormatValues, } from '../server/session-export.js'; import { createSessionOrganizationService } from '../session-organization-helpers.js'; +import { replayTranscriptRecordPage } from '../../acp-integration/session/history-replay-page.js'; import { requireSessionRuntime } from './session-runtime.js'; import { resolveWorkspaceRuntimeFromParam, @@ -90,15 +94,15 @@ interface RegisterSessionRoutesDeps { languageCodes: string[]; } -function isReadOnlyWorkspaceCatalog(runtime: WorkspaceRuntime): boolean { +function isReadOnlyWorkspaceInspection(runtime: WorkspaceRuntime): boolean { return !runtime.primary && !runtime.trusted; } -function runCatalogReadWithWorkspaceLogPolicy( +function runWorkspaceInspectionWithLogPolicy( runtime: WorkspaceRuntime, read: () => Promise, ): Promise { - return isReadOnlyWorkspaceCatalog(runtime) + return isReadOnlyWorkspaceInspection(runtime) ? runWithoutDebugLogSession(read) : read(); } @@ -258,6 +262,30 @@ export function registerSessionRoutes( sessionShellCommandEnabled, } = deps; const LANGUAGE_CODES = deps.languageCodes; + const transcriptCursorMasterKey = crypto.randomBytes(32); + const transcriptCursorCodecs = new Map< + string, + SessionTranscriptCursorCodec + >(); + + const getTranscriptCursorCodec = ( + runtime: WorkspaceRuntime, + ): SessionTranscriptCursorCodec => { + const canonicalCwd = canonicalizeWorkspace(runtime.workspaceCwd); + const cacheKey = `${runtime.workspaceId}\0${canonicalCwd}`; + const cached = transcriptCursorCodecs.get(cacheKey); + if (cached) return cached; + const derivedKey = crypto.hkdfSync( + 'sha256', + transcriptCursorMasterKey, + Buffer.alloc(0), + Buffer.from(cacheKey, 'utf8'), + 32, + ); + const codec = new SessionTranscriptCursorCodec(new Uint8Array(derivedKey)); + transcriptCursorCodecs.set(cacheKey, codec); + return codec; + }; const logSessionRoutingFailure = ( route: string, @@ -1320,6 +1348,98 @@ export function registerSessionRoutes( } }); + app.get('/workspaces/:workspace/session/:id/transcript', async (req, res) => { + const route = 'GET /workspaces/:workspace/session/:id/transcript'; + const sessionId = requireSessionId(req, res); + if (sessionId === null) return; + const runtime = resolveWorkspaceRuntimeFromParam( + workspaceRegistry, + req, + res, + ); + if (!runtime) return; + if (!runtime.trusted && runtime.primary) { + sendUntrustedWorkspaceResponse(res, { + sessionId, + workspaceCwd: runtime.workspaceCwd, + workspaceId: runtime.workspaceId, + }); + return; + } + const limit = parseTranscriptLimitQuery(req.query['limit'], res); + if (limit === null) return; + const cursor = parseTranscriptCursorQuery(req.query['cursor'], res); + if (cursor === null) return; + + try { + const result = await runWithoutDebugLogSession(() => + archiveCoordinator.runSharedMany([sessionId], async () => { + const service = new SessionService(runtime.workspaceCwd); + if (cursor === undefined) { + await assertSessionLoadable(runtime.workspaceCwd, sessionId); + } + const codec = getTranscriptCursorCodec(runtime); + const reader = new SessionTranscriptReader( + runtime.workspaceCwd, + codec, + ); + let page; + try { + page = await reader.readPage(sessionId, { + ...(limit !== undefined ? { limit } : {}), + ...(cursor !== undefined ? { cursor } : {}), + }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { + throw error; + } + if (cursor !== undefined) { + throw new SessionTranscriptSnapshotUnavailableError(sessionId); + } + const location = await service.getSessionLocation(sessionId); + if (location === 'archived') { + throw new SessionArchivedError(sessionId); + } + if (location === 'conflict') { + throw new SessionConflictError(sessionId); + } + throw new SessionNotFoundError(sessionId); + } + if (page.records.some((record) => record.sessionId !== sessionId)) { + throw new SessionTranscriptSnapshotUnavailableError(sessionId); + } + const replay = await replayTranscriptRecordPage({ + sessionId, + page, + encodeCursor: (state) => codec.encode(state), + }); + return { + v: 1 as const, + sessionId, + events: replay.updates.map((update) => ({ + v: 1 as const, + type: 'session_update' as const, + data: update, + })), + ...(replay.nextCursor ? { nextCursor: replay.nextCursor } : {}), + hasMore: replay.hasMore, + startTime: replay.startTime, + lastUpdated: replay.lastUpdated, + ...(replay.partial + ? { + partial: true as const, + replayError: replay.replayError, + } + : {}), + }; + }), + ); + res.status(200).set('Cache-Control', 'no-store').json(result); + } catch (err) { + sendBridgeError(res, err, { route, sessionId }); + } + }); + app.get( '/session/:id/context', withOwnerReadSession( @@ -2139,7 +2259,7 @@ export function registerSessionRoutes( res .status(200) .json( - await runCatalogReadWithWorkspaceLogPolicy(runtime, () => + await runWorkspaceInspectionWithLogPolicy(runtime, () => createSessionOrganizationService(runtime.workspaceCwd).listGroups(), ), ); @@ -2228,7 +2348,7 @@ export function registerSessionRoutes( res .status(200) .json( - await runCatalogReadWithWorkspaceLogPolicy(runtime, () => + await runWorkspaceInspectionWithLogPolicy(runtime, () => createSessionOrganizationService(runtime.workspaceCwd).listGroups(), ), ); @@ -2322,7 +2442,7 @@ export function registerSessionRoutes( const runtime = resolveRuntimeForCatalogRoute(req, res, paramName, route); if (runtime === null) return; const key = runtime.workspaceCwd; - const readOnlySecondary = isReadOnlyWorkspaceCatalog(runtime); + const readOnlySecondary = isReadOnlyWorkspaceInspection(runtime); try { const cursor = typeof req.query['cursor'] === 'string' @@ -2425,7 +2545,7 @@ export function registerSessionRoutes( ); } const result = usePersisted - ? await runCatalogReadWithWorkspaceLogPolicy(runtime, () => + ? await runWorkspaceInspectionWithLogPolicy(runtime, () => listWorkspaceSessionsForResponse(runtime.bridge, key, options, { mergeLive: !readOnlySecondary, }), diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index db61af18ad8..bc707273630 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -309,6 +309,7 @@ const EXPECTED_STAGE1_FEATURES = [ 'workspace_extensions', 'session_branch', 'workspace_qualified_rest_core', + 'workspace_persisted_transcript', // Baseline (always advertised) — presence means the `/voice/stream` // endpoint exists; the WS errors if no voice model is configured. 'voice_transcribe', @@ -354,6 +355,7 @@ const EXPECTED_REGISTERED_FEATURES = [ f !== 'workspace_extensions' && f !== 'session_branch' && f !== 'workspace_qualified_rest_core' && + f !== 'workspace_persisted_transcript' && f !== 'voice_transcribe', ), 'workspace_settings', @@ -388,6 +390,7 @@ const EXPECTED_REGISTERED_FEATURES = [ 'multi_workspace_sessions', 'persistent_workspace_registration', 'workspace_qualified_rest_core', + 'workspace_persisted_transcript', 'workspace_qualified_acp', 'client_mcp_over_ws', 'cdp_tunnel_over_ws', diff --git a/packages/cli/src/serve/server/error-response.ts b/packages/cli/src/serve/server/error-response.ts index f08024017f3..c883a492a8f 100644 --- a/packages/cli/src/serve/server/error-response.ts +++ b/packages/cli/src/serve/server/error-response.ts @@ -6,8 +6,11 @@ import { emitDaemonLog, + InvalidSessionTranscriptCursorError, recordDaemonBridgeError, recordDaemonError, + SessionTranscriptSnapshotUnavailableError, + SessionTranscriptTooLargeError, TrustGateError, } from '@qwen-code/qwen-code-core'; import type { Response } from 'express'; @@ -148,6 +151,32 @@ export function sendBridgeError( ctx?: BridgeErrorContext, daemonLog?: DaemonLogger, ): void { + if (err instanceof InvalidSessionTranscriptCursorError) { + res.status(400).json({ + error: err.message, + code: 'invalid_transcript_cursor', + ...(ctx?.sessionId ? { sessionId: ctx.sessionId } : {}), + }); + return; + } + if (err instanceof SessionTranscriptSnapshotUnavailableError) { + res.status(409).json({ + error: err.message, + code: 'transcript_snapshot_unavailable', + ...(ctx?.sessionId ? { sessionId: ctx.sessionId } : {}), + }); + return; + } + if (err instanceof SessionTranscriptTooLargeError) { + res.status(413).json({ + error: err.message, + code: 'transcript_too_large', + sessionId: err.sessionId, + snapshotSize: err.snapshotSize, + maxBytes: err.maxBytes, + }); + return; + } if (err instanceof WorkspaceInitConflictError) { // The target file already exists with non- // whitespace content and the caller did not pass `force: true`. diff --git a/packages/cli/src/serve/server/telemetry.test.ts b/packages/cli/src/serve/server/telemetry.test.ts index 345bc57402e..31e1ebabec8 100644 --- a/packages/cli/src/serve/server/telemetry.test.ts +++ b/packages/cli/src/serve/server/telemetry.test.ts @@ -118,6 +118,28 @@ describe('daemonTelemetryMiddleware — recordRequest seam', () => { ); }); + it('attributes workspace transcript reads to the target workspace and session', () => { + const mw = daemonTelemetryMiddleware(() => '/workspace/secondary'); + const res = mockRes(200); + + mw( + mockReq('GET', '/workspaces/ws-secondary/session/session-1/transcript'), + res, + vi.fn() as unknown as NextFunction, + ); + res.emit('finish'); + + expect(coreMocks.withDaemonRequestSpan).toHaveBeenCalledWith( + expect.objectContaining({ + method: 'GET', + route: 'GET /workspaces/:workspace/session/:id/transcript', + sessionId: 'session-1', + workspaceHash: 'hash:/workspace/secondary', + }), + expect.any(Function), + ); + }); + it('normalizes plural workspace agent routes to stable route labels', () => { const mw = daemonTelemetryMiddleware(() => '/ws'); for (const [method, path, route] of [ diff --git a/packages/cli/src/serve/server/telemetry.ts b/packages/cli/src/serve/server/telemetry.ts index 7ea042b6b83..170abf9ce0c 100644 --- a/packages/cli/src/serve/server/telemetry.ts +++ b/packages/cli/src/serve/server/telemetry.ts @@ -117,6 +117,15 @@ export function resolveDaemonTelemetryRoute( if (req.method === 'GET' && /^\/workspaces\/[^/]+\/sessions$/.test(path)) { return { route: 'GET /workspace/:id/sessions' }; } + const workspaceTranscript = path.match( + /^\/workspaces\/[^/]+\/session\/([^/]+)\/transcript$/, + ); + if (workspaceTranscript?.[1] && req.method === 'GET') { + return { + route: 'GET /workspaces/:workspace/session/:id/transcript', + sessionId: workspaceTranscript[1], + }; + } const pluralWorkspacePrefix = /^\/workspaces\/[^/]+/; if (pluralWorkspacePrefix.test(path)) { const suffix = path.replace(pluralWorkspacePrefix, '/workspace'); diff --git a/packages/cli/src/ui/utils/export/collect.ts b/packages/cli/src/ui/utils/export/collect.ts index 59ed40c895f..cc046fb8f42 100644 --- a/packages/cli/src/ui/utils/export/collect.ts +++ b/packages/cli/src/ui/utils/export/collect.ts @@ -9,7 +9,7 @@ import type { ChatRecord, Config } from '@qwen-code/qwen-code-core'; import type { GenerateContentResponseUsageMetadata } from '@google/genai'; import type { SessionContext } from '../../../acp-integration/session/types.js'; import type { SessionUpdate, ToolCall } from '@agentclientprotocol/sdk'; -import { HistoryReplayer } from '../../../acp-integration/session/HistoryReplayer.js'; +import { HistoryReplayer } from '../../../acp-integration/session/history-replayer.js'; import { getExplicitToolResultCallId } from '../../../utils/chat-record-tool-call-id.js'; import type { ExportConfig, diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 4693d9264f3..8d6774133c5 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -241,6 +241,7 @@ export { SESSION_TRANSCRIPT_DEFAULT_LIMIT, SESSION_TRANSCRIPT_MAX_INDEX_BYTES, SESSION_TRANSCRIPT_MAX_LIMIT, + SessionTranscriptCursorCodec, SessionTranscriptReader, SessionTranscriptSnapshotUnavailableError, SessionTranscriptTooLargeError, diff --git a/packages/core/src/services/session-transcript-reader.test.ts b/packages/core/src/services/session-transcript-reader.test.ts index d7c0ebf584c..989155b0d3f 100644 --- a/packages/core/src/services/session-transcript-reader.test.ts +++ b/packages/core/src/services/session-transcript-reader.test.ts @@ -30,6 +30,7 @@ import { SESSION_TRANSCRIPT_MAX_LIMIT, resetSessionTranscriptIndexCacheForTest, setSessionTranscriptIndexCacheMaxBytesForTest, + SessionTranscriptCursorCodec, SessionTranscriptSnapshotUnavailableError, SessionTranscriptReader, } from './session-transcript-reader.js'; @@ -381,6 +382,62 @@ describe('SessionTranscriptReader', () => { expect(second.hasMore).toBe(true); }); + it('uses an injected in-memory codec without creating a cursor key file', async () => { + await writeRecords([ + record('u1', null, 'hello'), + record('a1', 'u1', 'reply'), + ]); + const key = Buffer.alloc(32, 7); + const codec = new SessionTranscriptCursorCodec(key); + key.fill(9); + const sameOriginalKey = new SessionTranscriptCursorCodec( + Buffer.alloc(32, 7), + ); + const reader = new SessionTranscriptReader(workspaceDir, codec); + const first = await reader.readPage(sessionId, { limit: 1 }); + const cursor = codec.encode(first.nextCursorState!); + expect(sameOriginalKey.decode(cursor).sessionId).toBe(sessionId); + const second = await reader.readPage(sessionId, { cursor, limit: 1 }); + + expect(second.records.map((item) => item.uuid)).toEqual(['a1']); + await expect( + fs.stat( + path.join( + new Storage(workspaceDir).getProjectDir(), + 'session-transcript-cursor-key', + ), + ), + ).rejects.toMatchObject({ code: 'ENOENT' }); + }); + + it('rejects in-memory cursors signed with another key or tampered', () => { + const first = new SessionTranscriptCursorCodec(Buffer.alloc(32, 1)); + const second = new SessionTranscriptCursorCodec(Buffer.alloc(32, 2)); + const cursor = first.encode({ + v: 1, + sessionId, + fileIdentity: { dev: 1, ino: 2 }, + snapshotSize: 3, + position: 1, + leafUuid: 'leaf', + startTime: 'start', + lastUpdated: 'end', + }); + + expect(() => second.decode(cursor)).toThrow( + InvalidSessionTranscriptCursorError, + ); + expect(() => first.decode(`${cursor.slice(0, -1)}A`)).toThrow( + InvalidSessionTranscriptCursorError, + ); + }); + + it('rejects an invalid in-memory cursor key length', () => { + expect(() => new SessionTranscriptCursorCodec(Buffer.alloc(31))).toThrow( + /must be 32 bytes/, + ); + }); + it('warns and replaces a corrupt persisted cursor signing key', async () => { const projectDir = new Storage(workspaceDir).getProjectDir(); const keyPath = path.join(projectDir, 'session-transcript-cursor-key'); diff --git a/packages/core/src/services/session-transcript-reader.ts b/packages/core/src/services/session-transcript-reader.ts index 9caf12b5f1d..31906cee153 100644 --- a/packages/core/src/services/session-transcript-reader.ts +++ b/packages/core/src/services/session-transcript-reader.ts @@ -240,25 +240,22 @@ function getCursorHmacKey(workspaceCwd: string): Buffer { return key; } -function signCursorPayload( +function signCursorPayloadWithKey( payload: Record, - workspaceCwd: string, + key: Uint8Array, ): string { return crypto - .createHmac('sha256', getCursorHmacKey(workspaceCwd)) + .createHmac('sha256', key) .update(JSON.stringify(payload)) .digest('base64url'); } -function hasValidCursorMac( +function hasValidCursorMacWithKey( payload: Record, mac: string, - workspaceCwd: string, + key: Uint8Array, ): boolean { - const expected = Buffer.from( - signCursorPayload(payload, workspaceCwd), - 'utf8', - ); + const expected = Buffer.from(signCursorPayloadWithKey(payload, key), 'utf8'); const actual = Buffer.from(mac, 'utf8'); return ( expected.length === actual.length && @@ -268,28 +265,21 @@ function hasValidCursorMac( function encodeCursorState( state: SessionTranscriptCursorState, - workspaceCwd: string, + key: Uint8Array, ): string { const payload = cursorPayload(state); return Buffer.from( JSON.stringify({ ...payload, - mac: signCursorPayload(payload, workspaceCwd), + mac: signCursorPayloadWithKey(payload, key), }), 'utf8', ).toString('base64url'); } -export function encodeSessionTranscriptCursor( - state: SessionTranscriptCursorState, - workspaceCwd: string, -): string { - return encodeCursorState(state, workspaceCwd); -} - -export function decodeSessionTranscriptCursor( +function decodeCursorState( cursor: string, - workspaceCwd: string, + key: Uint8Array, ): SessionTranscriptCursorState { try { const decoded = Buffer.from(cursor, 'base64url').toString('utf8'); @@ -328,7 +318,7 @@ export function decodeSessionTranscriptCursor( lastUpdated: parsed['lastUpdated'], ...(parsed['replay'] !== undefined ? { replay: parsed['replay'] } : {}), }; - if (!hasValidCursorMac(cursorPayload(state), parsed['mac'], workspaceCwd)) { + if (!hasValidCursorMacWithKey(cursorPayload(state), parsed['mac'], key)) { debugLogger.debug( `cursor decode failed: mac mismatch session=${state.sessionId} ` + `position=${state.position} snapshotSize=${state.snapshotSize}`, @@ -353,6 +343,41 @@ export function decodeSessionTranscriptCursor( } } +export class SessionTranscriptCursorCodec { + private readonly key: Buffer; + + constructor(key: Uint8Array) { + if (key.byteLength !== CURSOR_HMAC_KEY_BYTES) { + throw new RangeError( + `Transcript cursor signing key must be ${CURSOR_HMAC_KEY_BYTES} bytes`, + ); + } + this.key = Buffer.from(key); + } + + encode(state: SessionTranscriptCursorState): string { + return encodeCursorState(state, this.key); + } + + decode(cursor: string): SessionTranscriptCursorState { + return decodeCursorState(cursor, this.key); + } +} + +export function encodeSessionTranscriptCursor( + state: SessionTranscriptCursorState, + workspaceCwd: string, +): string { + return encodeCursorState(state, getCursorHmacKey(workspaceCwd)); +} + +export function decodeSessionTranscriptCursor( + cursor: string, + workspaceCwd: string, +): SessionTranscriptCursorState { + return decodeCursorState(cursor, getCursorHmacKey(workspaceCwd)); +} + function normalizeLimit(limit: number | undefined): number { if (limit === undefined) return SESSION_TRANSCRIPT_DEFAULT_LIMIT; if ( @@ -834,7 +859,10 @@ async function getCachedIndex(params: { export class SessionTranscriptReader { private readonly storage: Storage; - constructor(private readonly workspaceCwd: string) { + constructor( + private readonly workspaceCwd: string, + private readonly cursorCodec?: SessionTranscriptCursorCodec, + ) { this.storage = new Storage(workspaceCwd); } @@ -857,7 +885,8 @@ export class SessionTranscriptReader { const limit = normalizeLimit(options.limit); const cursor = options.cursor !== undefined - ? decodeSessionTranscriptCursor(options.cursor, this.workspaceCwd) + ? (this.cursorCodec?.decode(options.cursor) ?? + decodeSessionTranscriptCursor(options.cursor, this.workspaceCwd)) : undefined; if (cursor && cursor.sessionId !== sessionId) { debugLogger.debug( diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index f8b0c8079a7..f788d6eddf9 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -687,6 +687,7 @@ export class DaemonClient { body?: unknown; clientId?: string; timeoutMs?: number; + mode?: 'transport' | 'rest'; } = {}, ): Promise { return await this.jsonRequest( @@ -3621,6 +3622,28 @@ export class WorkspaceDaemonClient { return page.sessions; } + /** + * Read one page from an active persisted session transcript in this + * workspace. + * The daemon performs replay locally without attaching to the session or + * starting ACP. This method always uses native REST transport. + */ + getSessionTranscriptPage( + sessionId: string, + opts: DaemonSessionTranscriptPageOptions = {}, + ): Promise { + const query = new URLSearchParams(); + if (opts.cursor !== undefined) query.set('cursor', opts.cursor); + if (opts.limit !== undefined) query.set('limit', String(opts.limit)); + const suffix = query.size > 0 ? `?${query.toString()}` : ''; + return this.client.workspaceJsonRequest( + this.workspaceSelector, + `/session/${urlEncode(sessionId)}/transcript${suffix}`, + 'GET /workspaces/:workspace/session/:id/transcript', + { clientId: opts.clientId, mode: 'rest' }, + ); + } + listSessionGroups(): Promise { return this.get( '/session-groups', diff --git a/packages/sdk-typescript/test/unit/DaemonClient.test.ts b/packages/sdk-typescript/test/unit/DaemonClient.test.ts index b55fecc7ad9..dda95467a15 100644 --- a/packages/sdk-typescript/test/unit/DaemonClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonClient.test.ts @@ -4609,6 +4609,50 @@ describe('DaemonClient', () => { } }); + it('workspace transcript paging forces direct REST transport', async () => { + const body = { + v: 1 as const, + sessionId: 'session/1', + events: [], + nextCursor: 'next', + hasMore: true, + }; + const { fetch, calls } = recordingFetch(() => jsonResponse(200, body)); + const transportFetch = vi.fn(async () => { + throw new Error('replaceable transport must not be used'); + }); + const transport: DaemonTransport = { + type: 'acp-http', + supportsReplay: true, + connected: true, + fetch: transportFetch, + async *subscribeEvents() {}, + dispose() {}, + }; + const client = new DaemonClient({ + baseUrl: 'http://daemon', + fetch, + transport, + }); + + await expect( + client + .workspaceById('workspace/id') + .getSessionTranscriptPage('session/1', { + cursor: 'cur 1', + limit: 500, + clientId: 'client-1', + }), + ).resolves.toEqual(body); + + expect(transportFetch).not.toHaveBeenCalled(); + expect(calls[0]).toMatchObject({ + method: 'GET', + url: 'http://daemon/workspaces/workspace%2Fid/session/session%2F1/transcript?cursor=cur+1&limit=500', + }); + expect(calls[0]?.headers['x-qwen-client-id']).toBe('client-1'); + }); + it('workspaceByCwd deleteSessionGroup uses workspace-qualified group route', async () => { const { fetch, calls } = recordingFetch(() => jsonResponse(200, { deleted: true }),