diff --git a/docs/design/2026-08-19-prompt-terminal-ledger-design.md b/docs/design/2026-08-19-prompt-terminal-ledger-design.md new file mode 100644 index 00000000000..b9a5fef7d7a --- /dev/null +++ b/docs/design/2026-08-19-prompt-terminal-ledger-design.md @@ -0,0 +1,151 @@ +# Prompt Terminal Ledger + +## Problem + +The daemon's turn terminal events (`turn_complete` / `turn_error`) are synthesized by the ACP bridge from agent signals and published over SSE. They are never persisted. After a daemon restart, `POST /session/:id/load` performs a cold restore whose replay is produced by the agent subprocess re-reading the session JSONL transcript (`collectHistoryReplayUpdates` → `HistoryReplayer`), which emits only `session_update` chunk-class events — never terminal events. + +External orchestrators that mediate prompts by id therefore cannot resolve a prompt that was in flight when the daemon died: the replay contract "a terminal event for exactly this promptId" can never be satisfied, and the only safe answer is `unknown`. + +Two building blocks already exist and this design builds on them instead of adding new state machines: + +- `bridge.shutdown()` already flushes a formal error terminal (`flushPromptTerminals(entry, 'daemon_shutdown', ...)`) for every unfinished prompt through `publishPromptTerminal` — but only to memory and SSE, which die with the process. +- `detectTurnInterruption` (`packages/core/src/core/turn-interruption.ts`) is a pure read-only classifier over an api-history tail that distinguishes a clean tail from `interrupted_prompt` / `interrupted_turn`. + +## Goals + +- Persist one append-only ledger record per prompt admission and per prompt terminal, per session, so terminal facts survive daemon restarts. +- Reconstruct a terminal verdict for the dangling in-flight prompt of a cold-restored session by classifying the transcript tail, and expose the last 64 terminal records on the load response. +- Fail closed: when a verdict cannot be attributed, emit no terminal (the prompt stays `unknown`). +- Add no new core coupling to `acp-bridge` for the ledger; ledger writes are pure `node:fs` and best-effort. + +## Non-Goals + +- No backfill for sessions whose prompts predate the ledger (no ledger evidence → no reconstruction). +- No ledger truncation/compaction in this PR; records are tiny id/state lines and the sidecar follows the transcript lifecycle (archive/unarchive move it alongside). +- No reconstruction for queued-but-never-started prompts (see the attribution guard, the temporal-evidence check, and the multiple-dangling bail-out) and none for live-entry loads. +- No new SSE events and no change to replay semantics. + +## Design + +### Ledger format and location + +Each session owns a sidecar ledger next to its transcript: `/projects//chats/.ledger.jsonl`, resolved by `SessionService.getPromptLedgerPath(sessionId)`. The naming follows the existing `.worktree.json` sidecar convention and does not match `SESSION_FILE_PATTERN`, so directory scans ignore it. + +Records are single-line JSON objects, append-only: + +```json +{"v":1,"promptId":"...","state":"in_flight","at":1692000000000} +{"v":1,"promptId":"...","state":"in_flight","tailUuid":"rec-uuid","at":1692000000000} +{"v":1,"promptId":"...","terminal":"completed","stopReason":"stop","at":1692000000123} +{"v":1,"promptId":"...","terminal":"error","code":"daemon_shutdown","at":1692000000456} +{"v":1,"promptId":"...","terminal":"interrupted","code":"daemon_lost","at":1692000000789} +``` + +`terminal` is one of `completed | cancelled | error | interrupted`. `code` carries the flush origin (`daemon_shutdown`, `session_killed`, `channel_closed`, `session_closed`) or the normalized turn error code; `stopReason` carries the agent stop reason when present. `tailUuid` (in_flight only) is the dispatch marker: the uuid of the transcript's last record at admission, best-effort (absent when the transcript is missing/unreadable or for records written before the marker existed). + +The reader (`readPromptLedgerRecords`) tolerates torn tails: lines that fail structural validation are dropped, a missing file reads as empty. The writer (`appendPromptLedgerRecord`) seals a torn tail before appending: if the file is non-empty and its last byte is not `\n` (a crash mid-append), a newline is appended first so the next record cannot fuse with the torn fragment — without the seal, one torn tail plus one fresh append loses both records. + +`danglingInFlightPromptIds` reduces records per promptId (last write wins) and returns ids whose latest record is `in_flight`, in first-appearance order (admission order). + +### Write points (acp-bridge) + +All writes go through the module-level `appendPromptLedgerBestEffort` helper: any failure is logged via `writeStderrLine` and swallowed. A ledger problem must never block prompt execution or terminal flush. + +1. **Admission** — when `sendPrompt` pushes onto `pendingPromptList`, an `in_flight` record is appended synchronously (write-ahead: the in_flight fact must be on disk before the prompt can produce a terminal). Before the append, the bridge asks the sink for the transcript's last record uuid (`transcriptTailUuid`, best-effort) and stamps it as `tailUuid` — the dispatch marker that binds cold-load evidence to this admission (see step 5 below). +2. **Terminal** — immediately after the `terminalPublished` latch is set inside `publishPromptTerminal`, the terminal record is appended. Because all four `flushPromptTerminals` scenarios (`channel_closed`, `closeSession`/`session_closed`, `killSession`/`session_killed`, `bridge.shutdown`/`daemon_shutdown`) funnel unfinished prompts through `publishPromptTerminal`, one write point covers graceful shutdown too. `daemon_shutdown` persistence therefore precedes process exit without any extra sync path beyond the append being synchronous (`appendFileSync`). + +### Layering + +`acp-bridge` must gain no new core coupling for the ledger (the ledger module stays dependency-free beyond `node:fs`), and the bridge cannot know the serve-layer storage layout. `BridgeOptions` therefore gains an optional injected sink: + +```ts +promptLedger?: PromptLedgerSink; // { appendSync, transcriptTailUuid? } +``` + +`run-qwen-serve.ts` assembles it (`createPromptLedgerSink(workspaceCwd, sessionRuntimeBaseDir)`, backed by `SessionService.getPromptLedgerPath`) and injects it at the three bridge construction sites (primary, secondary, websocket-workspace — the latter skips live-conversation entries, which have no transcript to reconcile against). Reading, reconciliation, and HTTP exposure live in `packages/cli/src/serve/prompt-terminal-ledger.ts`, which may import core. + +### Cold-load reconciliation (lazy boot reconciliation) + +Hook: `restoreSessionHandler` (`POST /session/:id/load`), after `bridge.loadSession` resolves and before the response, only when `action === 'load' && !restored.attached && !restored.hasActivePrompt && provenance !== 'live-conversation'`. Concurrent loads of the same session already coalesce through the existing `inFlightRestores` map, so reconciliation runs at most once per cold restore. + +Algorithm (every step that cannot attribute the tail with confidence returns without appending — fail closed): + +1. Read the ledger; on failure return (no evidence, nothing appended). If there are no dangling in-flight prompt ids, return. +2. **Multiple dangling ids → return.** Under FIFO admission the visible transcript tail belongs to the _oldest_ running prompt, but with several prompts dangling the tail's owner cannot be verified (the queued ones never wrote a turn). Synthesizing a terminal for any of them — including the newest — could attribute an earlier prompt's turn to the wrong id, so they all stay `unknown` (omitted from `promptTerminals`). +3. Let `target` be the oldest dangling id. **Attribution guard**: walking the ledger forward, skip the `in_flight` records of prompts that have settled (a terminal record exists for them); the last remaining `in_flight` record must be `target`'s own admission. Skipping settled prompts matters for `[A if, B if, B cancelled]` (B queued, then cancelled while A still ran): the tail belongs to A even though B's `in_flight` line is the later record — a naive "last in_flight must match target" guard would wrongly veto A with B's settled admission. In `[if p1, if p2, term p1]` (valid interleave: p1 settled while p2 runs, daemon dies) the guard passes and p2 is attributed the tail. +4. Load the transcript (`loadSession`); failure or `undefined` → return. +5. **Attribution evidence** (five checks, each closing a concrete wrong-terminal class): + - **Dispatch marker**: when the target's `in_flight` record carries `tailUuid`, the projection must contain that record AND at least one visible write (non-`system`, with a `message`) after it. The transcript is append-only, so anything after the marker postdates admission — an identity/ordering check immune to clock skew. A marker absent from the projection, or present with no visible write beyond it, fails closed. Records without a marker (legacy, or capture failure) fall through to the temporal chain below. + - **Projection-consistent temporal evidence**: the last transcript write that actually enters the api history the verdict runs on (non-`system` records with a `message`, mirroring `SessionApiHistoryAccumulator`) must be strictly after `target`'s `in_flight` `at`. Measuring the raw stream instead would let evidence the verdict never sees — a post-admission `ui_telemetry`/`custom_title`/… record — pass the check for a prompt that never reached the model. An empty message list (or system-only tail) fails the same check. Equality is vetoed too: both clocks are 1 ms-granularity `Date.now()` reads, so a write landing in the admission millisecond cannot be attributed. + - **Compression fence**: a `chat_compression` record carrying a `compressedHistory` written at or after `target`'s admission → return. The accumulator swaps the whole history for the compressed snapshot, so the verdict's projection no longer carries `target`'s turn and nothing may be attributed. Marker-bearing admissions detect "after admission" by position (any compression record past the marker) so a backward clock step cannot hide the reset; marker-less admissions compare wall clocks. + - **FIFO evidence**: the visible tail must be strictly after every _other_ prompt's settled terminal `at` (same-millisecond equality is vetoed, for the same clock-collision reason as above). Under FIFO admission `target`'s turn can only start after every predecessor settled, so an older tail belongs to that predecessor's turn. This closes the queued-never-dispatched class (`[A if, B if(queued), A term]` — A's tail predates A's own terminal, so B gets nothing) and the stale-dangling class left by restore paths that skip reconciliation (a later prompt's completed tail predates its own terminal, so the stale prompt gets nothing). + - **Deadline fence**: if any other prompt's terminal carries `code: 'prompt_deadline_exceeded'` → return. The deadline path releases the FIFO while the wedged agent is explicitly allowed to keep streaming (DAEMON-003), so that terminal's timestamp does not fence its turn's writes: stale writes can postdate both the terminal and `target`'s admission and the temporal checks above cannot veto them. The veto is deliberately unconditional: the append-only ledger never expires records, so one deadline-exceeded prompt keeps the session fail-closed for every later dangling prompt (a missing terminal, never a wrong one). A recency bound cannot distinguish the stale case from the adjacent-overlap case without daemon-generation tracking, which the ledger does not carry. +6. Build the api history (`buildApiHistoryFromConversation`) and classify the last `TURN_INTERRUPTION_HISTORY_TAIL_COUNT` entries with `detectTurnInterruption`, then apply the **id-less tool-call guard**: when the verdict is `none` but the api-history tail's last entry is a model turn holding any `functionCall` part (with or without an id), upgrade to interrupted — `detectTurnInterruption` ignores id-less functionCalls because they cannot be paired on the wire, but reconciliation needs no wire pairing; a model tail holding a tool call means the daemon died mid tool-run. + - `none` (clean tail) → append `{"terminal":"completed","stopReason":"reconstructed_from_transcript"}`. + - `interrupted_prompt` / `interrupted_turn` / upgraded tool-call guard → append `{"terminal":"interrupted","code":"daemon_lost"}`. + - transcript unreadable or history undefined → append nothing (fail closed). +7. **TOCTOU fence**: re-read the ledger immediately before the append; any change since the step-1 snapshot (the ledger is append-only, so an unchanged record count proves it) → return. A prompt admitted while `loadSession` ran appends its `in_flight` after the snapshot, and the visible tail the verdict was computed from may now belong to it — the verdict must not be stamped onto the old dangling id. +8. Append best-effort; an append failure leaves the prompt `unknown`. + +**Residual attribution risk.** The dispatch marker binds evidence to the target's admission by ORDERING, not by OWNERSHIP: transcript records carry no prompt or writer identity, so any visible write landing after the marker passes the guard regardless of which writer produced it. Ordering-breakage classes are closed wherever the marker is present (a recordless predecessor's tail predates the successor's marker, a backward clock step cannot fake record order, a system-reminder-only tail contains no visible write beyond the marker), but two ownership classes survive even with a marker: + +- **Recordless predecessor with continued writes**: a predecessor whose best-effort ledger appends were swallowed leaves no records while its turn keeps writing past the queued successor's marker (reachable only under compound failure: swallowed appends plus post-settle streaming, e.g. the DAEMON-003 deadline wedge whose terminal append was also swallowed). +- **Ledger-less cross-client writer**: serve and the interactive CLI share the `/projects//chats` tree, and `/resume` lists serve-created sessions without a provenance filter; an interactive turn writes transcript records but no ledger records, invisible to every guard. + +Closing ownership requires binding transcript records to a writer identity (a transcript-record schema change, tracked in https://github.com/QwenLM/qwen-code/issues/9483); until then these entrances stay fail-open by design limitation, not by evidence. Marker-less admissions — legacy `in_flight` records written before the marker existed, marker capture failure (unreadable transcript, or a final record larger than the 64 KiB tail window) — additionally fall back to the temporal evidence chain and its documented classes (recordless-predecessor inheritance, backward clock steps, non-model-record tails). Ledger records written before this change never gain a marker retroactively; they stay on the temporal chain permanently. + +### Load response + +The serve-layer response type extends `BridgeRestoredSession` with an optional `promptTerminals` array (the trailing 64 terminal records, including reconciliation output). The bridge-level `BridgeRestoredSession` is untouched: the field is serve-layer evidence, so its type lives in the serve layer. When the ledger has no terminal records the field is omitted entirely. + +## Concurrency and idempotence + +- Ledger appends are single-line and synchronous; concurrent writers on one session are serialized by the OS append path and the reader's last-write-wins reduction absorbs duplicates. +- Reconciliation appends only when a dangling id exists, so a second load of the same session finds no dangling id and appends nothing (persisted verdict, single flight via `inFlightRestores`). +- A terminal record for a prompt that already has one is harmless (reduction keeps the latest), though the `terminalPublished` latch makes bridge duplicates impossible. +- `archiveSessions` / `unarchiveSessions` move the sidecar alongside the transcript via `moveLedgerSidecar` (warn-only on failure, both directions log the full source and destination paths). When the destination already exists (a partially completed earlier archive cycle), the source is not clobbered and the move does not wedge: the source contents are appended to the destination (append-only JSONL, write order preserved) and the source is unlinked — merge semantics instead of a permanent split. +- `removeSessionFiles` deletes the ledger in both states (active and archived) alongside the worktree sidecars, so removing a session leaves no orphan evidence. +- Insight and usage scans exclude the sidecar: `DataProcessor.scanChatFiles` and `usageHistoryService.rebuildFromSessionJsonl` select `.jsonl` files but reject `.ledger.jsonl` — the ledger is not a transcript and must never be parsed as chat records or usage evidence. + +### Ledger file lifecycle + +The sidecar follows the transcript through every state transition: created on first admission (best-effort), moved alongside on archive/unarchive (merge semantics on collision), and deleted in both states on session removal. All paths are derived from one helper (`getPromptLedgerPathForState`) so no call site hand-assembles the file name. + +## Privacy boundary + +Records contain only `v`, `promptId`, `state`/`terminal`, `code`, `stopReason`, `at`. No prompt text, user content, tool input/output, or file paths are ever written. The ledger inherits the transcript directory's permissions. + +## Compatibility matrix + +| Daemon | Client | Behavior | +| ------ | ------ | ---------------------------------------------------------------------------- | +| new | new | Cold load returns `promptTerminals`; orchestrators resolve dangling prompts. | +| new | old | Client ignores the unknown field; behavior identical to today. | +| old | new | No ledger file → field omitted; client falls back to `unknown` as today. | +| old | old | Unchanged. | + +## Fail-closed invariants + +- No verdict is ever synthesized without ledger evidence of an in-flight admission. +- A verdict requires both a readable transcript tail and a passing attribution guard. +- When the admission carries a dispatch marker, at least one visible write must land beyond it (ordering-based attribution, immune to clock skew); a missing marker or no write beyond it appends nothing. +- Multiple dangling prompts never receive a synthesized terminal; their tails cannot be attributed. +- The temporal evidence is measured on the same projection the verdict uses: the last write that enters the api history must be strictly after the target's admission (same-millisecond equality is vetoed — both clocks are 1 ms-granularity `Date.now()` reads), otherwise the tail belongs to an earlier turn and nothing is appended. +- A compression checkpoint written at or after the target's admission voids the evidence chain (nothing appended). +- The visible tail must be strictly after every other prompt's settled terminal (FIFO evidence); otherwise it belongs to that prompt's turn. +- A `prompt_deadline_exceeded` terminal of another prompt voids the evidence chain: the deadline path releases the FIFO while the wedged agent may keep writing, so that terminal does not fence its turn's writes. The veto is unconditional and permanent (missing terminals over wrong ones). +- The verdict is re-checked against the ledger immediately before the append (TOCTOU fence): any record landed during the reconciliation window voids the verdict. +- Residual ordering-breakage classes (recordless predecessors, backward clock steps, non-model-record tails) survive only on marker-less admissions (legacy records, capture failure); they are documented under "Residual attribution risk" and can only degrade attribution quality under compound failures. Ownership classes (recordless predecessor with continued writes, ledger-less cross-client writers) survive even with a marker: transcript records carry no writer identity, and closing them requires a transcript-record schema change (see "Residual attribution risk"). +- The ledger sidecar is created owner-only (`0o600`), matching the transcript's protection; it is never created with umask-default permissions. +- A model tail holding any tool call (id or not) is treated as interrupted, never as a clean completion. +- Everything downstream of the guard (ledger read failure, transcript read failure, append failure) degrades to "no terminal emitted", never to a wrong terminal. +- Ledger write failures never affect prompt execution or shutdown flush; ledger move failures never block archive/unarchive (warn-only). + +## Verification Plan + +- Unit-test the ledger module: append/read round-trip, torn-tail tolerance, torn-tail sealing (a sealed fragment cannot fuse with the next appended record), dangling reduction, recent-terminal windowing. +- Unit-test bridge write points with the existing FakeAgent harness: in_flight on admission, terminal on completion, flush on `daemon_shutdown`, in_flight recorded for queued admissions and both prompts flushed on shutdown, best-effort failure containment. +- Unit-test reconciliation branches with a real `SessionService` fixture: clean tail → completed, `interrupted_prompt`/`interrupted_turn` → interrupted, id-less functionCall tail → interrupted, missing transcript → fail closed, no dangling → no-op, multiple dangling → fail closed (nothing appended), settled-then-queued interleave (`[A if, B if, B cancelled]`) → A attributed, valid interleave (`[if p1, if p2, term p1]` on a real time axis) → p2 attributed, stale tail (last transcript write predates the admission) → fail closed, queued-never-dispatched (`[A if, B if, A term]` with A's terminal postdating its turn) → fail closed, stale dangling behind a later settled prompt → fail closed, post-admission compression checkpoint → fail closed, post-admission system-record-only tail → fail closed, idempotence. +- Unit-test the sidecar lifecycle in `sessionService`: archive/unarchive move the ledger, move failure is warn-only, destination-exists merges instead of wedging, session removal deletes both states, insight/usage scans skip `.ledger.jsonl`. +- Route-level test through `POST /session/:id/load`: field presence, omission without ledger, attached loads skip reconciliation, active prompts skip reconciliation, resume responses stay free of `promptTerminals`. +- Final verification on root `npm run build` and `npm run typecheck`. diff --git a/packages/acp-bridge/package.json b/packages/acp-bridge/package.json index 54e8416b8b0..8a281910cce 100644 --- a/packages/acp-bridge/package.json +++ b/packages/acp-bridge/package.json @@ -67,6 +67,10 @@ "types": "./dist/bridgeOptions.d.ts", "import": "./dist/bridgeOptions.js" }, + "./promptLedger": { + "types": "./dist/prompt-ledger.d.ts", + "import": "./dist/prompt-ledger.js" + }, "./sessionRestoreTimeout": { "types": "./dist/session-restore-timeout.d.ts", "import": "./dist/session-restore-timeout.js" diff --git a/packages/acp-bridge/src/bridge-prompt-ledger.test.ts b/packages/acp-bridge/src/bridge-prompt-ledger.test.ts new file mode 100644 index 00000000000..f83c7dda805 --- /dev/null +++ b/packages/acp-bridge/src/bridge-prompt-ledger.test.ts @@ -0,0 +1,309 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, expect, it } from 'vitest'; +import { makeBridge, makeChannel, WS_A } from './internal/testUtils.js'; +import type { PromptLedgerSink } from './bridgeOptions.js'; +import type { PromptLedgerRecord } from './prompt-ledger.js'; + +function recordingLedger(): { + records: PromptLedgerRecord[]; + sink: PromptLedgerSink; +} { + const records: PromptLedgerRecord[] = []; + return { + records, + sink: { + appendSync: (_sessionId, record) => { + records.push(record); + }, + }, + }; +} + +function terminalRecords(records: readonly PromptLedgerRecord[]) { + return records.filter( + (record): record is Extract => + 'terminal' in record, + ); +} + +describe('bridge prompt terminal ledger writes', () => { + it('appends in_flight at admission and completed at settle', async () => { + const handle = makeChannel(); + const ledger = recordingLedger(); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + promptLedger: ledger.sink, + }); + try { + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const running = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'hello' }], + }, + undefined, + { promptId: 'p-ledger-1' }, + ); + const inFlight = ledger.records.filter( + (record) => !('terminal' in record), + ); + expect(inFlight).toHaveLength(1); + expect(inFlight[0]?.promptId).toBe('p-ledger-1'); + + const result = await running; + expect(result.stopReason).toBe('end_turn'); + expect(terminalRecords(ledger.records)).toEqual([ + { + v: 1, + promptId: 'p-ledger-1', + terminal: 'completed', + stopReason: 'end_turn', + at: expect.any(Number), + }, + ]); + } finally { + await bridge.shutdown(); + } + }); + + it('stamps the dispatch marker from the sink into the in_flight record', async () => { + const handle = makeChannel(); + const records: PromptLedgerRecord[] = []; + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + promptLedger: { + appendSync: (_sessionId, record) => { + records.push(record); + }, + transcriptTailUuid: () => 'tail-at-admission', + }, + }); + try { + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'hello' }], + }, + undefined, + { promptId: 'p-marker' }, + ); + expect(records).toContainEqual({ + v: 1, + promptId: 'p-marker', + state: 'in_flight', + tailUuid: 'tail-at-admission', + at: expect.any(Number), + }); + } finally { + await bridge.shutdown(); + } + }); + + it('keeps admitting when the dispatch marker lookup fails', async () => { + const handle = makeChannel(); + const records: PromptLedgerRecord[] = []; + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + promptLedger: { + appendSync: (_sessionId, record) => { + records.push(record); + }, + transcriptTailUuid: () => { + throw new Error('transcript unreadable'); + }, + }, + }); + try { + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const result = await bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'hello' }], + }, + undefined, + { promptId: 'p-no-marker' }, + ); + expect(result.stopReason).toBe('end_turn'); + expect(records).toContainEqual({ + v: 1, + promptId: 'p-no-marker', + state: 'in_flight', + at: expect.any(Number), + }); + } finally { + await bridge.shutdown(); + } + }); + + it('persists the daemon_shutdown error terminal when shutdown flushes a pending prompt', async () => { + const handle = makeChannel({ + promptImpl: () => new Promise(() => {}), + }); + const ledger = recordingLedger(); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + promptLedger: ledger.sink, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const running = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'long work' }], + }, + undefined, + { promptId: 'p-ledger-2' }, + ); + void running.catch(() => undefined); + await bridge.shutdown(); + await running.catch(() => undefined); + expect(terminalRecords(ledger.records)).toEqual([ + { + v: 1, + promptId: 'p-ledger-2', + terminal: 'error', + code: 'daemon_shutdown', + at: expect.any(Number), + }, + ]); + }); + + it('records in_flight for queued admissions and flushes both on shutdown', async () => { + const handle = makeChannel({ + promptImpl: () => new Promise(() => {}), + }); + const ledger = recordingLedger(); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + promptLedger: ledger.sink, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const first = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'never resolves' }], + }, + undefined, + { promptId: 'p-queued-a' }, + ); + void first.catch(() => undefined); + const second = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'queued behind' }], + }, + undefined, + { promptId: 'p-queued-b' }, + ); + void second.catch(() => undefined); + + // Admission is synchronous (write-ahead): both in_flight records are + // on the ledger before either prompt settles — the queued one included. + const inFlight = ledger.records.filter((record) => !('terminal' in record)); + expect(inFlight).toHaveLength(2); + expect(inFlight.map((record) => record.promptId)).toEqual([ + 'p-queued-a', + 'p-queued-b', + ]); + + await bridge.shutdown(); + await first.catch(() => undefined); + await second.catch(() => undefined); + + const terminals = terminalRecords(ledger.records); + expect(terminals.map((record) => record.promptId).sort()).toEqual([ + 'p-queued-a', + 'p-queued-b', + ]); + for (const terminal of terminals) { + expect(terminal.terminal).toBe('error'); + expect(terminal.code).toBe('daemon_shutdown'); + } + }); + + it('maps a cancelled stopReason to a cancelled terminal record', async () => { + const handle = makeChannel({ + promptImpl: () => ({ stopReason: 'cancelled' }), + }); + const ledger = recordingLedger(); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + promptLedger: ledger.sink, + }); + try { + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'stop early' }], + }, + undefined, + { promptId: 'p-ledger-3' }, + ); + expect(terminalRecords(ledger.records)).toEqual([ + { + v: 1, + promptId: 'p-ledger-3', + terminal: 'cancelled', + at: expect.any(Number), + }, + ]); + } finally { + await bridge.shutdown(); + } + }); + + it('keeps prompt execution working when the ledger sink throws', async () => { + const handle = makeChannel(); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + promptLedger: { + appendSync: () => { + throw new Error('disk full'); + }, + }, + }); + try { + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const result = await bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'hello' }], + }); + expect(result.stopReason).toBe('end_turn'); + } finally { + await bridge.shutdown(); + } + }); + + it('writes nothing when no ledger sink is configured', async () => { + const handle = makeChannel(); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + try { + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'hello' }], + }); + // No assertion target beyond "does not throw"; the interesting + // guarantee is that omitting the sink is valid, exercised by every + // other bridge test that never configures one. + expect(bridge.sessionCount).toBe(1); + } finally { + await bridge.shutdown(); + } + }); +}); diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index c00091fdc01..5155d452ecf 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -209,7 +209,12 @@ import type { LiveScreenContextCaptureHandler, LiveSpeakToUserHandler, LiveTaskToolRequestHandler, + PromptLedgerSink, } from './bridgeOptions.js'; +import type { + PromptLedgerRecord, + PromptLedgerTerminalRecord, +} from './prompt-ledger.js'; import { MCP_RESTART_SERVER_DEADLINE_MS } from './mcpTimeouts.js'; import { defaultSpawnChannelFactory } from './spawnChannel.js'; import { writeStderrLine } from './internal/stderrLine.js'; @@ -982,6 +987,11 @@ interface SessionEntry { /** Accepted prompts that have not settled yet (queued + active). */ pendingPromptCount: number; pendingAgentNotificationCount: number; + /** + * Optional prompt terminal ledger sink (injected via BridgeOptions). + * Best-effort synchronous appends; absence keeps pre-existing behavior. + */ + promptLedger?: PromptLedgerSink; /** * Last hold set the owning child reported for this Session, or `null` while * the channel has negotiated reporting but has not yet been heard from. @@ -2000,6 +2010,72 @@ function advanceTurnActivity(entry: SessionEntry): void { previous === undefined ? Date.now() : Math.max(Date.now(), previous + 1); } +/** + * Best-effort ledger append: a failure must never block prompt execution + * or teardown, so it is logged and swallowed. Synchronous by design — the + * daemon-shutdown flush path requires the record to land before process + * exit. + */ +function appendPromptLedgerBestEffort( + entry: SessionEntry, + record: PromptLedgerRecord, +): void { + const ledger = entry.promptLedger; + if (!ledger) return; + try { + ledger.appendSync(entry.sessionId, record); + } catch (error) { + writeStderrLine( + `qwen serve: prompt ledger append failed for session=${entry.sessionId} promptId=${record.promptId}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } +} + +/** + * Project a PromptTerminal into the persisted ledger record. Mirrors the + * in-memory `rememberTerminalTurnStatus` mapping so the ledger and the SSE + * terminal agree, including the `stopReason: 'cancelled'` special case that + * the complete path reports as a cancellation. + */ +function promptLedgerTerminalRecord( + pendingEntry: PendingPromptEntry, + terminal: PromptTerminal, +): PromptLedgerTerminalRecord { + const at = Date.now(); + if (terminal.kind === 'complete') { + if (terminal.result.stopReason === 'cancelled') { + return { + v: 1, + promptId: pendingEntry.promptId, + terminal: 'cancelled', + at, + }; + } + return { + v: 1, + promptId: pendingEntry.promptId, + terminal: 'completed', + ...(terminal.result.stopReason !== undefined + ? { stopReason: terminal.result.stopReason } + : {}), + at, + }; + } + if (terminal.kind === 'cancelled') { + return { v: 1, promptId: pendingEntry.promptId, terminal: 'cancelled', at }; + } + const normalized = normalizeTurnResultError(terminal.err); + return { + v: 1, + promptId: pendingEntry.promptId, + terminal: 'error', + ...(normalized.code !== undefined ? { code: normalized.code } : {}), + at, + }; +} + /** * Publish the formal terminal event for an accepted prompt exactly once. * All terminal paths (agent settle, queued removal, deadline, session @@ -2025,6 +2101,10 @@ function publishPromptTerminal( return; } pendingEntry.terminalPublished = true; + appendPromptLedgerBestEffort( + entry, + promptLedgerTerminalRecord(pendingEntry, terminal), + ); rememberTerminalTurnStatus(entry, pendingEntry, terminal); const originatorClientId = pendingEntry.originatorClientId; // Only a running prompt's terminal belongs to the active turn. The @@ -5871,6 +5951,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { promptQueue: Promise.resolve(), pendingPromptCount: 0, pendingAgentNotificationCount: 0, + ...(opts.promptLedger ? { promptLedger: opts.promptLedger } : {}), pendingPromptList: [], terminalTurnStatuses: new Map(), enrichedTerminalPromptIds: new Set(), @@ -8213,6 +8294,27 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { state: isQueued ? 'queued' : 'running', }; entry.pendingPromptList.push(pendingEntry); + // Dispatch marker: capture the transcript tail uuid before the + // write-ahead `in_flight` lands so cold-load reconciliation can + // require visible writes beyond it (identity-based attribution, + // immune to clock skew). Best-effort: absence degrades reconcile + // to its marker-less evidence chain, never blocks admission. + let tailUuid: string | undefined; + try { + tailUuid = entry.promptLedger?.transcriptTailUuid?.(sessionId); + } catch (error) { + opts.onDiagnosticLine?.( + `qwen serve: prompt ledger dispatch marker failed for session=${sessionId}: ${error instanceof Error ? error.message : String(error)}`, + 'warn', + ); + } + appendPromptLedgerBestEffort(entry, { + v: 1, + promptId, + state: 'in_flight', + ...(tailUuid !== undefined ? { tailUuid } : {}), + at: queuedAt, + }); try { context?.onPromptAdmitted?.(); } catch (error) { diff --git a/packages/acp-bridge/src/bridgeOptions.ts b/packages/acp-bridge/src/bridgeOptions.ts index 1fd3d6304fb..15cccbdc4d6 100644 --- a/packages/acp-bridge/src/bridgeOptions.ts +++ b/packages/acp-bridge/src/bridgeOptions.ts @@ -22,6 +22,7 @@ import type { PermissionAuditPublisher } from './permissionMediator.js'; import type { ServePreflightCell, ServeWorkspaceEnvStatus } from './status.js'; import type { BridgeFileSystem } from './bridgeFileSystem.js'; import type { JournalGrowthSessionLimit } from './replayWindowLimits.js'; +import type { PromptLedgerRecord } from './prompt-ledger.js'; /** * Sink for serve-level diagnostic lines (set by the cli daemon logger). @@ -35,6 +36,26 @@ export type DiagnosticLineSink = ( level?: 'info' | 'warn' | 'error', ) => void; +/** + * Append-only sink for the per-session prompt terminal ledger. The bridge + * owns only the writes (best-effort, synchronous so the daemon-shutdown + * flush lands before process exit); path resolution, reads, and cold-load + * reconciliation live in the serve layer, which knows the session storage + * layout. Keeping this a bare callable seam means the bridge needs no + * filesystem layout knowledge and no core dependency for ledger paths. + */ +export interface PromptLedgerSink { + appendSync(sessionId: string, record: PromptLedgerRecord): void; + /** + * Uuid of the transcript's last record right now, or `undefined` when + * there is no transcript evidence (fresh session, unreadable file). The + * bridge stamps it into the `in_flight` record at admission as the + * dispatch marker; best-effort — a failure or absence only degrades + * cold-load reconciliation back to its marker-less evidence chain. + */ + transcriptTailUuid?(sessionId: string): string | undefined; +} + export interface BridgeFreshSessionAdmissionContext { readonly operation: 'spawn' | 'load' | 'resume' | 'branch'; readonly workspaceCwd: string; @@ -440,6 +461,17 @@ export interface BridgeOptions { statusProvider?: DaemonStatusProvider; /** Optional daemon telemetry seam. Omitted callers get no-op spans/logs. */ telemetry?: BridgeTelemetry; + /** + * Optional prompt terminal ledger sink. When provided, the bridge appends + * an `in_flight` record when a prompt is admitted and a terminal record at + * the single `publishPromptTerminal` exit (including the close/kill/ + * channel-crash/daemon-shutdown flushes), so a restarted daemon can + * reconcile dangling prompts on cold session load. Writes are best-effort: + * failures are logged to stderr and never block prompt execution or + * teardown. Omitted callers keep the pre-existing behavior (no + * persistence, cold loads answer "unknown" for pre-restart prompts). + */ + promptLedger?: PromptLedgerSink; /** * Whether ACP text reads are delegated to the client filesystem service. diff --git a/packages/acp-bridge/src/index.ts b/packages/acp-bridge/src/index.ts index 13a93fafeed..fab71a279a0 100644 --- a/packages/acp-bridge/src/index.ts +++ b/packages/acp-bridge/src/index.ts @@ -17,6 +17,7 @@ export * from './sessionAttachments.js'; export * from './bridgeTypes.js'; export * from './session-source.js'; export * from './bridgeOptions.js'; +export * from './prompt-ledger.js'; export * from './session-restore-timeout.js'; export * from './replayWindowLimits.js'; export * from './spawnChannel.js'; diff --git a/packages/acp-bridge/src/prompt-ledger.test.ts b/packages/acp-bridge/src/prompt-ledger.test.ts new file mode 100644 index 00000000000..06debd83407 --- /dev/null +++ b/packages/acp-bridge/src/prompt-ledger.test.ts @@ -0,0 +1,358 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { mkdtempSync, rmSync, statSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import * as path from 'node:path'; +import { afterAll, describe, expect, it } from 'vitest'; +import { + appendPromptLedgerRecord, + danglingInFlightPromptIds, + isPromptLedgerTerminalRecord, + readPromptLedgerRecords, + recentPromptTerminalRecords, + type PromptLedgerRecord, +} from './prompt-ledger.js'; + +const tmpRoot = mkdtempSync(path.join(tmpdir(), 'prompt-ledger-test-')); +afterAll(() => { + rmSync(tmpRoot, { recursive: true, force: true }); +}); + +function ledgerPath(name: string): string { + return path.join(tmpRoot, `${name}.ledger.jsonl`); +} + +describe('appendPromptLedgerRecord + readPromptLedgerRecords', () => { + it('round-trips the in_flight dispatch marker, dropping invalid ones', () => { + const filePath = ledgerPath('marker'); + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p1', + state: 'in_flight', + tailUuid: 'rec-tail', + at: 1, + }); + // A non-string marker on disk must be dropped, not fatal. + writeFileSync( + filePath, + `${JSON.stringify({ v: 1, promptId: 'p2', state: 'in_flight', tailUuid: 42, at: 2 })}\n`, + { flag: 'a' }, + ); + expect(readPromptLedgerRecords(filePath)).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', tailUuid: 'rec-tail', at: 1 }, + { v: 1, promptId: 'p2', state: 'in_flight', at: 2 }, + ]); + }); + + it('round-trips in_flight and terminal records in order', () => { + const filePath = ledgerPath('roundtrip'); + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: 1, + }); + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p1', + terminal: 'completed', + stopReason: 'end_turn', + at: 2, + }); + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p2', + terminal: 'error', + code: 'daemon_shutdown', + at: 3, + }); + expect(readPromptLedgerRecords(filePath)).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { + v: 1, + promptId: 'p1', + terminal: 'completed', + stopReason: 'end_turn', + at: 2, + }, + { + v: 1, + promptId: 'p2', + terminal: 'error', + code: 'daemon_shutdown', + at: 3, + }, + ]); + }); + + it('creates the parent directory on first append', () => { + const filePath = path.join( + tmpRoot, + 'nested', + 'dir', + 'created', + 'session.ledger.jsonl', + ); + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: 1, + }); + expect(readPromptLedgerRecords(filePath)).toHaveLength(1); + }); + + it('creates the ledger owner-only, not umask-default', () => { + if (process.platform === 'win32') return; // POSIX mode bits only. + // The ledger holds per-prompt activity metadata and must follow the + // transcript's 0o600 convention at creation instead of inheriting the + // umask default (typically 0o644 on shared hosts). + const filePath = path.join(tmpRoot, 'perm', 'session.ledger.jsonl'); + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: 1, + }); + expect(statSync(filePath).mode & 0o777).toBe(0o600); + }); + + it('treats a missing file as an empty ledger', () => { + expect(readPromptLedgerRecords(ledgerPath('missing'))).toEqual([]); + }); + + it('drops a torn tail and malformed lines, keeps valid ones', () => { + const filePath = ledgerPath('torn-tail'); + writeFileSync( + filePath, + [ + JSON.stringify({ v: 1, promptId: 'p1', state: 'in_flight', at: 1 }), + '{"v":1,"promptId":"p2","state":"in_fli', // torn mid-append + JSON.stringify({ + v: 1, + promptId: 'p2', + terminal: 'completed', + at: 2, + }), + 'not json at all', + JSON.stringify({ v: 2, promptId: 'p3', at: 3 }), // unknown version + JSON.stringify({ + v: 1, + promptId: 'p4', + terminal: 'bogus', + at: 4, + }), // unknown terminal state + JSON.stringify({ v: 1, promptId: 42, state: 'in_flight', at: 5 }), // bad id + '', + ].join('\n'), + 'utf8', + ); + const records = readPromptLedgerRecords(filePath); + expect(records.map((record) => record.promptId)).toEqual(['p1', 'p2']); + }); + + it('seals a torn tail so the next appended record survives', () => { + const filePath = ledgerPath('torn-tail-seal'); + // Production crash shape: a complete record, then an append torn + // mid-line (no trailing newline). + writeFileSync( + filePath, + `${JSON.stringify({ v: 1, promptId: 'p1', state: 'in_flight', at: 1 })}\n{"v":1,"promptId":"p2","state":"in_fli`, + 'utf8', + ); + + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p2', + terminal: 'completed', + at: 2, + }); + + // Without the seal the new record would fuse with the torn fragment + // into one unparseable line and BOTH would be lost; with it the torn + // fragment stays droppable and p2's complete record survives. + const records = readPromptLedgerRecords(filePath); + expect(records).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'p2', terminal: 'completed', at: 2 }, + ]); + }); + + it('does not add a seal when the tail is already newline-terminated', () => { + const filePath = ledgerPath('torn-tail-clean'); + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: 1, + }); + + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p2', + terminal: 'completed', + at: 2, + }); + + // No stray blank lines: exactly two records, in order. Assert the raw + // layout too — the reader skips blank lines, so it cannot see a stray + // seal newline; a regression that always appends one would keep every + // read-based assertion green. + expect(statSync(filePath).size).toBe( + JSON.stringify({ v: 1, promptId: 'p1', state: 'in_flight', at: 1 }) + .length + + 1 + + JSON.stringify({ v: 1, promptId: 'p2', terminal: 'completed', at: 2 }) + .length + + 1, + ); + expect(readPromptLedgerRecords(filePath)).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'p2', terminal: 'completed', at: 2 }, + ]); + }); +}); + +describe('readPromptLedgerRecords tail window', () => { + it('returns all records when the file fits inside the window', () => { + const filePath = ledgerPath('tail-fits'); + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: 1, + }); + appendPromptLedgerRecord(filePath, { + v: 1, + promptId: 'p1', + terminal: 'completed', + at: 2, + }); + + expect(readPromptLedgerRecords(filePath, { tailBytes: 4096 })).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'p1', terminal: 'completed', at: 2 }, + ]); + }); + + it('reads only the trailing window and drops the torn first line', () => { + const filePath = ledgerPath('tail-window'); + const lines: string[] = []; + for (let i = 0; i < 6; i += 1) { + lines.push( + `${JSON.stringify({ + v: 1, + promptId: `p${i}`, + state: 'in_flight', + at: i, + })}\n`, + ); + } + writeFileSync(filePath, lines.join(''), 'utf8'); + const lineLength = lines[0]?.length ?? 0; + const fileSize = statSync(filePath).size; + expect(fileSize).toBe(lineLength * 6); + + // The window starts 10 bytes into p3's line (torn) and must still yield + // the two fully-contained trailing records. + const tailBytes = lineLength * 2 + 10; + expect(readPromptLedgerRecords(filePath, { tailBytes })).toEqual([ + { v: 1, promptId: 'p4', state: 'in_flight', at: 4 }, + { v: 1, promptId: 'p5', state: 'in_flight', at: 5 }, + ]); + }); + + it('drops the first window line even when the window starts on a line boundary', () => { + const filePath = ledgerPath('tail-window-aligned'); + const lines: string[] = []; + for (let i = 0; i < 4; i += 1) { + lines.push( + `${JSON.stringify({ + v: 1, + promptId: `p${i}`, + state: 'in_flight', + at: i, + })}\n`, + ); + } + writeFileSync(filePath, lines.join(''), 'utf8'); + const lineLength = lines[0]?.length ?? 0; + + // The window aligns exactly with p2's line start; p2 is dropped anyway + // per the documented "always drop the first window line" contract. + expect( + readPromptLedgerRecords(filePath, { tailBytes: lineLength * 2 }), + ).toEqual([{ v: 1, promptId: 'p3', state: 'in_flight', at: 3 }]); + }); +}); + +describe('danglingInFlightPromptIds', () => { + it('reports prompts whose latest record is still in_flight', () => { + const records: PromptLedgerRecord[] = [ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'p2', state: 'in_flight', at: 2 }, + { v: 1, promptId: 'p1', terminal: 'completed', at: 3 }, + { v: 1, promptId: 'p3', state: 'in_flight', at: 4 }, + ]; + expect(danglingInFlightPromptIds(records)).toEqual(['p2', 'p3']); + }); + + it('keeps first-appearance order and drops settled prompts', () => { + const records: PromptLedgerRecord[] = [ + { v: 1, promptId: 'later', state: 'in_flight', at: 2 }, + { v: 1, promptId: 'first', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'later', terminal: 'error', code: 'x', at: 3 }, + ]; + expect(danglingInFlightPromptIds(records)).toEqual(['first']); + }); + + it('returns empty for an empty ledger', () => { + expect(danglingInFlightPromptIds([])).toEqual([]); + }); +}); + +describe('recentPromptTerminalRecords', () => { + it('filters to terminals and keeps file order', () => { + const records: PromptLedgerRecord[] = [ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'p1', terminal: 'completed', at: 2 }, + { v: 1, promptId: 'p2', state: 'in_flight', at: 3 }, + { + v: 1, + promptId: 'p2', + terminal: 'interrupted', + code: 'daemon_lost', + at: 4, + }, + ]; + const terminals = recentPromptTerminalRecords(records); + expect(terminals.map((record) => record.promptId)).toEqual(['p1', 'p2']); + expect(isPromptLedgerTerminalRecord(terminals[0])).toBe(true); + }); + + it('returns only the trailing limit records', () => { + const records: PromptLedgerRecord[] = []; + for (let i = 0; i < 70; i += 1) { + records.push({ + v: 1, + promptId: `p${i}`, + state: 'in_flight', + at: i, + }); + records.push({ + v: 1, + promptId: `p${i}`, + terminal: 'completed', + at: i + 100, + }); + } + const terminals = recentPromptTerminalRecords(records); + expect(terminals).toHaveLength(64); + expect(terminals[0]?.promptId).toBe('p6'); + expect(terminals[63]?.promptId).toBe('p69'); + }); +}); diff --git a/packages/acp-bridge/src/prompt-ledger.ts b/packages/acp-bridge/src/prompt-ledger.ts new file mode 100644 index 00000000000..37a68e17381 --- /dev/null +++ b/packages/acp-bridge/src/prompt-ledger.ts @@ -0,0 +1,276 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + appendFileSync, + closeSync, + mkdirSync, + openSync, + readFileSync, + readSync, + statSync, + type Stats, +} from 'node:fs'; +import * as path from 'node:path'; + +/** + * Per-session append-only ledger of prompt terminal outcomes. + * + * The daemon's turn terminal events (`turn_complete` / `turn_error`) are + * synthesized at the SSE layer and never persisted, so after a daemon + * restart a cold `POST /session/:id/load` replay carries no terminal + * evidence and clients keyed on `promptId` can only answer "unknown". This + * ledger closes that gap with a small sidecar JSONL file stored next to the + * session transcript: one `in_flight` record when a prompt is admitted, one + * terminal record when its formal terminal publishes (including the + * close/kill/channel-crash/daemon-shutdown flushes). + * + * Records deliberately carry no prompt content — only ids, state, and + * timestamps (privacy boundary; see + * docs/design/2026-08-19-prompt-terminal-ledger-design.md). + * This module is intentionally dependency-free beyond `node:fs`: the bridge + * only writes; reads and reconciliation live in the serve layer. + */ + +export interface PromptLedgerInFlightRecord { + v: 1; + promptId: string; + state: 'in_flight'; + at: number; + /** + * Dispatch marker: uuid of the transcript's last record at admission. + * The transcript is append-only, so any record after this marker was + * written after admission — reconciliation requires at least one visible + * write beyond it before attributing an outcome (an identity check immune + * to clock skew; absent for records written before the marker existed). + */ + tailUuid?: string; +} + +export type PromptLedgerTerminalState = + | 'completed' + | 'cancelled' + | 'error' + | 'interrupted'; + +export interface PromptLedgerTerminalRecord { + v: 1; + promptId: string; + terminal: PromptLedgerTerminalState; + /** Machine-readable cause: `daemon_shutdown`, `channel_closed`, `daemon_lost`, ... */ + code?: string; + stopReason?: string; + at: number; +} + +export type PromptLedgerRecord = + | PromptLedgerInFlightRecord + | PromptLedgerTerminalRecord; + +export function isPromptLedgerTerminalRecord( + record: PromptLedgerRecord, +): record is PromptLedgerTerminalRecord { + return 'terminal' in record; +} + +/** Append one record as a JSON line. Synchronous by design: the + * daemon-shutdown flush must land before process exit. Throws on I/O + * failure; callers own the best-effort policy. */ +export function appendPromptLedgerRecord( + filePath: string, + record: PromptLedgerRecord, +): void { + mkdirSync(path.dirname(filePath), { recursive: true }); + sealTornTailSync(filePath); + // Owner-only at creation (mode applies only to the creating call): + // the ledger holds per-prompt activity metadata and must follow the + // transcript's 0o600 convention rather than the umask default. + appendFileSync(filePath, `${JSON.stringify(record)}\n`, { + encoding: 'utf8', + mode: 0o600, + }); +} + +/** + * Seal a torn tail left by a crash mid-append: when the file is non-empty + * and its last byte is not a newline, the next append would fuse with the + * truncated line and the reader would drop BOTH records (the fused line + * fails JSON parsing). A leading newline keeps the torn fragment droppable + * and the new record intact. Missing files need no seal. + */ +function sealTornTailSync(filePath: string): void { + let stats: Stats; + try { + stats = statSync(filePath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; + throw error; + } + if (stats.size === 0) return; + const fd = openSync(filePath, 'r'); + try { + const lastByte = Buffer.alloc(1); + const bytesRead = readSync(fd, lastByte, 0, 1, stats.size - 1); + if (bytesRead === 1 && lastByte[0] !== 0x0a) { + appendFileSync(filePath, '\n', 'utf8'); + } + } finally { + closeSync(fd); + } +} + +function coercePromptLedgerRecord( + value: unknown, +): PromptLedgerRecord | undefined { + if (typeof value !== 'object' || value === null) return undefined; + const record = value as Record; + const promptId = record['promptId']; + const at = record['at']; + if (record['v'] !== 1 || typeof promptId !== 'string') return undefined; + if (typeof at !== 'number' || !Number.isFinite(at)) return undefined; + if (record['state'] === 'in_flight') { + const tailUuid = record['tailUuid']; + return { + v: 1, + promptId, + state: 'in_flight', + ...(typeof tailUuid === 'string' && tailUuid.length > 0 + ? { tailUuid } + : {}), + at, + }; + } + const terminal = record['terminal']; + if ( + terminal !== 'completed' && + terminal !== 'cancelled' && + terminal !== 'error' && + terminal !== 'interrupted' + ) { + return undefined; + } + const code = record['code']; + const stopReason = record['stopReason']; + return { + v: 1, + promptId, + terminal, + ...(typeof code === 'string' ? { code } : {}), + ...(typeof stopReason === 'string' ? { stopReason } : {}), + at, + }; +} + +/** Options for {@link readPromptLedgerRecords}. */ +export interface ReadPromptLedgerOptions { + /** + * Read at most this many trailing bytes instead of the whole file. The + * first line of the window is dropped (the window start can tear a line + * in half), so callers must size the window for the records they need. + * Used by the per-request load path to avoid reading and parsing the + * entire ledger of a long session. + */ + tailBytes?: number; +} + +/** + * Read all records in file order. A torn tail (crash mid-append) or any + * malformed line is dropped rather than fatal: the ledger is advisory + * evidence, and reconciliation treats "unreadable" the same as "absent" + * (fail-closed). Only ENOENT maps to an empty ledger; other I/O errors + * propagate to the caller. + * + * With `options.tailBytes`, files larger than the window are read from the + * tail only; the first line inside the window is always dropped because the + * window start may fall mid-line. + */ +export function readPromptLedgerRecords( + filePath: string, + options?: ReadPromptLedgerOptions, +): PromptLedgerRecord[] { + let contents: string; + let dropFirstLine = false; + try { + const tailBytes = options?.tailBytes; + if (tailBytes === undefined) { + contents = readFileSync(filePath, 'utf8'); + } else { + const size = statSync(filePath).size; + if (size <= tailBytes) { + contents = readFileSync(filePath, 'utf8'); + } else { + const fd = openSync(filePath, 'r'); + try { + const buffer = Buffer.alloc(tailBytes); + const bytesRead = readSync( + fd, + buffer, + 0, + tailBytes, + size - tailBytes, + ); + // A UTF-8 sequence torn at the window start is replaced with + // U+FFFD by toString; that torn first line is dropped below anyway. + contents = buffer.subarray(0, bytesRead).toString('utf8'); + dropFirstLine = true; + } finally { + closeSync(fd); + } + } + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return []; + throw error; + } + const records: PromptLedgerRecord[] = []; + const lines = contents.split('\n'); + for (let i = dropFirstLine ? 1 : 0; i < lines.length; i++) { + const line = lines[i]; + if (line === undefined || line.length === 0) continue; + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + continue; + } + const record = coercePromptLedgerRecord(parsed); + if (record !== undefined) records.push(record); + } + return records; +} + +/** + * promptIds whose latest record is still `in_flight` (no terminal followed), + * in first-appearance order. Later records supersede earlier ones for the + * same promptId, so a reconciliation append or a duplicate write is + * naturally idempotent on the read side. + */ +export function danglingInFlightPromptIds( + records: readonly PromptLedgerRecord[], +): string[] { + const latest = new Map(); + for (const record of records) { + latest.set(record.promptId, record); + } + const dangling: string[] = []; + for (const [promptId, record] of latest) { + if (!isPromptLedgerTerminalRecord(record)) dangling.push(promptId); + } + return dangling; +} + +/** Cap for terminal records embedded in a load response. */ +const PROMPT_TERMINALS_RESPONSE_LIMIT = 64; + +/** The most recent terminal records (file order), up to the response cap. */ +export function recentPromptTerminalRecords( + records: readonly PromptLedgerRecord[], +): PromptLedgerTerminalRecord[] { + const terminals = records.filter(isPromptLedgerTerminalRecord); + return terminals.length <= PROMPT_TERMINALS_RESPONSE_LIMIT + ? terminals + : terminals.slice(-PROMPT_TERMINALS_RESPONSE_LIMIT); +} diff --git a/packages/cli/src/serve/prompt-terminal-ledger.test.ts b/packages/cli/src/serve/prompt-terminal-ledger.test.ts new file mode 100644 index 00000000000..66c96b6b1b3 --- /dev/null +++ b/packages/cli/src/serve/prompt-terminal-ledger.test.ts @@ -0,0 +1,1075 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import * as path from 'node:path'; +import { randomUUID } from 'node:crypto'; +import { afterAll, describe, expect, it } from 'vitest'; +import { SessionService, type ChatRecord } from '@qwen-code/qwen-code-core'; +import { + appendPromptLedgerRecord, + readPromptLedgerRecords, + type PromptLedgerRecord, +} from '@qwen-code/acp-bridge/promptLedger'; +import { + createPromptLedgerSink, + readRecentPromptTerminals, + readTranscriptTailUuid, + reconcileDanglingPromptTerminals, + withPromptTerminals, +} from './prompt-terminal-ledger.js'; + +const tmpRoot = mkdtempSync(path.join(tmpdir(), 'prompt-terminals-test-')); +afterAll(() => { + rmSync(tmpRoot, { recursive: true, force: true }); +}); + +interface Fixture { + workspaceDir: string; + runtimeBaseDir: string; + sessionService: SessionService; + sessionId: string; + transcriptPath: string; + ledgerPath: string; +} + +function makeFixture(): Fixture { + const workspaceDir = path.join(tmpRoot, randomUUID()); + mkdirSync(workspaceDir, { recursive: true }); + const runtimeBaseDir = path.join(tmpRoot, randomUUID()); + const sessionService = new SessionService(workspaceDir, { + runtimeBaseDir, + }); + const sessionId = randomUUID(); + const ledgerPath = sessionService.getPromptLedgerPath(sessionId); + const transcriptPath = path.join( + path.dirname(ledgerPath), + `${sessionId}.jsonl`, + ); + return { + workspaceDir, + runtimeBaseDir, + sessionService, + sessionId, + transcriptPath, + ledgerPath, + }; +} + +const RECORD_BASE_MS = Date.UTC(2026, 0, 1, 0, 0, 0); +let recordSeq = 0; +function record( + fixture: Fixture, + uuid: string, + parentUuid: string | null, + text: string, +): ChatRecord { + const isModel = uuid.startsWith('a'); + return { + uuid, + parentUuid, + sessionId: fixture.sessionId, + timestamp: new Date(RECORD_BASE_MS + recordSeq++ * 1000).toISOString(), + type: isModel ? 'assistant' : 'user', + provenance: isModel ? 'assistant_output' : 'real_user', + cwd: fixture.workspaceDir, + version: '1.0.0', + message: { + role: isModel ? 'model' : 'user', + parts: [{ text }], + }, + }; +} + +function recordAt( + fixture: Fixture, + uuid: string, + parentUuid: string | null, + text: string, + atMs: number, +): ChatRecord { + return { + ...record(fixture, uuid, parentUuid, text), + timestamp: new Date(atMs).toISOString(), + }; +} + +function toolCallRecord( + fixture: Fixture, + uuid: string, + parentUuid: string, + callId: string | null, +): ChatRecord { + return { + ...record(fixture, uuid, parentUuid, ''), + message: { + role: 'model', + parts: [ + { + functionCall: { + name: 'run_shell_command', + ...(callId === null ? {} : { id: callId }), + args: {}, + }, + }, + ], + }, + }; +} + +function systemRecord( + fixture: Fixture, + uuid: string, + parentUuid: string, + subtype: NonNullable, + systemPayload: ChatRecord['systemPayload'], +): ChatRecord { + return { + ...record(fixture, uuid, parentUuid, ''), + type: 'system', + subtype, + systemPayload, + }; +} + +function writeTranscript( + fixture: Fixture, + records: readonly ChatRecord[], +): void { + mkdirSync(path.dirname(fixture.transcriptPath), { recursive: true }); + writeFileSync( + fixture.transcriptPath, + records.map((entry) => JSON.stringify(entry)).join('\n') + '\n', + 'utf8', + ); +} + +function writeLedger( + fixture: Fixture, + records: readonly PromptLedgerRecord[], +): void { + for (const record of records) { + appendPromptLedgerRecord(fixture.ledgerPath, record); + } +} + +describe('reconcileDanglingPromptTerminals', () => { + it('marks a transcript-clean dangling prompt completed', async () => { + const fixture = makeFixture(); + writeLedger(fixture, [{ v: 1, promptId: 'p1', state: 'in_flight', at: 1 }]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { + v: 1, + promptId: 'p1', + terminal: 'completed', + stopReason: 'reconstructed_from_transcript', + at: expect.any(Number), + }, + ]); + }); + + it('marks an interrupted_prompt dangling prompt interrupted', async () => { + const fixture = makeFixture(); + writeLedger(fixture, [{ v: 1, promptId: 'p1', state: 'in_flight', at: 1 }]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + record(fixture, 'u2', 'a1', 'orphaned follow-up'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { + v: 1, + promptId: 'p1', + terminal: 'interrupted', + code: 'daemon_lost', + at: expect.any(Number), + }, + ]); + }); + + it('marks an interrupted_turn dangling prompt interrupted', async () => { + const fixture = makeFixture(); + writeLedger(fixture, [{ v: 1, promptId: 'p1', state: 'in_flight', at: 1 }]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'run something'), + toolCallRecord(fixture, 'a1', 'u1', 'call-1'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { + v: 1, + promptId: 'p1', + terminal: 'interrupted', + code: 'daemon_lost', + at: expect.any(Number), + }, + ]); + }); + + it('stays fail-closed when the transcript cannot be read', async () => { + const fixture = makeFixture(); + writeLedger(fixture, [{ v: 1, promptId: 'p1', state: 'in_flight', at: 1 }]); + // No transcript file at all: loadSession yields undefined. + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + ]); + }); + + it('stays fail-closed when the last transcript write predates the admission', async () => { + const fixture = makeFixture(); + // The sole dangling prompt was admitted AFTER the transcript's last + // write: it never produced a transcript entry (still queued when the + // daemon died), so the visible tail belongs to an earlier settled turn + // and must not be attributed to it. `at` sits far in the future of the + // fixture's timestamps so the ordering cannot be accidental. + const admissionAt = Date.UTC(2030, 0, 1); + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: admissionAt, + }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toEqual([ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: admissionAt, + }, + ]); + }); + + it('appends nothing when there is no dangling prompt', async () => { + const fixture = makeFixture(); + writeLedger(fixture, [ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'p1', terminal: 'completed', at: 2 }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(2); + }); + + it('appends nothing when several prompts are dangling', async () => { + const fixture = makeFixture(); + // Queued scenario: p1 never ran, p2 was running when the daemon died. + // Under FIFO the visible tail belongs to the oldest running prompt, + // but with both dangling the tail's owner cannot be verified — fail + // closed and keep both unknown instead of guessing. + writeLedger(fixture, [ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'p2', state: 'in_flight', at: 2 }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(2); + }); + + it('attributes the tail to the sole dangling prompt behind a settled one', async () => { + const fixture = makeFixture(); + // Valid interleave on a real time axis: p2 was admitted (queued) while + // p1 still ran, p1 settled, then p2 dispatched and produced the visible + // tail before the daemon died. p1's terminal postdates p1's own turn + // but predates p2's writes (transcript timestamps start at + // RECORD_BASE_MS), so the FIFO evidence attributes the tail to p2 even + // though a terminal record sits after its in_flight line. + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: RECORD_BASE_MS - 3000, + }, + { + v: 1, + promptId: 'p2', + state: 'in_flight', + at: RECORD_BASE_MS - 2000, + }, + { + v: 1, + promptId: 'p1', + terminal: 'completed', + at: RECORD_BASE_MS - 1000, + }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u2', null, 'p2 question'), + record(fixture, 'a2', 'u2', 'p2 answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + const records = readPromptLedgerRecords(fixture.ledgerPath); + expect(records).toHaveLength(4); + expect(records[3]).toMatchObject({ + promptId: 'p2', + terminal: 'completed', + stopReason: 'reconstructed_from_transcript', + }); + }); + + it('stays fail-closed for a queued prompt that never dispatched', async () => { + const fixture = makeFixture(); + // B was admitted (its in_flight written at admission) and queued while + // A still ran; A settled and the daemon died before B dispatched. The + // visible tail is A's turn — it predates A's own settled terminal, so + // the FIFO evidence cannot attribute it to B. + writeLedger(fixture, [ + { + v: 1, + promptId: 'A', + state: 'in_flight', + at: RECORD_BASE_MS - 3000, + }, + { + v: 1, + promptId: 'B', + state: 'in_flight', + at: RECORD_BASE_MS - 2000, + }, + // A's settle postdates its turn's writes (the real ordering). + { v: 1, promptId: 'A', terminal: 'completed', at: Date.now() }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'A question'), + record(fixture, 'a1', 'u1', 'A answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(3); + }); + + it('stays fail-closed for a stale dangling behind a later settled prompt', async () => { + const fixture = makeFixture(); + // A restore path that skips reconciliation left p1's in_flight + // dangling; prompt c1 later ran to completion. c1's clean tail + // predates c1's own settled terminal, so it must not be attributed to + // the stale p1. + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: RECORD_BASE_MS - 5000, + }, + { + v: 1, + promptId: 'c1', + state: 'in_flight', + at: RECORD_BASE_MS - 4000, + }, + { v: 1, promptId: 'c1', terminal: 'completed', at: Date.now() }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'c1 question'), + record(fixture, 'a1', 'u1', 'c1 answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(3); + }); + + it('attributes the outcome when a visible write lands beyond the dispatch marker', async () => { + const fixture = makeFixture(); + // The admission marker points at u1: a1 was written after admission, + // so the clean tail can be attributed to p1. + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + tailUuid: 'u1', + at: RECORD_BASE_MS - 1000, + }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'p1 question'), + record(fixture, 'a1', 'u1', 'p1 answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + const records = readPromptLedgerRecords(fixture.ledgerPath); + expect(records).toHaveLength(2); + expect(records[1]).toEqual( + expect.objectContaining({ promptId: 'p1', terminal: 'completed' }), + ); + }); + + it('stays fail-closed when nothing was written beyond the dispatch marker', async () => { + const fixture = makeFixture(); + // The marker is the transcript's last record: the admitted turn never + // wrote anything visible, so no outcome may be synthesized even though + // every temporal guard passes. + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + tailUuid: 'a1', + at: RECORD_BASE_MS - 1000, + }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'p1 question'), + record(fixture, 'a1', 'u1', 'p1 answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(1); + }); + + it('stays fail-closed when the dispatch marker is absent from the transcript', async () => { + const fixture = makeFixture(); + // A marker that the projection does not contain (e.g. a restore that + // rewrote the transcript) cannot prove any write postdates admission. + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + tailUuid: 'gone-uuid', + at: RECORD_BASE_MS - 1000, + }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'p1 question'), + record(fixture, 'a1', 'u1', 'p1 answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(1); + }); + + it('stays fail-closed when the visible tail shares a millisecond with the FIFO clocks', async () => { + const fixture = makeFixture(); + // Both compared clocks are 1 ms-granularity `Date.now()` reads: p1's + // final transcript write and p1's settled terminal can land in the same + // millisecond T. Equality must veto — a strict `<` never fires on + // `T < T`, and attributing p1's clean tail to the queued p2 would + // synthesize a terminal for a prompt that never executed. + const t = RECORD_BASE_MS + 60_000; + writeLedger(fixture, [ + { v: 1, promptId: 'p1', state: 'in_flight', at: t - 3000 }, + { v: 1, promptId: 'p2', state: 'in_flight', at: t - 2000 }, + { v: 1, promptId: 'p1', terminal: 'completed', at: t }, + ]); + writeTranscript(fixture, [ + recordAt(fixture, 'u1', null, 'p1 question', t), + recordAt(fixture, 'a1', 'u1', 'p1 answer', t), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(3); + }); + + it('stays fail-closed when the visible tail shares a millisecond with the admission', async () => { + const fixture = makeFixture(); + // A transcript record persisted in the same millisecond the prompt was + // admitted cannot prove the admitted prompt's turn wrote it. + const t = RECORD_BASE_MS + 60_000; + writeLedger(fixture, [{ v: 1, promptId: 'p1', state: 'in_flight', at: t }]); + writeTranscript(fixture, [ + recordAt(fixture, 'u1', null, 'earlier turn question', t), + recordAt(fixture, 'a1', 'u1', 'earlier turn answer', t), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(1); + }); + + it('stays fail-closed behind a prompt_deadline_exceeded terminal', async () => { + const fixture = makeFixture(); + // The deadline path releases the FIFO while the wedged agent is + // explicitly allowed to keep streaming (DAEMON-003): p1's stale writes + // postdate both its deadline terminal and p2's admission, so the + // temporal comparisons alone cannot veto them — the deadline code + // itself must. + const deadlineAt = RECORD_BASE_MS + 30_000; + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: RECORD_BASE_MS - 10_000, + }, + { + v: 1, + promptId: 'p1', + terminal: 'error', + code: 'prompt_deadline_exceeded', + at: deadlineAt, + }, + { + v: 1, + promptId: 'p2', + state: 'in_flight', + at: deadlineAt + 1000, + }, + ]); + writeTranscript(fixture, [ + recordAt(fixture, 'u1', null, 'p1 stale write', deadlineAt + 2000), + recordAt(fixture, 'a1', 'u1', 'p1 stale answer', deadlineAt + 3000), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(3); + }); + + it('stays fail-closed behind a stale prompt_deadline_exceeded terminal', async () => { + const fixture = makeFixture(); + // The deadline veto is intentionally unconditional: the append-only + // ledger never expires records, so a stale deadline terminal (here one + // hour before the target's admission, with a clean post-admission tail) + // still keeps the session permanently fail-closed. The trade is missing + // terminals over wrong ones; pin the behavior so any future recency + // bound is a deliberate change. + const staleDeadlineAt = RECORD_BASE_MS - 3_600_000; + const admissionAt = RECORD_BASE_MS - 1000; + writeLedger(fixture, [ + { v: 1, promptId: 'p1', state: 'in_flight', at: staleDeadlineAt - 1000 }, + { + v: 1, + promptId: 'p1', + terminal: 'error', + code: 'prompt_deadline_exceeded', + at: staleDeadlineAt, + }, + { v: 1, promptId: 'p2', state: 'in_flight', at: admissionAt }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'p2 question'), + record(fixture, 'a1', 'u1', 'p2 answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(3); + }); + + it('stays fail-closed when a prompt is admitted during the reconciliation window', async () => { + const fixture = makeFixture(); + writeLedger(fixture, [ + { v: 1, promptId: 'p-old', state: 'in_flight', at: 1 }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + ]); + + // Race: a new prompt is admitted while `loadSession` runs, so its + // `in_flight` lands after reconcile's ledger snapshot — the visible + // tail may now belong to it, and the verdict computed from the + // snapshot must not be stamped onto p-old. + class RacingSessionService extends SessionService { + override async loadSession(sessionId: string) { + appendPromptLedgerRecord(this.getPromptLedgerPath(sessionId), { + v: 1, + promptId: 'p-new', + state: 'in_flight', + // The admission must predate the visible tail (fixture records + // sit just past RECORD_BASE_MS): a wall-clock admission ~months + // after the tail could never own it, so the race would not + // actually threaten the verdict. + at: RECORD_BASE_MS + 26_000, + }); + return super.loadSession(sessionId); + } + } + const racing = new RacingSessionService(fixture.workspaceDir, { + runtimeBaseDir: fixture.runtimeBaseDir, + }); + + await reconcileDanglingPromptTerminals(racing, fixture.sessionId); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(2); + }); + + it('stays fail-closed when a compression checkpoint postdates the admission', async () => { + const fixture = makeFixture(); + // A chat_compression record written after p1's admission replaces the + // api history wholesale; the verdict's projection no longer carries + // p1's turn, so nothing may be attributed. + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: RECORD_BASE_MS - 1000, + }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + systemRecord(fixture, 'c1', 'a1', 'chat_compression', { + info: { + originalTokenCount: 100, + newTokenCount: 50, + // CompressionStatus.COMPRESSED is not exported from the core + // barrel; reconcile only reads compressedHistory. + compressionStatus: 1, + }, + compressedHistory: [{ role: 'user', parts: [{ text: 'summary' }] }], + } as ChatRecord['systemPayload']), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(1); + }); + + it('stays fail-closed when a backward clock step hides a post-marker compression', async () => { + const fixture = makeFixture(); + // The compression checkpoint sits past p1's dispatch marker but its + // wall clock stepped backward below the admission time. A marker + // admission must fence compression by position (anything past the + // marker postdates admission), or the clock step hides the reset and + // the compressed tail is wrongly attributed to p1. + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + tailUuid: 'u1', + at: RECORD_BASE_MS - 1000, + }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + { + ...systemRecord(fixture, 'c1', 'a1', 'chat_compression', { + info: { + originalTokenCount: 100, + newTokenCount: 50, + compressionStatus: 1, + }, + compressedHistory: [{ role: 'user', parts: [{ text: 'summary' }] }], + } as ChatRecord['systemPayload']), + // Backward clock step: pre-admission wall time, post-marker + // position. + timestamp: new Date(RECORD_BASE_MS - 60_000).toISOString(), + }, + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(1); + }); + + it('stays fail-closed when only post-admission system records follow', async () => { + const fixture = makeFixture(); + // After p1's admission the transcript gains only a system record + // (custom_title here) that stays outside the api history; the raw tail + // postdates the admission but the projection the verdict runs on holds + // no evidence of p1's turn. + writeLedger(fixture, [ + { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: RECORD_BASE_MS + 3_600_000, + }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'earlier question'), + record(fixture, 'a1', 'u1', 'earlier answer'), + systemRecord(fixture, 's1', 'a1', 'custom_title', { + customTitle: 'Later title', + }), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(1); + }); + + it('attributes the tail to the running prompt behind a cancelled queued one', async () => { + const fixture = makeFixture(); + // S3 shape: A was running, B queued behind it, B was cancelled from + // the queue, then the daemon died while A still ran. A is the only + // dangling prompt and the interrupted tail belongs to it — B's + // settled in_flight must not veto A. + writeLedger(fixture, [ + { v: 1, promptId: 'A', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'B', state: 'in_flight', at: 2 }, + { v: 1, promptId: 'B', terminal: 'cancelled', at: 3 }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'A question'), + record(fixture, 'a1', 'u1', 'partial answer'), + record(fixture, 'u2', 'a1', 'orphaned follow-up'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + const records = readPromptLedgerRecords(fixture.ledgerPath); + expect(records).toHaveLength(4); + expect(records[3]).toEqual({ + v: 1, + promptId: 'A', + terminal: 'interrupted', + code: 'daemon_lost', + at: expect.any(Number), + }); + }); + + it('marks a dangling prompt interrupted on an id-less functionCall tail', async () => { + const fixture = makeFixture(); + // detectTurnInterruption ignores functionCalls without an id (no wire + // pairing), but a model tail holding ANY functionCall still means the + // daemon died mid tool-run — the reconcile-side guard must upgrade the + // verdict to interrupted. + writeLedger(fixture, [{ v: 1, promptId: 'p1', state: 'in_flight', at: 1 }]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'run something'), + toolCallRecord(fixture, 'a1', 'u1', null), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { + v: 1, + promptId: 'p1', + terminal: 'interrupted', + code: 'daemon_lost', + at: expect.any(Number), + }, + ]); + }); + + it('is idempotent: a second reconcile appends nothing new', async () => { + const fixture = makeFixture(); + writeLedger(fixture, [{ v: 1, promptId: 'p1', state: 'in_flight', at: 1 }]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(2); + }); + + it('stays fail-closed when the dangling prompt was re-admitted after a settled turn', async () => { + const fixture = makeFixture(); + // Re-admission shape: p1 settled, then the same promptId was admitted + // again and dangled. The guard skips in_flight records of prompts with + // a terminal on disk (their settle state is ambiguous), so no verdict + // is attributed. The old "anomalous interleave" veto (last in_flight + // must match target) was superseded by this guard: it wrongly vetoed + // the running prompt behind a cancelled queued one (see the S3-shaped + // test above). + writeLedger(fixture, [ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'p1', terminal: 'completed', at: 2 }, + { v: 1, promptId: 'p1', state: 'in_flight', at: 3 }, + ]); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + ]); + + await reconcileDanglingPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(3); + }); +}); + +describe('readRecentPromptTerminals + withPromptTerminals', () => { + it('returns undefined without ledger evidence', () => { + const fixture = makeFixture(); + expect( + readRecentPromptTerminals(fixture.sessionService, fixture.sessionId), + ).toBeUndefined(); + }); + + it('returns undefined when the ledger holds only in_flight records', () => { + const fixture = makeFixture(); + writeLedger(fixture, [{ v: 1, promptId: 'p1', state: 'in_flight', at: 1 }]); + expect( + readRecentPromptTerminals(fixture.sessionService, fixture.sessionId), + ).toBeUndefined(); + }); + + it('returns the trailing terminal records', () => { + const fixture = makeFixture(); + writeLedger(fixture, [ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + { v: 1, promptId: 'p1', terminal: 'completed', at: 2 }, + { v: 1, promptId: 'p2', state: 'in_flight', at: 3 }, + { + v: 1, + promptId: 'p2', + terminal: 'error', + code: 'daemon_shutdown', + at: 4, + }, + ]); + expect( + readRecentPromptTerminals(fixture.sessionService, fixture.sessionId), + ).toEqual([ + { v: 1, promptId: 'p1', terminal: 'completed', at: 2 }, + { + v: 1, + promptId: 'p2', + terminal: 'error', + code: 'daemon_shutdown', + at: 4, + }, + ]); + }); + + it('reads only the trailing window, not the whole ledger', () => { + const fixture = makeFixture(); + // A distinctive sentinel terminal, then >256 KiB of in_flight filler, + // then <64 trailing terminals. A full read would return the sentinel + // too (every terminal fits under the 64-record response cap); the + // windowed call-site read cannot see it — dropping tailBytes from + // readRecentPromptTerminals flips this assertion. + const lines: string[] = [ + `${JSON.stringify({ + v: 1, + promptId: 'sentinel', + terminal: 'completed', + at: 0, + })}\n`, + ]; + for (let i = 0; i < 5000; i += 1) { + lines.push( + `${JSON.stringify({ + v: 1, + promptId: `filler${String(i).padStart(6, '0')}`, + state: 'in_flight', + at: i + 1, + })}\n`, + ); + } + for (let i = 0; i < 10; i += 1) { + lines.push( + `${JSON.stringify({ + v: 1, + promptId: `tail${String(i).padStart(2, '0')}`, + terminal: 'completed', + at: 5001 + i, + })}\n`, + ); + } + mkdirSync(path.dirname(fixture.ledgerPath), { recursive: true }); + writeFileSync(fixture.ledgerPath, lines.join(''), 'utf8'); + + const terminals = readRecentPromptTerminals( + fixture.sessionService, + fixture.sessionId, + ); + expect(terminals).toHaveLength(10); + expect(terminals!.map((t) => t.promptId)).not.toContain('sentinel'); + }); + + it('leaves the response untouched without terminals', () => { + const session = { + sessionId: 's1', + attached: false, + state: {}, + workspaceCwd: '/workspace/a', + }; + expect(withPromptTerminals(session, undefined)).toBe(session); + expect(withPromptTerminals(session, [])).toBe(session); + }); + + it('attaches the promptTerminals field', () => { + const session = { + sessionId: 's1', + attached: false, + state: {}, + workspaceCwd: '/workspace/a', + }; + const terminals = [ + { v: 1 as const, promptId: 'p1', terminal: 'completed' as const, at: 2 }, + ]; + expect(withPromptTerminals(session, terminals)).toMatchObject({ + sessionId: 's1', + attached: false, + promptTerminals: terminals, + }); + }); +}); + +describe('createPromptLedgerSink', () => { + it('appends through the SessionService path layout', () => { + const fixture = makeFixture(); + const sink = createPromptLedgerSink( + fixture.workspaceDir, + fixture.runtimeBaseDir, + ); + sink.appendSync(fixture.sessionId, { + v: 1, + promptId: 'p1', + state: 'in_flight', + at: 1, + }); + expect(readPromptLedgerRecords(fixture.ledgerPath)).toEqual([ + { v: 1, promptId: 'p1', state: 'in_flight', at: 1 }, + ]); + }); + + it('reads the transcript tail uuid through the same path layout', () => { + const fixture = makeFixture(); + const sink = createPromptLedgerSink( + fixture.workspaceDir, + fixture.runtimeBaseDir, + ); + expect(sink.transcriptTailUuid?.(fixture.sessionId)).toBeUndefined(); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + ]); + expect(sink.transcriptTailUuid?.(fixture.sessionId)).toBe('a1'); + }); +}); + +describe('readTranscriptTailUuid', () => { + it('returns the last record uuid, degrading on missing or torn evidence', () => { + const fixture = makeFixture(); + expect(readTranscriptTailUuid(fixture.transcriptPath)).toBeUndefined(); + writeTranscript(fixture, [ + record(fixture, 'u1', null, 'question'), + record(fixture, 'a1', 'u1', 'answer'), + ]); + expect(readTranscriptTailUuid(fixture.transcriptPath)).toBe('a1'); + // A crash mid-append leaves a truncated final line: no reliable marker. + writeFileSync( + fixture.transcriptPath, + '{"uuid":"u1"}\n{"uuid":"a1","text":"answ', + 'utf8', + ); + expect(readTranscriptTailUuid(fixture.transcriptPath)).toBeUndefined(); + }); +}); diff --git a/packages/cli/src/serve/prompt-terminal-ledger.ts b/packages/cli/src/serve/prompt-terminal-ledger.ts new file mode 100644 index 00000000000..13b921117fe --- /dev/null +++ b/packages/cli/src/serve/prompt-terminal-ledger.ts @@ -0,0 +1,376 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { Content } from '@google/genai'; +import { closeSync, openSync, readSync, statSync } from 'node:fs'; +import { + buildApiHistoryFromConversation, + detectTurnInterruption, + SessionService, + TURN_INTERRUPTION_HISTORY_TAIL_COUNT, + type ChatRecord, + type ResumedSessionData, +} from '@qwen-code/qwen-code-core'; +import { + appendPromptLedgerRecord, + danglingInFlightPromptIds, + isPromptLedgerTerminalRecord, + readPromptLedgerRecords, + recentPromptTerminalRecords, + type PromptLedgerInFlightRecord, + type PromptLedgerRecord, + type PromptLedgerTerminalRecord, +} from '@qwen-code/acp-bridge/promptLedger'; +import type { PromptLedgerSink } from '@qwen-code/acp-bridge/bridgeOptions'; +import type { BridgeRestoredSession } from '@qwen-code/acp-bridge/bridgeTypes'; + +/** + * Serve-layer assembly of the bridge's ledger sink: the bridge only calls + * `appendSync`, and this module owns the path layout via `SessionService` + * (the ledger lives beside the transcript in the session storage dir). + */ +export function createPromptLedgerSink( + workspaceCwd: string, + sessionRuntimeBaseDir: string, +): PromptLedgerSink { + const sessionService = new SessionService(workspaceCwd, { + runtimeBaseDir: sessionRuntimeBaseDir, + }); + return { + appendSync(sessionId, record) { + appendPromptLedgerRecord( + sessionService.getPromptLedgerPath(sessionId), + record, + ); + }, + transcriptTailUuid(sessionId) { + return readTranscriptTailUuid( + sessionService.getSessionTranscriptPath(sessionId), + ); + }, + }; +} + +/** + * Byte window for the dispatch-marker read: only the trailing record + * matters, so the hot admission path never reads (or JSON-parses) a whole + * multi-megabyte transcript. A final record larger than the window (or a + * torn tail) simply yields no marker — admission and reconciliation both + * degrade to the marker-less evidence chain. + */ +const TRANSCRIPT_TAIL_BYTES = 64 * 1024; + +/** + * Uuid of the transcript's last record, or `undefined` without readable + * evidence (missing file, empty file, torn/corrupt tail). Best-effort by + * contract: any failure maps to "no marker", never to an admission error. + */ +export function readTranscriptTailUuid( + transcriptPath: string, +): string | undefined { + let contents: string; + try { + const size = statSync(transcriptPath).size; + if (size === 0) return undefined; + const windowBytes = Math.min(size, TRANSCRIPT_TAIL_BYTES); + const buffer = Buffer.alloc(windowBytes); + const fd = openSync(transcriptPath, 'r'); + try { + readSync(fd, buffer, 0, windowBytes, size - windowBytes); + } finally { + closeSync(fd); + } + contents = buffer.toString('utf8'); + } catch { + return undefined; + } + const lines = contents.split('\n'); + for (let i = lines.length - 1; i >= 0; i--) { + const line = lines[i]; + if (line === undefined || line.length === 0) continue; + try { + const uuid = (JSON.parse(line) as { uuid?: unknown }).uuid; + return typeof uuid === 'string' && uuid.length > 0 ? uuid : undefined; + } catch { + return undefined; // Torn or corrupt final line: no reliable marker. + } + } + return undefined; +} + +/** + * Close the loop for prompts left `in_flight` by a daemon that died before + * publishing (and persisting) their terminal. Called on the cold + * `POST /session/:id/load` path after `bridge.loadSession` returned: + * + * - dangling detection on the ledger (a prompt with `in_flight` and no + * terminal); + * - `detectTurnInterruption` on the transcript tail decides the outcome; + * - the verdict is appended back to the ledger so the response (and every + * later load) sees it. + * + * Attribution is guarded four ways (each mirrors a concrete wrong-terminal + * probe; see the design doc): the dispatch marker (when admission recorded + * the transcript tail uuid, the target must have written a visible record + * beyond it — an identity check immune to clock skew), the temporal + * evidence measured on the same projection the verdict uses, a compression + * checkpoint after the target's admission voiding the evidence chain, and + * under FIFO admission the visible tail being strictly newer than every + * other prompt's settled terminal (a same-millisecond tail, and any tail + * behind a `prompt_deadline_exceeded` terminal whose wedged turn may still + * be writing, cannot be attributed). + * + * Fail-closed invariant: when the outcome cannot be attributed with + * confidence, nothing is appended and the prompt stays "unknown" — a + * wrong terminal is never synthesized. + */ +export async function reconcileDanglingPromptTerminals( + sessionService: SessionService, + sessionId: string, +): Promise { + const ledgerPath = sessionService.getPromptLedgerPath(sessionId); + let records: PromptLedgerRecord[]; + try { + records = readPromptLedgerRecords(ledgerPath); + } catch { + return; // Unreadable ledger: no evidence, fail-closed. + } + const snapshotLength = records.length; + const dangling = danglingInFlightPromptIds(records); + if (dangling.length === 0) return; + // Fail closed on multiple dangling prompts. Under FIFO admission the + // visible transcript tail belongs to the OLDEST running prompt, but with + // several prompts dangling the tail's owner cannot be verified (the + // queued ones never wrote a turn): synthesizing a terminal for any of + // them — including the newest — could attribute an earlier prompt's turn + // to the wrong id. They all stay `unknown` + // (see docs/design/2026-08-19-prompt-terminal-ledger-design.md). + if (dangling.length > 1) return; + const target = dangling[0]; + if (target === undefined) return; + // Attribution guard: skip the in_flight records of prompts that settled + // (a terminal record exists for them) and require the last remaining + // in_flight record to be target's own admission. In `[A if, B if, + // B cancelled]` (B queued then cancelled while A still ran) the tail + // belongs to A even though B's in_flight is the later record — the naive + // "last in_flight must match target" guard wrongly vetoed A with B's + // settled in_flight. + const settledPromptIds = new Set( + records.filter(isPromptLedgerTerminalRecord).map((r) => r.promptId), + ); + let targetAdmission: PromptLedgerInFlightRecord | undefined; + for (const record of records) { + if ( + !isPromptLedgerTerminalRecord(record) && + !settledPromptIds.has(record.promptId) + ) { + targetAdmission = record; + } + } + if (targetAdmission === undefined || targetAdmission.promptId !== target) { + return; + } + let resumed: ResumedSessionData | undefined; + try { + resumed = await sessionService.loadSession(sessionId); + } catch { + return; // Degraded transcript: fail-closed. + } + if (resumed === undefined) return; + const messages = resumed.conversation.messages; + // Dispatch marker evidence: when admission recorded the transcript tail + // uuid, the target's turn must have written at least one visible record + // beyond it (the transcript is append-only, so anything after the marker + // postdates admission). This is an identity/ordering check immune to + // clock skew; a marker missing from the projection, or present with no + // visible write after it, fails closed. + const admissionMarker = targetAdmission.tailUuid; + let markerIndex = -1; + if (admissionMarker !== undefined) { + markerIndex = messages.findIndex( + (record) => record.uuid === admissionMarker, + ); + let wroteAfterMarker = false; + for (let i = markerIndex + 1; i < messages.length; i++) { + const record = messages[i]; + if (record === undefined || record.type === 'system') continue; + if (!record.message || record.subtype === 'realtime_message') continue; + wroteAfterMarker = true; + break; + } + if (markerIndex < 0 || !wroteAfterMarker) return; + } + // Projection-consistent temporal evidence: only records that actually + // enter the api history the verdict runs on can prove the target's turn + // wrote anything. System records (ui_telemetry, custom_title, ...) stay + // outside the projection, and a compression candidate replaces it wholesale + // (mirrors SessionApiHistoryAccumulator, packages/core). Measuring the + // last write on the raw stream instead would let evidence that the + // verdict never sees pass the guard. + let lastVisibleWriteMs = NaN; + let compressedAfterAdmission = false; + for (let idx = 0; idx < messages.length; idx++) { + const record = messages[idx]; + const writeMs = Date.parse(record.timestamp); + if (record.type === 'system') { + if (isCompressionResetRecord(record)) { + // Marker-bearing admissions order by position: anything past the + // marker postdates admission, so a backward clock step cannot hide + // a post-admission compression. Marker-less admissions fall back + // to the wall clock. + const afterAdmission = + admissionMarker !== undefined + ? idx > markerIndex + : Number.isFinite(writeMs) && writeMs >= targetAdmission.at; + if (afterAdmission) compressedAfterAdmission = true; + } + continue; + } + if (!record.message || record.subtype === 'realtime_message') continue; + if (Number.isFinite(writeMs)) lastVisibleWriteMs = writeMs; + } + // FIFO evidence: under FIFO admission the target's turn can only start + // after every other prompt settled, so any visible tail not strictly + // newer than some other prompt's terminal belongs to that prompt's turn + // — a queued prompt that never dispatched and a stale dangling left by a + // restore path that skips reconciliation both fail here. Equality is + // vetoed as well: both clocks are 1 ms-granularity `Date.now()` reads, so + // a same-millisecond tail cannot be attributed with confidence. + let lastOtherTerminalAt = 0; + for (const record of records) { + if (isPromptLedgerTerminalRecord(record) && record.promptId !== target) { + // A `prompt_deadline_exceeded` terminal does not fence its turn's + // writes: the deadline path releases the FIFO while the wedged agent + // is explicitly allowed to keep streaming (DAEMON-003), so stale + // writes postdating the terminal could be attributed to the target. + if (record.code === 'prompt_deadline_exceeded') return; + lastOtherTerminalAt = Math.max(lastOtherTerminalAt, record.at); + } + } + if ( + compressedAfterAdmission || + !Number.isFinite(lastVisibleWriteMs) || + lastVisibleWriteMs <= targetAdmission.at || + lastVisibleWriteMs <= lastOtherTerminalAt + ) { + return; + } + const apiHistory = buildApiHistoryFromConversation(resumed.conversation); + const historyTail = apiHistory.slice(-TURN_INTERRUPTION_HISTORY_TAIL_COUNT); + const verdict = detectTurnInterruption(historyTail); + // Id-less tool-call guard: `detectTurnInterruption` ignores functionCalls + // without an id (they cannot be paired on the wire), but reconciliation + // needs no wire pairing — a model tail holding ANY functionCall means the + // daemon died mid tool-run, so upgrade the verdict to interrupted + // (`interrupted_turn` semantics). + const interrupted = + verdict.kind !== 'none' || tailHoldsAnyFunctionCall(historyTail); + // TOCTOU fence: a prompt admitted while `loadSession` ran appended its + // `in_flight` after the snapshot above, and the visible tail may now + // belong to it — the verdict computed from the snapshot must not be + // stamped onto the old dangling id. The ledger is append-only, so an + // unchanged length proves no record landed during the window. + let refetch: PromptLedgerRecord[]; + try { + refetch = readPromptLedgerRecords(ledgerPath); + } catch { + return; + } + if (refetch.length !== snapshotLength) return; + const record: PromptLedgerTerminalRecord = interrupted + ? { + v: 1, + promptId: target, + terminal: 'interrupted', + code: 'daemon_lost', + at: Date.now(), + } + : { + v: 1, + promptId: target, + terminal: 'completed', + stopReason: 'reconstructed_from_transcript', + at: Date.now(), + }; + try { + appendPromptLedgerRecord(ledgerPath, record); + } catch { + // Best-effort: the dangling prompt stays unknown. + } +} + +/** + * Whether a system record resets the api history projection: a + * `chat_compression` record carrying a `compressedHistory` payload (the + * accumulator swaps the whole history for it). Kept inline instead of + * importing `isApiHistoryCompressionCandidate` so this module stays inside + * the cli package; the predicate mirrors that helper. + */ +function isCompressionResetRecord(record: ChatRecord): boolean { + if (record.type !== 'system' || record.subtype !== 'chat_compression') { + return false; + } + return Boolean( + (record.systemPayload as { compressedHistory?: unknown } | undefined) + ?.compressedHistory, + ); +} + +/** + * Whether the history tail's last entry is a model turn holding at least + * one `functionCall` part (id or not). See the id-less tool-call guard in + * {@link reconcileDanglingPromptTerminals}. + */ +function tailHoldsAnyFunctionCall(history: Content[]): boolean { + const last = history[history.length - 1]; + if (last?.role !== 'model') return false; + return (last.parts ?? []).some((part) => part.functionCall !== undefined); +} + +/** + * Tail byte window for load-response reads. Records are ~150 bytes and the + * response caps at 64 terminals, so 256 KiB holds hundreds of terminals even + * with in_flight lines interleaved — the response is the full trailing + * window for any realistic session while the per-load hot path never reads + * (or JSON-parses) a whole multi-megabyte ledger. Sessions whose ledger + * outgrows the window return a best-effort subset, which the response + * contract already allows. + */ +const RECENT_TERMINALS_TAIL_BYTES = 256 * 1024; + +/** + * The most recent ledger terminals for the load response, or `undefined` + * when there is no ledger evidence (field omitted entirely — old clients + * and no-ledger sessions see the exact pre-existing response shape). + */ +export function readRecentPromptTerminals( + sessionService: SessionService, + sessionId: string, +): PromptLedgerTerminalRecord[] | undefined { + try { + const terminals = recentPromptTerminalRecords( + readPromptLedgerRecords(sessionService.getPromptLedgerPath(sessionId), { + tailBytes: RECENT_TERMINALS_TAIL_BYTES, + }), + ); + return terminals.length > 0 ? terminals : undefined; + } catch { + return undefined; + } +} + +/** + * Attach `promptTerminals` to a load response. Kept as a wrapper (rather + * than mutating the bridge's `BridgeRestoredSession` type) so the serve + * layer owns this response extension alone. + */ +export function withPromptTerminals( + session: T, + terminals: readonly PromptLedgerTerminalRecord[] | undefined, +): T | (T & { promptTerminals: PromptLedgerTerminalRecord[] }) { + if (terminals === undefined || terminals.length === 0) return session; + return { ...session, promptTerminals: [...terminals] }; +} diff --git a/packages/cli/src/serve/routes/session-prompt-terminals.test.ts b/packages/cli/src/serve/routes/session-prompt-terminals.test.ts new file mode 100644 index 00000000000..f2e30225253 --- /dev/null +++ b/packages/cli/src/serve/routes/session-prompt-terminals.test.ts @@ -0,0 +1,289 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import * as path from 'node:path'; +import { randomUUID } from 'node:crypto'; +import express, { type Response } from 'express'; +import request from 'supertest'; +import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'; +import { SessionService, type ChatRecord } from '@qwen-code/qwen-code-core'; +import { + appendPromptLedgerRecord, + readPromptLedgerRecords, +} from '@qwen-code/acp-bridge/promptLedger'; +import { + SessionNotFoundError, + type AcpSessionBridge, +} from '../acp-session-bridge.js'; +import { + createWorkspaceRegistry, + type WorkspaceRuntime, +} from '../workspace-registry.js'; + +const archiveMocks = vi.hoisted(() => ({ + assertSessionLoadable: vi.fn(), +})); + +vi.mock('../server/session-archive.js', async (importOriginal) => ({ + ...(await importOriginal()), + assertSessionLoadable: archiveMocks.assertSessionLoadable, +})); + +import { registerSessionRoutes } from './session.js'; + +const tmpRoot = mkdtempSync(path.join(tmpdir(), 'session-prompt-terminals-')); +afterAll(() => { + rmSync(tmpRoot, { recursive: true, force: true }); +}); + +interface Fixture { + workspaceDir: string; + sessionService: SessionService; + sessionId: string; + ledgerPath: string; + runtime: WorkspaceRuntime; +} + +function makeFixture( + loadOverrides: { attached?: boolean; hasActivePrompt?: boolean } = {}, +): Fixture { + const workspaceDir = path.join(tmpRoot, randomUUID()); + mkdirSync(workspaceDir, { recursive: true }); + const runtimeBaseDir = path.join(tmpRoot, randomUUID()); + const sessionService = new SessionService(workspaceDir, { + runtimeBaseDir, + }); + const sessionId = randomUUID(); + const ledgerPath = sessionService.getPromptLedgerPath(sessionId); + const bridge = bridgeWithColdLoad(sessionId, workspaceDir, loadOverrides); + return { + workspaceDir, + sessionService, + sessionId, + ledgerPath, + runtime: { + workspaceId: randomUUID(), + workspaceCwd: workspaceDir, + sessionRuntimeBaseDir: runtimeBaseDir, + primary: true, + trusted: true, + bridge, + } as WorkspaceRuntime, + }; +} + +function writeTranscript(fixture: Fixture, records: readonly ChatRecord[]) { + const transcriptPath = path.join( + path.dirname(fixture.ledgerPath), + `${fixture.sessionId}.jsonl`, + ); + mkdirSync(path.dirname(transcriptPath), { recursive: true }); + writeFileSync( + transcriptPath, + records.map((record) => JSON.stringify(record)).join('\n') + '\n', + 'utf8', + ); +} + +function chatRecord( + fixture: Fixture, + uuid: string, + parentUuid: string | null, + text: string, +): ChatRecord { + const isModel = uuid.startsWith('a'); + return { + uuid, + parentUuid, + sessionId: fixture.sessionId, + timestamp: new Date(Date.UTC(2026, 0, 1, 0, 0, 0)).toISOString(), + type: isModel ? 'assistant' : 'user', + provenance: isModel ? 'assistant_output' : 'real_user', + cwd: fixture.workspaceDir, + version: '1.0.0', + message: { + role: isModel ? 'model' : 'user', + parts: [{ text }], + }, + }; +} + +function bridgeWithColdLoad( + sessionId: string, + workspaceCwd: string, + loadOverrides: { attached?: boolean; hasActivePrompt?: boolean }, +): AcpSessionBridge { + const restored = { + sessionId, + attached: loadOverrides.attached ?? false, + hasActivePrompt: loadOverrides.hasActivePrompt ?? false, + currentCwd: workspaceCwd, + }; + return { + loadSession: vi.fn(async () => restored), + resumeSession: vi.fn(async () => restored), + getSessionSummary: vi.fn((requestedId: string) => { + throw new SessionNotFoundError(requestedId); + }), + } as unknown as AcpSessionBridge; +} + +function makeApp(fixture: Fixture) { + const app = express(); + app.use(express.json()); + const registry = createWorkspaceRegistry([fixture.runtime]); + registerSessionRoutes(app, { + boundWorkspace: fixture.workspaceDir, + bridge: fixture.runtime.bridge, + workspaceRegistry: registry, + archiveCoordinator: { + runSharedMany: async (_sessionIds, fn) => await fn(), + } as Parameters[1]['archiveCoordinator'], + mutate: () => (_req, _res, next) => next(), + sendBridgeError: (res: Response, err: unknown) => { + res.status(500).json({ + error: 'test bridge error', + detail: + err instanceof Error ? `${err.name}: ${err.message}` : String(err), + }); + }, + sessionShellCommandEnabled: true, + languageCodes: ['en'], + }); + return app; +} + +describe('POST /session/:id/load prompt terminals', () => { + beforeEach(() => { + vi.clearAllMocks(); + archiveMocks.assertSessionLoadable.mockResolvedValue('active'); + }); + + it('reconciles a dangling prompt and returns promptTerminals', async () => { + const fixture = makeFixture(); + writeTranscript(fixture, [ + chatRecord(fixture, 'u1', null, 'question'), + chatRecord(fixture, 'a1', 'u1', 'answer'), + ]); + appendPromptLedgerRecord(fixture.ledgerPath, { + v: 1, + promptId: 'p-route-1', + state: 'in_flight', + at: 1, + }); + const app = makeApp(fixture); + + const res = await request(app) + .post(`/session/${fixture.sessionId}/load`) + .send({}); + + if (res.status !== 200) { + throw new Error(`load failed: ${JSON.stringify(res.body)}`); + } + expect(res.body.promptTerminals).toEqual([ + { + v: 1, + promptId: 'p-route-1', + terminal: 'completed', + stopReason: 'reconstructed_from_transcript', + at: expect.any(Number), + }, + ]); + // The verdict is persisted, so a later load sees it without redoing work. + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(2); + }); + + it('omits the field when the session has no ledger', async () => { + const fixture = makeFixture(); + writeTranscript(fixture, [ + chatRecord(fixture, 'u1', null, 'question'), + chatRecord(fixture, 'a1', 'u1', 'answer'), + ]); + const app = makeApp(fixture); + + const res = await request(app) + .post(`/session/${fixture.sessionId}/load`) + .send({}); + + expect(res.status).toBe(200); + expect(res.body.promptTerminals).toBeUndefined(); + }); + + it('does not reconcile an attached load', async () => { + const fixture = makeFixture({ attached: true }); + writeTranscript(fixture, [ + chatRecord(fixture, 'u1', null, 'question'), + chatRecord(fixture, 'a1', 'u1', 'answer'), + ]); + appendPromptLedgerRecord(fixture.ledgerPath, { + v: 1, + promptId: 'p-live-1', + state: 'in_flight', + at: 1, + }); + const app = makeApp(fixture); + + const res = await request(app) + .post(`/session/${fixture.sessionId}/load`) + .send({}); + + expect(res.status).toBe(200); + // Still dangling, no terminal to report, and no reconciliation ran. + expect(res.body.promptTerminals).toBeUndefined(); + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(1); + }); + + it('does not reconcile a load while a prompt is active', async () => { + const fixture = makeFixture({ hasActivePrompt: true }); + writeTranscript(fixture, [ + chatRecord(fixture, 'u1', null, 'question'), + chatRecord(fixture, 'a1', 'u1', 'answer'), + ]); + appendPromptLedgerRecord(fixture.ledgerPath, { + v: 1, + promptId: 'p-active-1', + state: 'in_flight', + at: 1, + }); + const app = makeApp(fixture); + + const res = await request(app) + .post(`/session/${fixture.sessionId}/load`) + .send({}); + + expect(res.status).toBe(200); + // The live entry owns the prompt's terminal; the ledger stays untouched. + expect(res.body.promptTerminals).toBeUndefined(); + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(1); + }); + + it('keeps the resume response free of promptTerminals and appends nothing', async () => { + const fixture = makeFixture(); + writeTranscript(fixture, [ + chatRecord(fixture, 'u1', null, 'question'), + chatRecord(fixture, 'a1', 'u1', 'answer'), + ]); + appendPromptLedgerRecord(fixture.ledgerPath, { + v: 1, + promptId: 'p-resume-1', + state: 'in_flight', + at: 1, + }); + const app = makeApp(fixture); + + const res = await request(app) + .post(`/session/${fixture.sessionId}/resume`) + .send({}); + + expect(res.status).toBe(200); + // Resume keeps its exact pre-existing response shape: no + // promptTerminals field and no reconciliation append. + expect(res.body.promptTerminals).toBeUndefined(); + expect(readPromptLedgerRecords(fixture.ledgerPath)).toHaveLength(1); + }); +}); diff --git a/packages/cli/src/serve/routes/session.ts b/packages/cli/src/serve/routes/session.ts index 086b02be4b2..1953e882be6 100644 --- a/packages/cli/src/serve/routes/session.ts +++ b/packages/cli/src/serve/routes/session.ts @@ -105,6 +105,11 @@ import { sessionExportFormatValues, } from '../server/session-export.js'; import { setDaemonTelemetryWorkspace } from '../server/telemetry.js'; +import { + readRecentPromptTerminals, + reconcileDanglingPromptTerminals, + withPromptTerminals, +} from '../prompt-terminal-ledger.js'; import { createSessionOrganizationService } from '../session-organization-helpers.js'; import { omitSkillDetailsForSdkSurface, @@ -3282,7 +3287,37 @@ export function registerSessionRoutes( throw error; } } - return restored; + // Prompt terminal ledger: reconcile prompts left in_flight by + // a dead previous daemon before responding. Only the cold path + // (no live entry attached and no active prompt on a live entry) + // is eligible — an attached load has a live owner that will + // publish the real terminal itself, and live-conversation + // workspaces store transcripts outside the runtime layout this + // reconciliation reads. Gated on `action === 'load'` to match + // the load-mediation contract (resume keeps its exact + // pre-existing response shape). + if ( + action === 'load' && + !restored.attached && + !restored.hasActivePrompt && + runtime.provenance !== 'live-conversation' + ) { + try { + await reconcileDanglingPromptTerminals( + sessionService, + sessionId, + ); + } catch { + // Best-effort: a failure leaves dangling prompts unknown + // (fail-closed) and must never fail the load itself. + } + } + return withPromptTerminals( + restored, + action === 'load' + ? readRecentPromptTerminals(sessionService, sessionId) + : undefined, + ); }, ); try { diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 6b4e37ea6b3..5db8e10719e 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/packages/cli/src/serve/run-qwen-serve.ts @@ -1178,6 +1178,7 @@ async function loadServeRuntimeModules() { workspaceSkillsStatusModule, totalSessionAdmissionModule, workspaceRegistryModule, + promptLedgerModule, ] = await Promise.all([ import('./server.js'), import('@qwen-code/acp-bridge/bridge'), @@ -1190,6 +1191,7 @@ async function loadServeRuntimeModules() { import('./workspace-skills-status.js'), import('./total-session-admission.js'), import('./workspace-registry.js'), + import('./prompt-terminal-ledger.js'), ]); return { createServeApp: serverModule.createServeApp, @@ -1219,6 +1221,7 @@ async function loadServeRuntimeModules() { workspaceRegistryModule.createWorkspaceSessionOwnerIndex, createWorkspaceGenerationGuard: workspaceRegistryModule.createWorkspaceGenerationGuard, + createPromptLedgerSink: promptLedgerModule.createPromptLedgerSink, }; } @@ -4360,6 +4363,12 @@ async function runQwenServeImpl( ? { permissionResponseTimeoutMs: opts.permissionResponseTimeoutMs } : {}), boundWorkspace, + // Prompt terminal ledger: persisted beside the transcript so a + // restarted daemon can reconcile dangling prompts on cold load. + promptLedger: runtime.createPromptLedgerSink( + boundWorkspace, + primarySessionRuntimeBaseDir, + ), sessionShellCommandEnabled, childEnvOverrides, channelFactory, @@ -4773,6 +4782,10 @@ async function runQwenServeImpl( ? { permissionResponseTimeoutMs: opts.permissionResponseTimeoutMs } : {}), boundWorkspace: workspaceInput.cwd, + promptLedger: runtime.createPromptLedgerSink( + workspaceInput.cwd, + secondaryEnv.sessionRuntimeBaseDir, + ), sessionShellCommandEnabled, childEnvOverrides, channelFactory: secondaryChannelFactory, @@ -5336,6 +5349,16 @@ async function runQwenServeImpl( ? { permissionResponseTimeoutMs: opts.permissionResponseTimeoutMs } : {}), boundWorkspace: cwd, + // Live-conversation workspaces keep transcripts outside the + // runtime storage layout, so no ledger sink is wired there. + ...(provenance === 'live-conversation' + ? {} + : { + promptLedger: runtime.createPromptLedgerSink( + cwd, + wsEnv.sessionRuntimeBaseDir, + ), + }), sessionShellCommandEnabled, childEnvOverrides, channelFactory: wsChannelFactory, diff --git a/packages/cli/src/services/insight/generators/DataProcessor.test.ts b/packages/cli/src/services/insight/generators/DataProcessor.test.ts index 5e4b8f63fea..8acd8352bd5 100644 --- a/packages/cli/src/services/insight/generators/DataProcessor.test.ts +++ b/packages/cli/src/services/insight/generators/DataProcessor.test.ts @@ -1083,6 +1083,57 @@ describe('DataProcessor', () => { expect(paths.some((p) => p.includes('chat3.jsonl'))).toBe(true); }); + it('should skip prompt terminal ledger sidecars when scanning chat files', async () => { + mockedFs.readdir.mockResolvedValueOnce(['project1'] as unknown as Awaited< + ReturnType + >); + + mockedFs.stat.mockImplementation((path) => { + const pathStr = String(path); + if (pathStr.includes('project1') && !pathStr.includes('chats')) { + return Promise.resolve({ + isDirectory: () => true, + mtimeMs: 1234567890, + } as Awaited>); + } + if (pathStr.endsWith('.jsonl')) { + return Promise.resolve({ + isDirectory: () => false, + mtimeMs: 1234567890, + } as Awaited>); + } + throw new Error('Unexpected path: ' + pathStr); + }); + + const sessionId = '550e8400-e29b-41d4-a716-446655440000'; + mockedFs.readdir.mockImplementation((path) => { + const pathStr = String(path); + if (pathStr.endsWith('chats')) { + return Promise.resolve([ + `${sessionId}.jsonl`, + `${sessionId}.ledger.jsonl`, + ] as unknown as Awaited>); + } + return Promise.resolve( + [] as unknown as Awaited>, + ); + }); + + const result = await ( + dataProcessor as unknown as { + scanChatFiles( + baseDir: string, + ): Promise>; + } + ).scanChatFiles('/base'); + + // The ledger sidecar is not a transcript: only the real session + // JSONL may be selected. + expect(result).toHaveLength(1); + expect(result[0].path).toContain(`${sessionId}.jsonl`); + expect(result[0].path).not.toContain('.ledger.jsonl'); + }); + it('should skip projects without chats directory', async () => { mockedFs.readdir.mockResolvedValueOnce([ 'project1', diff --git a/packages/cli/src/services/insight/generators/DataProcessor.ts b/packages/cli/src/services/insight/generators/DataProcessor.ts index 2aa786c53a8..74076fe5344 100644 --- a/packages/cli/src/services/insight/generators/DataProcessor.ts +++ b/packages/cli/src/services/insight/generators/DataProcessor.ts @@ -996,7 +996,13 @@ None captured`; try { // Get all chat files in the chats directory const files = await fs.readdir(chatsDir); - const chatFiles = files.filter((file) => file.endsWith('.jsonl')); + // The prompt terminal ledger sidecar (.ledger.jsonl) is not + // a transcript — only real session JSONL files carry chat + // records. + const chatFiles = files.filter( + (file) => + file.endsWith('.jsonl') && !file.endsWith('.ledger.jsonl'), + ); for (const file of chatFiles) { const filePath = path.join(chatsDir, file); diff --git a/packages/cli/vitest.config.ts b/packages/cli/vitest.config.ts index 818c217ef93..1ac489daead 100644 --- a/packages/cli/vitest.config.ts +++ b/packages/cli/vitest.config.ts @@ -89,6 +89,10 @@ export default defineConfig({ __dirname, '../acp-bridge/src/bridgeOptions.ts', ), + '@qwen-code/acp-bridge/promptLedger': path.resolve( + __dirname, + '../acp-bridge/src/prompt-ledger.ts', + ), '@qwen-code/acp-bridge/bridgeTypes': path.resolve( __dirname, '../acp-bridge/src/bridgeTypes.ts', diff --git a/packages/core/src/services/sessionService.test.ts b/packages/core/src/services/sessionService.test.ts index 07eef2a4348..ff185969f5d 100644 --- a/packages/core/src/services/sessionService.test.ts +++ b/packages/core/src/services/sessionService.test.ts @@ -1859,6 +1859,30 @@ describe('SessionService', () => { expect.stringContaining(`/chats/archive/${sessionIdA}.jsonl`), ); }); + + it('should remove prompt ledger sidecars in both archive states', async () => { + vi.mocked(jsonl.readLines).mockImplementation( + async (filePath: string) => { + if (filePath.includes('/chats/archive/')) return [recordA1]; + const error = new Error('ENOENT') as NodeJS.ErrnoException; + error.code = 'ENOENT'; + throw error; + }, + ); + existsSyncSpy.mockImplementation((filePath: fs.PathLike) => + filePath.toString().endsWith(`${sessionIdA}.ledger.jsonl`), + ); + + const result = await sessionService.removeSession(sessionIdA); + + expect(result).toBe(true); + expect(unlinkSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/${sessionIdA}.ledger.jsonl`), + ); + expect(unlinkSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/archive/${sessionIdA}.ledger.jsonl`), + ); + }); }); describe('archiveSessions', () => { @@ -1946,6 +1970,94 @@ describe('SessionService', () => { ); }); + it('should move the prompt ledger alongside the archived session', async () => { + mockActiveSessionOnly(); + existsSyncSpy.mockImplementation((filePath) => { + const value = filePath.toString(); + if (value.includes('/chats/archive/')) return false; + return value.endsWith(`${sessionIdA}.ledger.jsonl`); + }); + + const result = await sessionService.archiveSessions([sessionIdA]); + + expect(result.archived).toEqual([sessionIdA]); + expect(result.errors).toEqual([]); + expect(renameSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/${sessionIdA}.ledger.jsonl`), + expect.stringContaining(`/chats/archive/${sessionIdA}.ledger.jsonl`), + ); + }); + + it('should warn but still archive when the prompt ledger move fails', async () => { + mockActiveSessionOnly(); + existsSyncSpy.mockImplementation((filePath) => { + const value = filePath.toString(); + if (value.includes('/chats/archive/')) return false; + return value.endsWith(`${sessionIdA}.ledger.jsonl`); + }); + const warnings: string[] = []; + const service = new SessionService('/test/project/root', { + onWarning: (message) => warnings.push(message), + }); + const ledgerError = new Error('ledger move failed'); + renameSyncSpy.mockImplementation((sourcePath) => { + if (sourcePath.toString().endsWith('.ledger.jsonl')) { + throw ledgerError; + } + return undefined; + }); + + const result = await service.archiveSessions([sessionIdA]); + + expect(result.archived).toEqual([sessionIdA]); + expect(result.errors).toEqual([]); + expect(warnings).toHaveLength(1); + expect(warnings[0]).toContain( + `archiveSessions: failed to move prompt ledger for ${sessionIdA}`, + ); + // The warning carries the full paths so the split pair is debuggable. + expect(warnings[0]).toContain(`/chats/${sessionIdA}.ledger.jsonl`); + expect(warnings[0]).toContain( + `/chats/archive/${sessionIdA}.ledger.jsonl`, + ); + }); + + it('should merge the prompt ledger into an existing destination instead of wedging', async () => { + mockActiveSessionOnly(); + const sourceLedger = + '{"v":1,"promptId":"p1","state":"in_flight","at":1}\n'; + vi.spyOn(fs, 'readFileSync').mockReturnValue(sourceLedger); + const appendFileSyncSpy = vi + .spyOn(fs, 'appendFileSync') + .mockImplementation(() => undefined); + // Both the active and the archived ledger exist (e.g. a partially + // completed earlier archive cycle): the merge path must run. + existsSyncSpy.mockImplementation((filePath) => + filePath.toString().endsWith(`${sessionIdA}.ledger.jsonl`), + ); + + const result = await sessionService.archiveSessions([sessionIdA]); + + expect(result.archived).toEqual([sessionIdA]); + expect(result.errors).toEqual([]); + // Source records are concatenated onto the destination (append-only + // JSONL, write order preserved)... + expect(appendFileSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/archive/${sessionIdA}.ledger.jsonl`), + expect.stringContaining('"promptId":"p1"'), + 'utf8', + ); + // ...the source sidecar is unlinked... + expect(unlinkSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/${sessionIdA}.ledger.jsonl`), + ); + // ...and no rename was attempted for the ledger. + expect(renameSyncSpy).not.toHaveBeenCalledWith( + expect.stringContaining(`${sessionIdA}.ledger.jsonl`), + expect.anything(), + ); + }); + it('should not move worktree sidecar when archiving JSONL fails', async () => { mockActiveSessionOnly(); mockActiveWorktreeSidecarOnly(); @@ -2173,6 +2285,27 @@ describe('SessionService', () => { ); }); + it('should move the prompt ledger back to the active directory when unarchiving', async () => { + mockArchivedSessionOnly(); + existsSyncSpy.mockImplementation((filePath) => { + const value = filePath.toString(); + if (value.endsWith(`/chats/${sessionIdA}.jsonl`)) return false; + if (value.endsWith(`${sessionIdA}.ledger.jsonl`)) { + return value.includes('/chats/archive/'); + } + return false; + }); + + const result = await sessionService.unarchiveSessions([sessionIdA]); + + expect(result.unarchived).toEqual([sessionIdA]); + expect(result.errors).toEqual([]); + expect(renameSyncSpy).toHaveBeenCalledWith( + expect.stringContaining(`/chats/archive/${sessionIdA}.ledger.jsonl`), + expect.stringContaining(`/chats/${sessionIdA}.ledger.jsonl`), + ); + }); + it('should not move worktree sidecar when unarchiving JSONL fails', async () => { mockArchivedSessionOnly(); mockArchivedWorktreeSidecarOnly(); diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index e1cce9fc0de..ce9270fa64f 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -607,6 +607,40 @@ export class SessionService { return this.getWorktreeSessionPathForState(sessionId, 'active'); } + /** + * Returns the absolute path to the per-session prompt terminal ledger + * (append-only sidecar JSONL next to the transcript), in the given + * archive state's chats directory. The file may not exist yet — + * consumers must treat ENOENT as "no ledger evidence". + */ + private getPromptLedgerPathForState( + sessionId: string, + state: SessionArchiveState, + ): string { + return path.join( + this.getChatsDirForState(state), + `${sessionId}.ledger.jsonl`, + ); + } + + /** + * Returns the absolute path to the per-session prompt terminal ledger + * (append-only sidecar JSONL next to the transcript). The file may not + * exist yet — consumers must treat ENOENT as "no ledger evidence". + */ + getPromptLedgerPath(sessionId: string): string { + return this.getPromptLedgerPathForState(sessionId, 'active'); + } + + /** + * Returns the absolute path to the active session transcript + * (append-only JSONL). The file may not exist yet — consumers must + * treat ENOENT as "no transcript evidence". + */ + getSessionTranscriptPath(sessionId: string): string { + return this.getSessionFilePath(sessionId, 'active'); + } + getWorktreeSessionPathForArchiveState( sessionId: string, state: SessionArchiveState, @@ -902,6 +936,15 @@ export class SessionService { } } + private removePromptLedgers(sessionId: string): void { + for (const state of ['active', 'archived'] as const) { + const ledger = this.getPromptLedgerPathForState(sessionId, state); + if (fs.existsSync(ledger)) { + this.removeFileIfExists(ledger); + } + } + } + private removeFileHistoryBackups(sessionId: string): void { fs.rmSync( path.join(Storage.getGlobalQwenDir(), FILE_HISTORY_DIR, sessionId), @@ -950,6 +993,38 @@ export class SessionService { return true; } + /** + * Move a prompt terminal ledger sidecar across archive states. Unlike a + * bare rename, an existing destination does not wedge the pair forever: + * the ledger is append-only JSONL, so the source records are concatenated + * onto the destination (preserving write order) and the source is + * unlinked. Throws propagate to the caller, which owns the warn-only + * policy — a ledger problem must never block the transcript move. + */ + private moveLedgerSidecar(sourcePath: string, destinationPath: string): void { + if (!fs.existsSync(sourcePath)) { + return; + } + fs.mkdirSync(path.dirname(destinationPath), { recursive: true }); + if (!fs.existsSync(destinationPath)) { + fs.renameSync(sourcePath, destinationPath); + return; + } + // Destination already exists (e.g. a partially completed archive + // cycle): merge instead of wedging. Newline padding seals both + // boundaries so lines cannot fuse into one — the ledger reader skips + // blank lines, so a redundant newline is harmless while a fused line + // would drop both records. + const sourceContents = fs.readFileSync(sourcePath, 'utf8'); + if (sourceContents.length > 0) { + const payload = sourceContents.endsWith('\n') + ? sourceContents + : `${sourceContents}\n`; + fs.appendFileSync(destinationPath, `\n${payload}`, 'utf8'); + } + fs.unlinkSync(sourcePath); + } + private sessionFileMoveError( action: 'archive' | 'unarchive', error: unknown, @@ -1762,6 +1837,7 @@ export class SessionService { this.removeFileIfExists(archivedPath); } this.removeWorktreeSidecars(sessionId); + this.removePromptLedgers(sessionId); this.removeFileHistoryBackups(sessionId); return true; } @@ -1776,6 +1852,7 @@ export class SessionService { await this.salvageUsageBestEffort(archivedPath); this.removeFileIfExists(archivedPath); this.removeWorktreeSidecars(sessionId); + this.removePromptLedgers(sessionId); this.removeFileHistoryBackups(sessionId); return true; } catch (error) { @@ -1831,6 +1908,14 @@ export class SessionService { sessionId, 'archived', ); + const activeLedger = this.getPromptLedgerPathForState( + sessionId, + 'active', + ); + const archivedLedger = this.getPromptLedgerPathForState( + sessionId, + 'archived', + ); try { fs.renameSync(sourcePath, targetPath); } catch (error) { @@ -1843,6 +1928,13 @@ export class SessionService { `archiveSessions: failed to move worktree sidecar for ${sessionId} from ${activeSidecar} to ${archivedSidecar}: ${sidecarError}`, ); } + try { + this.moveLedgerSidecar(activeLedger, archivedLedger); + } catch (ledgerError) { + this.warn( + `archiveSessions: failed to move prompt ledger for ${sessionId} from ${activeLedger} to ${archivedLedger}: ${ledgerError}`, + ); + } archived.push(sessionId); } catch (error) { errors.push({ @@ -1908,6 +2000,21 @@ export class SessionService { `unarchiveSessions: failed to move worktree sidecar for ${sessionId} from ${archivedSidecar} to ${activeSidecar}: ${sidecarError}`, ); } + const archivedLedger = this.getPromptLedgerPathForState( + sessionId, + 'archived', + ); + const activeLedger = this.getPromptLedgerPathForState( + sessionId, + 'active', + ); + try { + this.moveLedgerSidecar(archivedLedger, activeLedger); + } catch (ledgerError) { + this.warn( + `unarchiveSessions: failed to move prompt ledger for ${sessionId} from ${archivedLedger} to ${activeLedger}: ${ledgerError}`, + ); + } unarchived.push(sessionId); } catch (error) { errors.push({ diff --git a/packages/core/src/services/usageHistoryService.test.ts b/packages/core/src/services/usageHistoryService.test.ts index 6d2ebd0dbd1..f7664fe5831 100644 --- a/packages/core/src/services/usageHistoryService.test.ts +++ b/packages/core/src/services/usageHistoryService.test.ts @@ -603,6 +603,31 @@ describe('loadUsageHistory + persistSessionUsage (issue #4994 regression)', () = expect(fs.existsSync(usagePath)).toBe(true); }); + it('rebuild excludes the prompt ledger sidecar from transcript enumeration', async () => { + plantChatJsonl('sess-real', 1600); + // The ledger sidecar shares the chats dir and ends in `.jsonl`; plant a + // summarizable transcript under a distinct sessionId and rename it to + // the sidecar name, so an accidental ingestion would surface as a + // second session. + plantChatJsonl('sess-ghost', 800); + const chatsDir = path.join( + process.env['QWEN_HOME']!, + 'projects', + 'repro-project', + 'chats', + ); + fs.renameSync( + path.join(chatsDir, 'sess-ghost.jsonl'), + path.join(chatsDir, 'sess-real.ledger.jsonl'), + ); + + const records = await loadUsageHistory(undefined, { + persistRebuild: false, + }); + expect(records).toHaveLength(1); + expect(records[0]!.sessionId).toBe('sess-real'); + }); + it('end-to-end: /stats during first turn + /clear must not 2x the session', async () => { const sessionId = 'sess-e2e'; plantChatJsonl(sessionId, 1600); diff --git a/packages/core/src/services/usageHistoryService.ts b/packages/core/src/services/usageHistoryService.ts index 8b01b13ef17..e17233baf8e 100644 --- a/packages/core/src/services/usageHistoryService.ts +++ b/packages/core/src/services/usageHistoryService.ts @@ -370,7 +370,11 @@ async function rebuildFromSessionJsonl( const chatsDir = path.join(projectsDir, projDir, 'chats'); let files: string[]; try { - files = fs.readdirSync(chatsDir).filter((f) => f.endsWith('.jsonl')); + // The prompt terminal ledger sidecar (.ledger.jsonl) is not a + // transcript — only real session JSONL files carry usage evidence. + files = fs + .readdirSync(chatsDir) + .filter((f) => f.endsWith('.jsonl') && !f.endsWith('.ledger.jsonl')); } catch (e) { debugLogger.debug( `rebuildFromSessionJsonl: cannot read chatsDir ${chatsDir}: ${e}`,