From c182543dd51a196d4c021593624a5e4075550c71 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Fri, 19 Jun 2026 02:44:33 +0800 Subject: [PATCH] refactor(acp-example): derive llm-replay script from the session JSONL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per a design revision, the per-scenario snapshot fixture becomes EXACTLY the persisted session JSONL (/session.jsonl) rather than a hand-authored llm.json. The log already holds all LLM behavior (assistant/chunk carries every StreamChunk) AND all harness behavior (tool/call, tool/result, turn/*, usage), so one artifact drives replay and doubles as a behavioral golden. llm-replay becomes replay-only (the record-tee is removed; recording is now "run the real agent once and harvest the .jsonl", done by the harness in a later commit). deriveReplayScript(events) groups assistant/chunk by (turn,step) in log order — exact because the loop makes one ctx.llm.stream() call per step and tags each chunk with the current (turn,step). The two failure modes the log can't express (a thrown stream — no terminal finish; cancel/hang — timing) use an optional replay.override.json sidecar. Hardens against a Codex review finding: a derived group is only valid if it ends in a `finish` chunk. A group without one is the fingerprint of a thrown stream() and is NOT silently replayed as a clean stop — deriveReplayScript throws, naming the (turn,step), so a missing sidecar override fails loud. Updates the unit tests (parse/derive/load helpers, sidecar override, finish- terminated grouping, HMR), the example README, and the RFC prose to the JSONL format. Two goldens (stdout transcript + re-persisted JSONL) and the harness wiring land in the next commit. --- .../2026-06-19-acp-snapshot-tests.md | 44 ++- examples/acp-agent/README.md | 2 +- examples/acp-agent/src/llm-replay.ts | 221 ++++++----- examples/acp-agent/tests/llm-replay.spec.ts | 347 ++++++++++-------- 4 files changed, 352 insertions(+), 262 deletions(-) diff --git a/docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md b/docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md index 900f9a9cd8..db05031169 100644 --- a/docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md +++ b/docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md @@ -14,44 +14,54 @@ This RFC records the decision to add a third test tier — **snapshot tests** ## Decision -A snapshot test boots the **real** `examples/acp-agent` subprocess, drives it over real ACP stdio with a deterministic input script, and diffs its (normalized) stdout transcript against a committed golden file. The model is made deterministic by **recording its streamed responses once** against the real API and **replaying them** on every subsequent run. +A snapshot test boots the **real** `examples/acp-agent` subprocess, drives it over real ACP stdio with a deterministic input script, and diffs its (normalized) output against committed golden files. The model is made deterministic by **recording a real run's session log once** against the real API and **replaying it** on every subsequent run. The committed fixture IS the persisted session JSONL — the same append-only log the harness writes for any session. -### Record/replay at the `llm/stream` waterfall +### The fixture is the persisted session JSONL -The record/replay seam is the provider-agnostic `llm/stream` waterfall ([packages/llm/src/index.ts](../../../packages/llm/src/index.ts)), not an HTTP-record library and not an adapter swap. The agent loop calls `ctx.llm.stream()` for every model call (and `generate()` also drains `stream()` internally), so a single waterfall listener intercepts every model interaction regardless of which adapter (deepseek, pi-ai) is installed. The recorded unit is the parsed `StreamChunk` — provider-neutral, JSON-serializable, and already the unit the loop treats as "the replay record" ([packages/agent-loop/src/loop.ts](../../../packages/agent-loop/src/loop.ts)). +The per-scenario fixture is `/session.jsonl`: the exact log produced by running the scenario once against the real API (the snapshot harness harvests the file the JSONL persistence backend writes). This log already contains everything needed to reproduce the run deterministically: its `assistant/chunk` events carry every parsed `StreamChunk` (the LLM's behavior), and its `tool/call`/`tool/result`/`turn/*`/`assistant/message`/`usage` events carry the harness's behavior. One artifact captures both, and it is the format the codebase already treats as the authoritative replay record ([packages/session/src/types.ts](../../../packages/session/src/types.ts): "raw chunks are the replay record"). -A byte-level HTTP-record library (Polly/nock/MSW) was rejected: it would be adapter-specific (the two adapters share only the `StreamChunk` contract, not transport internals), it handles streaming SSE awkwardly, and it records at a lower level than the thing under test. Recording parsed chunks is simpler, robust, and human-reviewable. +An earlier draft used a hand-authored `llm.json` of model chunks; reusing the real session log instead means the fixture is a genuine product of the system (not a hand-built mock), and it doubles as a behavioral golden (see below). A byte-level HTTP-record library (Polly/nock/MSW) was rejected: adapter-specific, awkward with streaming SSE, and lower-level than the thing under test. -### Fixture entry schema honors the full LLM contract +### Replay derives the model script from the log -The LLM contract ([packages/llm/src/types.ts](../../../packages/llm/src/types.ts)) allows an adapter to report failure two ways: *throw from `stream()`*, or *end the stream with a `finish {kind:'error'|'aborted'}` chunk*. A fixture of bare `StreamChunk[]` could only replay the second. So each `llm.json` entry is a discriminated record: +The replay seam is the provider-agnostic `llm/stream` waterfall ([packages/llm/src/index.ts](../../../packages/llm/src/index.ts)) — a single listener intercepts every model call regardless of adapter (deepseek, pi-ai), because the loop routes all model calls through `ctx.llm.stream()`. The `llm-replay` plugin short-circuits that waterfall (never calls `next()`) and serves back streams reconstructed from the log: `deriveReplayScript(events)` groups `assistant/chunk` events by `(turn, step)` in log order, yielding one model stream per group. This grouping is exact because the agent loop makes **exactly one `ctx.llm.stream()` call per step** and tags every chunk with the current `(turn, step)` ([packages/agent-loop/src/loop.ts](../../../packages/agent-loop/src/loop.ts)): `step` increments once per loop iteration, so `(turn, step)` is unique per model call. A `finish {kind:'error'}` chunk is part of its group and replays naturally — no special-casing. + +### The in-memory replay entry honors the full LLM contract + +`deriveReplayScript` produces a list of `ReplayEntry`, the in-memory unit the replay listener serves positionally: ``` { kind: 'chunks', chunks: StreamChunk[] } -| { kind: 'throw', message: string, code: string, status?: number } +| { kind: 'throw', chunks: StreamChunk[], message: string, code: string, status?: number } | { kind: 'hang' } ``` -`throw` replays the thrown-error branch (e.g. a provider 401), and `hang` (one chunk, then wait for abort) replays cancellation — mirroring the `hang` marker in the existing [MockAdapter](../../../packages/agent-loop/tests/mock-adapter.ts). This keeps both contract branches exercised through the real consumer, per the "honor cross-seam contracts on BOTH sides" defensive pattern. +`chunks` is what the log derives. The other two cover the LLM contract's failure branches the log **cannot** reconstruct from `assistant/chunk` alone: a *pure throw before any chunk* (e.g. an HTTP 401 — the log holds only a `turn/end {error}`, no chunks) and a *cancel/hang* (a timing behavior, not chunk content). A scenario needing those supplies an optional `/replay.override.json` (a `ReplayEntry[]`) that **replaces** the derived script. The `throw` entry carries any prefix chunks so a mid-stream failure replays its partial output before throwing — the "honor cross-seam contracts on BOTH sides" defensive pattern. Synthesizing throw/cancel from the log's `turn/end {kind:error|aborted}` was rejected: it would couple `llm-replay` to loop-internal turn-closing semantics and the `turn/end` reason is lossy (it can't distinguish a thrown 401 from a finish-error). An explicit sidecar is the cleaner seam. ### Positional replay, one in-flight stream -Replay is positional: the Nth `stream()` call in a scenario returns the Nth recorded entry (a cursor with `shift()` semantics, like MockAdapter). This is deterministic **only when at most one model stream is in flight at a time**. The first cut runs one ACP session per scenario, which guarantees that. Multi-session concurrency (the bridge multiplexes N sessions, which can prompt concurrently) would let scheduling decide which model call consumes which entry — so concurrent-session snapshots are explicitly out of scope until fixture entries are keyed by request rather than position. A scenario whose control flow changes the number/order of model calls must be re-recorded; the cursor **fails loud on overrun** rather than silently reusing or skipping an entry. +Replay is positional: the Nth `stream()` call serves the Nth `ReplayEntry`. This is deterministic **only when at most one model stream is in flight at a time**. The first cut runs one ACP session per scenario, which guarantees that. Multi-session concurrency (the bridge multiplexes N sessions, which can prompt concurrently) would let scheduling decide which model call consumes which entry — so concurrent-session snapshots are out of scope until entries are keyed by request rather than position. A scenario whose control flow changes the number/order of model calls must be re-recorded; the cursor **fails loud on overrun** rather than silently reusing or skipping an entry. A missing `session.jsonl` in replay fails loud too ("record first") — never a silent skip. -### Record flushes per-stream, not on dispose +### Recording harvests the log; keyless replay needs a providerless config -The example subprocess is terminated with `SIGKILL` by the test teardown, and `start.ts` has no disposal path, so a `ctx.effect` disposer that flushed `llm.json` at teardown would never run. Record mode therefore flushes the fixture **atomically after each completed stream** (write-temp-then-rename). A minimal graceful-shutdown path (SIGTERM / stdin-end → `await ctx.dispose()`) is added for cleanliness, but fixture durability does not depend on it. +Recording runs the scenario with the real `llm-deepseek` adapter and the JSONL persistence backend, then copies the produced `.jsonl` into the scenario dir. Per-event appends are durable, but the harness shuts the subprocess down gracefully (close stdin → `await ctx.dispose()`) before harvesting so the final events are flushed. `llm-replay` itself does no recording — it is replay-only. -### Keyless replay needs a providerless config +`examples/base.yml` always loads `@deepseek-ai/dsh-llm-deepseek`, whose `apply` throws when no API key is present ([packages/llm-deepseek/src/index.ts](../../../packages/llm-deepseek/src/index.ts)). So replay cannot reuse the normal config — it uses a dedicated `examples/acp-agent/cordis.snapshot.yml` that omits `llm-deepseek` and installs `llm-replay` in its place. Recording uses a config that loads the real adapter (no `llm-replay`). In replay mode `start.ts` also skips `.env` loading so a stray key cannot trigger a live call. -`examples/base.yml` always loads `@deepseek-ai/dsh-llm-deepseek`, whose `apply` throws when no API key is present ([packages/llm-deepseek/src/index.ts](../../../packages/llm-deepseek/src/index.ts)). So replay cannot reuse the normal config — it uses a dedicated `examples/acp-agent/cordis.snapshot.yml` that omits `llm-deepseek` and installs the replay plugin in its place. Record mode loads the real adapter *and* the replay plugin (which tees it). In replay mode `start.ts` also skips `.env` loading so a stray key cannot trigger a live call. +### Two goldens: normalize, then snapshot -### Normalize, then snapshot parsed frames +A snapshot run asserts **two** normalized goldens, because the harness's external surfaces are distinct: -The transcript contains non-deterministic values: `randomUUID()` session ids, the temp `mkdtemp` cwd (which appears in terminal-card `_meta`), JSON-RPC ids, and timing. A pure normalization function replaces these with stable tokens (`{{sessionId}}`, `{{cwd}}`, sequenced ids; volatile numerics dropped/rounded) **before** the snapshot. The golden holds normalized, stable-stringified frames (not opaque bytes) for clean, line-diffable PRs; a separate raw-purity assertion keeps the existing guarantee that every stdout line parses as JSON (no logger leak onto the protocol channel). Vitest's `toMatchFileSnapshot` provides the golden store and the `-u`/`--update` "accept the diff" workflow. +1. The **stdout transcript** — the framed `session/update` JSON-RPC the editor sees. Catches regressions in the ACP bridge's event→update translation (`streamSessionEventUpdate`). +2. The **re-derived session JSONL** — the log the replay run itself persists, compared against the recorded fixture. Catches regressions in the loop, tool dispatch, and turn/step structure that never surface on stdout. + +The two are genuinely additive: stdout is the bridge's *lossy projection* of the log (it drops `usage`, `step/*`, exact `seq`/`time`, and renders tool I/O differently), so a loop/tool/turn-structure regression can change the JSONL while leaving the stdout projection identical, and a bridge-translation regression can change stdout while the JSONL is untouched. Asserting the JSONL equality also echoes the proposed [universal replay fixture](../proposed/2026-06-11-deterministic-and-stress-testing.md) idea. + +Both surfaces contain non-deterministic values that a pure normalization function scrubs **before** the snapshot: `randomUUID()` session ids → `{{sessionId}}`, the temp `mkdtemp` cwd → `{{cwd}}` (it appears in terminal-card `_meta` and the log header), JSON-RPC ids → a stable sequence, and the log's per-event `time` (epoch ms) + header `createdAt` dropped or zeroed. Real bash runs during replay, so the JSONL normalizer additionally stabilizes tool-output volatility (any embedded paths/pids/timestamps) — scenarios keep bash commands tightly constrained (`echo`, file writes; no `date`/`env`/background/large-output) so this surface is small. The goldens hold normalized, stable-stringified frames/events (not opaque bytes) for clean, line-diffable PRs; a separate raw-purity assertion keeps the guarantee that every stdout line parses as JSON (no logger leak onto the protocol channel). Vitest's `toMatchFileSnapshot` provides the golden store and the `-u`/`--update` "accept the diff" workflow. ### Isolation: normalization now, sandbox later + Determinism of the tool environment comes from a per-test `mkdtemp` cwd, the executor's existing secret-scrubbing env (`/KEY|SECRET|TOKEN/i`), the fresh non-login `bash -c` per call, and the normalization pass — **not** from an OS sandbox. A real rootless sandbox (bwrap on Linux, sandbox-exec/Seatbelt on macOS) is the established cross-platform pattern (Claude Code, Codex), but it is per-OS, fragile on newer kernels (Ubuntu 24.04+ AppArmor blocks unprivileged user namespaces), and unnecessary for transcript determinism. It is reserved as a future tier via the documented `BashExecutor` capability seam ([a sandboxing executor replaces dsh-bash-local without touching a tool schema](2026-06-13-capability-seams.md)) — a new `bash-*` package, not a change here. Scenarios keep bash commands tightly constrained (no `date`/`env`/background/large-output) so the temp-dir tier suffices. ### Example-local plugin, not a new package @@ -60,10 +70,10 @@ The replay plugin lives at `examples/acp-agent/src/llm-replay.ts`, referenced fr ### Two subcommands, replay in the default gate -`pnpm run test:snapshot` runs replay (keyless) and is composed into the default `pnpm run test` gate so every PR gets the regression check (the main `vitest.config.ts` include stays narrow; the gate is `test && test:snapshot`). `pnpm run test:snapshot:record` requires `DEEPSEEK_API_KEY` (loaded from repo `.env` first), hits the real API, and `--update`s both `llm.json` and the golden in one pass. Both forward a scenario filter. A missing fixture in replay **fails loud** with a "record first" message rather than self-skipping (the e2e self-skip rule is a CI-secret accommodation, not appropriate here — a committed-fixture test that silently vanishes is a coverage hole). No-model scenarios commit `llm.json` as `[]` so fail-loud and the no-model case don't conflict. An orphan-fixture guard test fails on a golden/fixture not referenced by any scenario (Vitest does not prune orphaned raw goldens). +`pnpm run test:snapshot` runs replay (keyless) and is composed into the default `pnpm run test` gate so every PR gets the regression check (the main `vitest.config.ts` include stays narrow; the gate is `test && test:snapshot`). `pnpm run test:snapshot:record` requires `DEEPSEEK_API_KEY` (loaded from repo `.env` first), hits the real API, harvests the produced `session.jsonl`, and `--update`s both goldens in one pass. Both forward a scenario filter. A missing fixture in replay **fails loud** with a "record first" message rather than self-skipping (the e2e self-skip rule is a CI-secret accommodation, not appropriate here — a committed-fixture test that silently vanishes is a coverage hole). A no-model scenario's `session.jsonl` simply has no `assistant/chunk` events (empty derived script); fail-loud still applies if a model call happens with no entry. An orphan-fixture guard test fails on a golden/fixture not referenced by any scenario (Vitest does not prune orphaned raw goldens). ## Consequences -A new test tier and its fixtures to maintain: each scenario is a directory of `input.json` + `llm.json` + `stdout.golden.txt`, committed and reviewed. Re-recording when the model's phrasing changes churns the goldens — visible in review, which is the point of committing them. Bought: deterministic, keyless, full-transcript regression coverage that boots the real Loader (so it still guards the export-shape bug class), exercises the real bash executor, and gives a one-command accept-the-diff loop. The tier is ACP-first but the harness (subprocess + tee + input-DSL + normalization + record/replay waterfall) is example-agnostic and extends to other examples. +A new test tier and its fixtures to maintain: each scenario is a directory of `input.json` (the client stdin script) + `session.jsonl` (the recorded log) + an optional `replay.override.json` + the two `*.golden` files, committed and reviewed. Re-recording when the model's phrasing changes churns the goldens — visible in review, which is the point of committing them. Bought: deterministic, keyless, full-transcript regression coverage that boots the real Loader (so it still guards the export-shape bug class), exercises the real bash executor, and gives a one-command accept-the-diff loop. The tier is ACP-first but the harness (subprocess + tee + input-DSL + normalization + JSONL-derived replay) is example-agnostic and extends to other examples. This RFC relates to but does not supersede the [proposed determinism RFC](../proposed/2026-06-11-deterministic-and-stress-testing.md): that proposal's "universal replay fixture" re-derives session *message history* after every test (an internal-consistency invariant), whereas snapshot tests pin the *external protocol output*. They are complementary — one guards the event-sourcing invariant, the other guards the editor-facing contract. diff --git a/examples/acp-agent/README.md b/examples/acp-agent/README.md index 31c81dccee..a2436dbf33 100644 --- a/examples/acp-agent/README.md +++ b/examples/acp-agent/README.md @@ -32,7 +32,7 @@ The editor sets each session's `cwd` to the project it opens; the agent's bash t ## Snapshot tests (record-once / replay-deterministic) -This example is the home of the harness's **snapshot tests** — they boot this server as a real subprocess, drive it with a deterministic input script, and diff its normalized stdout transcript against a committed golden file. The model is made deterministic by `src/llm-replay.ts`, a function/namespace plugin that installs an `llm/stream` waterfall listener: in `record` mode it tees the real model's `StreamChunk`s into a per-scenario `llm.json` (flushed atomically after each call); in `replay` mode it short-circuits the waterfall and serves those chunks back, so replay needs no API key. Each fixture entry is a discriminated record — `{ kind: 'chunks' | 'throw' | 'hang' }` — so both LLM failure branches (throw vs. finish-error) and cancellation replay faithfully. See [docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md](../../docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md) for the full design. +This example is the home of the harness's **snapshot tests** — they boot this server as a real subprocess, drive it with a deterministic input script, and diff its normalized output against committed golden files. The model is made deterministic by `src/llm-replay.ts`, a function/namespace plugin that installs an `llm/stream` waterfall listener and short-circuits it, serving model streams reconstructed from a recorded **session JSONL** fixture (`/session.jsonl`) — so replay needs no API key. The fixture IS the persisted session log: its `assistant/chunk` events carry every `StreamChunk`, so grouping them by `(turn, step)` reconstructs each `stream()` call (one model call per loop step). Recording is therefore "run the real agent once and harvest the `.jsonl`". The two failure modes not expressible as logged chunks — a pure throw before any chunk, and cancel/hang — use an optional `/replay.override.json` sidecar (a `ReplayEntry[]` that replaces the derived script). See [docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md](../../docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md) for the full design. ## MVP limitations diff --git a/examples/acp-agent/src/llm-replay.ts b/examples/acp-agent/src/llm-replay.ts index ea1c13f186..d24299b275 100644 --- a/examples/acp-agent/src/llm-replay.ts +++ b/examples/acp-agent/src/llm-replay.ts @@ -1,12 +1,25 @@ /** - * Record/replay LLM plugin for snapshot tests. + * Replay LLM plugin for snapshot tests. * - * Installs a single `llm/stream` waterfall listener that, in `record` mode, - * tees the real model's streamed {@link StreamChunk}s into a fixture file, and - * in `replay` mode short-circuits the waterfall (never calls `next()`) to yield - * previously-recorded streams deterministically. This is the seam that lets a - * snapshot test boot the real agent against a fixed model transcript with no - * API key — see docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md. + * Installs a single `llm/stream` waterfall listener that short-circuits the + * waterfall (never calls `next()`) and yields model streams reconstructed from + * a recorded **session JSONL** fixture — so a snapshot test can boot the real + * agent against a fixed model transcript with no API key. See + * docs/rfc/implemented/2026-06-19-acp-snapshot-tests.md. + * + * The fixture IS the persisted session log (`/session.jsonl`): its + * `assistant/chunk` events carry every {@link StreamChunk}, so grouping them by + * `(turn, step)` reconstructs each `stream()` call's chunk sequence (one model + * call per loop step — see packages/agent-loop/src/loop.ts). Recording is + * therefore "run the real agent once and harvest the `.jsonl`", done by the + * snapshot harness — this plugin does not record. + * + * Two failure modes are NOT reconstructable from `assistant/chunk` alone — a + * pure throw before any chunk (e.g. an HTTP 401: the log holds only a + * `turn/end {error}`, no chunks) and a cancel/hang (timing, not chunk content). + * A scenario that needs those supplies an optional sidecar + * (`/replay.override.json`: a `ReplayEntry[]`) that REPLACES the + * derived script. * * It lives in the example (not packages/) because it is example/test * infrastructure with one consumer, exactly like echo-agent's `mock-llm.ts`; @@ -18,8 +31,9 @@ * so a stray default would drop the namespace — see docs/postmortem/0001). */ -import { existsSync, readFileSync, renameSync, writeFileSync } from 'node:fs' +import { existsSync, readFileSync } from 'node:fs' import type { Context } from 'cordis' +import type { SessionEvent } from '@deepseek-ai/dsh-session' import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm' import { LlmError, assertNever } from '@deepseek-ai/dsh-llm' @@ -34,6 +48,11 @@ import { LlmError, assertNever } from '@deepseek-ai/dsh-llm' * a mid-stream transport failure (partial output then `STREAM_CLOSED`) replays * the partial chunks first and only then throws — exactly what the agent loop * saw live (it may already have emitted partial assistant chunks). + * + * The normal/finish-terminated cases are DERIVED from the session JSONL + * ({@link deriveReplayScript}); only the throw and hang cases need a + * hand-authored sidecar entry (a thrown stream leaves no terminal `finish` in + * the log, so it cannot be derived as `chunks`). */ export type ReplayEntry = | { kind: 'chunks'; chunks: StreamChunk[] } @@ -42,38 +61,103 @@ export type ReplayEntry = /** Resolved plugin configuration. */ export interface ReplayConfig { - /** `record` tees the real model to `file`; `replay` serves `file` back. */ - mode: 'record' | 'replay' - /** Path to the per-scenario `llm.json` fixture. */ + /** Path to the per-scenario `session.jsonl` fixture (the recorded log). */ file: string + /** + * Optional path to a `ReplayEntry[]` sidecar that REPLACES the derived + * script. Used by the two scenarios not expressible as `assistant/chunk` + * (pure throw-before-chunk, cancel/hang). Absent for normal scenarios. + */ + overrideFile?: string } /** - * Read and validate a fixture file. Throws a clear, fail-loud error when the - * file is missing (the scenario was never recorded) or malformed — never + * Parse a session `.jsonl` buffer into its event list. Line 0 is the session + * header (a `{type:'session',…}` record), every subsequent non-empty line is a + * {@link SessionEvent}. The header is skipped; malformed lines fail loud. + */ +export function parseSessionLog(text: string): SessionEvent[] { + const lines = text.split('\n').filter(line => line.trim().length > 0) + const events: SessionEvent[] = [] + // Skip line 0 (the header). A reader distinguishes it by its `type:'session'` + // tag; we simply drop the first line, which the JSONL backend guarantees is + // the header. + for (let i = 1; i < lines.length; i++) { + const parsed: unknown = JSON.parse(lines[i] as string) + events.push(parsed as SessionEvent) + } + return events +} + +/** + * Reconstruct the per-`stream()` replay script from a recorded session log. + * + * The agent loop makes exactly one `ctx.llm.stream()` call per step and appends + * every chunk as an `assistant/chunk` event tagged with the current + * `(turn, step)`. Grouping those events by `(turn, step)` in log order + * therefore yields one `{kind:'chunks'}` entry per model call, in call order. + * + * A group is only valid if it ends in a `finish` chunk — the adapter contract + * guarantees a successful (or finish-error) stream terminates with `finish`, + * and the loop relies on it. A group WITHOUT a terminal `finish` is the + * fingerprint of a *thrown* `stream()` (the loop recorded the prefix chunks, + * then an `error`/`turn/end`, but no `finish`): such a stream cannot be + * faithfully replayed as `{kind:'chunks'}` (that would look like a clean stop), + * so deriving it is an error — the scenario must supply a `replay.override.json` + * sidecar with an explicit `throw` (or `hang`) entry instead. {@link + * deriveReplayScript} throws, naming the offending `(turn, step)`, so a missing + * override fails loud rather than silently replaying a thrown call as success. + */ +export function deriveReplayScript(events: SessionEvent[]): ReplayEntry[] { + const script: ReplayEntry[] = [] + let currentKey: string | undefined + let current: StreamChunk[] = [] + const close = (key: string | undefined, chunks: StreamChunk[]): void => { + if (chunks.length === 0) return + if (chunks[chunks.length - 1]?.type !== 'finish') { + throw new Error( + `llm-replay: model call ${key} ended without a finish chunk (a thrown stream); ` + + 'this scenario needs a replay.override.json sidecar', + ) + } + script.push({ kind: 'chunks', chunks }) + } + for (const event of events) { + if (event.type !== 'assistant/chunk') continue + const { turn, step, chunk } = event.data + const key = `${turn}/${step}` + if (key !== currentKey) { + // A new (turn, step) — i.e. a new stream() call. Close the previous one + // (skip the initial empty buffer before any chunk has been seen). + close(currentKey, current) + currentKey = key + current = [] + } + current.push(chunk) + } + close(currentKey, current) + return script +} + +/** + * Build the replay script for a scenario: the sidecar override if present, + * otherwise the script derived from the recorded session JSONL. Fail-loud if + * the JSONL fixture is missing (the scenario was never recorded) — never * silently returns an empty script, so a coverage hole can't masquerade as a * passing replay. */ -export function loadFixture(file: string): ReplayEntry[] { - if (!existsSync(file)) { - throw new Error(`llm-replay: fixture not found: ${file} — run \`pnpm run test:snapshot:record\` first`) +export function loadReplayScript(config: ReplayConfig): ReplayEntry[] { + if (config.overrideFile !== undefined && existsSync(config.overrideFile)) { + const parsed: unknown = JSON.parse(readFileSync(config.overrideFile, 'utf8')) + if (!Array.isArray(parsed)) { + throw new Error(`llm-replay: override is not a JSON array: ${config.overrideFile}`) + } + return parsed as ReplayEntry[] } - const parsed: unknown = JSON.parse(readFileSync(file, 'utf8')) - if (!Array.isArray(parsed)) { - throw new Error(`llm-replay: fixture is not a JSON array: ${file}`) + if (!existsSync(config.file)) { + throw new Error(`llm-replay: fixture not found: ${config.file} — run \`pnpm run test:snapshot:record\` first`) } - // The fixture round-trips StreamChunk through JSON; branded CallId fields - // deserialize as plain strings, which are structurally StreamChunk. We trust - // the file shape (it is committed and produced by record mode) rather than - // deep-validating every chunk. - return parsed as ReplayEntry[] -} - -/** Atomically write the recorded entries to `file` (temp write + rename). */ -function flushFixture(file: string, entries: ReplayEntry[]): void { - const tmp = `${file}.tmp-${process.pid}` - writeFileSync(tmp, `${JSON.stringify(entries, null, 2)}\n`, { encoding: 'utf8' }) - renameSync(tmp, file) + return deriveReplayScript(parseSessionLog(readFileSync(config.file, 'utf8'))) } /** Yield a recorded stream back, honoring abort like a real adapter. */ @@ -107,70 +191,35 @@ async function* replayEntry(entry: ReplayEntry, signal: AbortSignal | undefined) return default: // Closed local union: an unknown kind means malformed (hand-edited or - // drifted) fixture data — fail loud with a runtime diagnostic. - return assertNever(entry, 'llm-replay fixture entry') + // drifted) sidecar data — fail loud with a runtime diagnostic. + return assertNever(entry, 'llm-replay replay entry') } } /** - * Install the record/replay `llm/stream` listener on `ctx`. Returns the - * listener disposer (so a fiber dispose removes it — HMR safety). Exported - * separately from {@link apply} so unit tests can drive it without the Loader - * or env vars. + * Install the replay `llm/stream` listener on `ctx`. Returns the listener + * disposer (so a fiber dispose removes it — HMR safety). Exported separately + * from {@link apply} so unit tests can drive it without the Loader or env vars. * - * Replay is POSITIONAL: the Nth `stream()` call serves the Nth fixture entry. + * Replay is POSITIONAL: the Nth `stream()` call serves the Nth script entry. * This is deterministic only with at most one model stream in flight at a time; * the snapshot harness runs one ACP session per scenario to guarantee that. The * cursor is advanced synchronously at listener-invocation time (not lazily * inside the generator) so call ORDER, not iteration order, fixes the mapping. */ export function installLlmReplay(ctx: Context, config: ReplayConfig): () => void { - if (config.mode === 'replay') { - const entries = loadFixture(config.file) - let cursor = 0 - return ctx.on('llm/stream', (_options: GenerateOptions, _next) => { - const index = cursor++ - const entry: ReplayEntry | undefined = entries[index] - return (async function* () { - if (entry === undefined) { - throw new Error( - `llm-replay: fixture exhausted — requested model call #${index + 1} but ${config.file} has only ${entries.length}; re-record the scenario`, - ) - } - yield* replayEntry(entry, _options.signal) - })() - }) - } - - // Record mode: delegate to the real adapter via next(), tee each chunk, and - // flush atomically after EACH completed stream — the subprocess is SIGKILLed - // by the test teardown and start.ts has no disposal path, so a dispose-time - // flush would never run (see the RFC). - const recorded: ReplayEntry[] = [] - return ctx.on('llm/stream', (_options: GenerateOptions, next) => { - const inner = next() + const entries = loadReplayScript(config) + let cursor = 0 + return ctx.on('llm/stream', (options: GenerateOptions, _next) => { + const index = cursor++ + const entry: ReplayEntry | undefined = entries[index] return (async function* () { - const chunks: StreamChunk[] = [] - try { - for await (const chunk of inner) { - chunks.push(chunk) - yield chunk - } - } catch (error) { - const code = error instanceof LlmError ? error.code : 'UNKNOWN' - const status = error instanceof LlmError ? error.status : undefined - const message = error instanceof Error ? error.message : String(error) - // Record the chunks emitted before the throw alongside the error, so - // replay reproduces the same partial output + failure. - const entry: ReplayEntry = status === undefined - ? { kind: 'throw', chunks, message, code } - : { kind: 'throw', chunks, message, code, status } - recorded.push(entry) - flushFixture(config.file, recorded) - throw error + if (entry === undefined) { + throw new Error( + `llm-replay: script exhausted — requested model call #${index + 1} but the fixture has only ${entries.length}; re-record the scenario`, + ) } - recorded.push({ kind: 'chunks', chunks }) - flushFixture(config.file, recorded) + yield* replayEntry(entry, options.signal) })() }) } @@ -179,17 +228,17 @@ export const name = 'llm-replay' export const inject = ['llm'] export interface Config { - /** Override the mode; defaults to `$DSH_SNAPSHOT` (`record`) else `replay`. */ - mode?: 'record' | 'replay' /** Override the fixture path; defaults to `$DSH_SNAPSHOT_FILE`. */ file?: string + /** Override the sidecar path; defaults to `$DSH_SNAPSHOT_OVERRIDE`. */ + overrideFile?: string } export function apply(ctx: Context, config: Config = {}): void { - const mode = config.mode ?? (process.env.DSH_SNAPSHOT === 'record' ? 'record' : 'replay') const file = config.file ?? process.env.DSH_SNAPSHOT_FILE if (file === undefined || file.length === 0) { throw new Error('llm-replay: a fixture path is required (Config.file or $DSH_SNAPSHOT_FILE)') } - installLlmReplay(ctx, { mode, file }) + const overrideFile = config.overrideFile ?? process.env.DSH_SNAPSHOT_OVERRIDE + installLlmReplay(ctx, overrideFile === undefined || overrideFile.length === 0 ? { file } : { file, overrideFile }) } diff --git a/examples/acp-agent/tests/llm-replay.spec.ts b/examples/acp-agent/tests/llm-replay.spec.ts index 2da8981532..9dee417686 100644 --- a/examples/acp-agent/tests/llm-replay.spec.ts +++ b/examples/acp-agent/tests/llm-replay.spec.ts @@ -3,16 +3,24 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { Context } from 'cordis' -import LlmService, { GenerateOptions, LlmAdapter, LlmError, StreamChunk } from '@deepseek-ai/dsh-llm' -import { type ReplayEntry, installLlmReplay, loadFixture } from '../src/llm-replay.ts' +import type { SessionEvent } from '@deepseek-ai/dsh-session' +import LlmService, { GenerateOptions, LlmAdapter, StreamChunk } from '@deepseek-ai/dsh-llm' +import { + type ReplayEntry, + deriveReplayScript, + installLlmReplay, + loadReplayScript, + parseSessionLog, +} from '../src/llm-replay.ts' /** - * Unit tests for the record/replay llm/stream plugin. These drive the listener - * through the REAL LlmService waterfall (not a hand-rolled stub) so they verify - * the actual seam the snapshot harness depends on. + * Unit tests for the replay llm/stream plugin. These drive the listener through + * the REAL LlmService waterfall (not a hand-rolled stub) so they verify the + * actual seam the snapshot harness depends on, plus the pure + * derive/parse/load helpers that turn a recorded session JSONL into a script. */ -const TEXT_SCRIPT: StreamChunk[] = [ +const TEXT_CHUNKS: StreamChunk[] = [ { type: 'block-start', index: 0, blockType: 'text' }, { type: 'text-delta', index: 0, text: 'hi' }, { type: 'block-end', index: 0, block: { type: 'text', text: 'hi' } }, @@ -20,19 +28,15 @@ const TEXT_SCRIPT: StreamChunk[] = [ { type: 'finish', reason: { kind: 'stop' } }, ] -/** A scripted adapter whose every call yields one of a list of scripts. */ -class MultiScriptAdapter extends LlmAdapter { - calls = 0 - constructor(private scripts: (StreamChunk[] | (() => never))[]) { - super() - } +/** Build a minimal session-JSONL string: a header line + the given events. */ +function sessionJsonl(events: SessionEvent[]): string { + const header = JSON.stringify({ type: 'session', version: 1, id: 's1', createdAt: 0 }) + return [header, ...events.map(e => JSON.stringify(e))].join('\n') + '\n' +} - async * stream(_options: GenerateOptions): AsyncIterable { - const script = this.scripts[this.calls++] - if (script === undefined) throw new Error('MultiScriptAdapter: script exhausted') - if (typeof script === 'function') return script() // throws (returns never) - yield* script - } +/** A SessionEvent of type assistant/chunk for (turn, step). */ +function chunkEvent(seq: number, turn: number, step: number, chunk: StreamChunk): SessionEvent { + return { type: 'assistant/chunk', seq, time: 0, data: { turn, step, chunk } } } let dir: string @@ -40,7 +44,7 @@ let file: string beforeEach(() => { dir = mkdtempSync(join(tmpdir(), 'llm-replay-spec-')) - file = join(dir, 'llm.json') + file = join(dir, 'session.jsonl') }) afterEach(() => { @@ -53,139 +57,183 @@ async function drain(iter: AsyncIterable): Promise { return out } -describe('llm-replay record mode', () => { - it('tees the real stream unchanged and flushes one chunks-entry per call', async () => { - const ctx = new Context() - await ctx.plugin(LlmService) - ctx.llm.registerAdapter(['m'], new MultiScriptAdapter([TEXT_SCRIPT])) - installLlmReplay(ctx, { mode: 'record', file }) - - const seen = await drain(ctx.llm.stream({ model: 'm', messages: [] })) - expect(seen).toEqual(TEXT_SCRIPT) // consumer sees the real chunks unchanged - - const fixture = loadFixture(file) - expect(fixture).toEqual([{ kind: 'chunks', chunks: TEXT_SCRIPT }]) +describe('parseSessionLog', () => { + it('skips the header line and parses each event', () => { + const events = [chunkEvent(1, 1, 1, TEXT_CHUNKS[0] as StreamChunk)] + expect(parseSessionLog(sessionJsonl(events))).toEqual(events) }) - it('flushes after EACH stream (durable without dispose)', async () => { - const ctx = new Context() - await ctx.plugin(LlmService) - ctx.llm.registerAdapter(['m'], new MultiScriptAdapter([TEXT_SCRIPT, TEXT_SCRIPT])) - installLlmReplay(ctx, { mode: 'record', file }) - - await drain(ctx.llm.stream({ model: 'm', messages: [] })) - expect(loadFixture(file)).toHaveLength(1) // already on disk, no dispose needed - await drain(ctx.llm.stream({ model: 'm', messages: [] })) - expect(loadFixture(file)).toHaveLength(2) - }) - - it('records a throw-entry then re-throws when the adapter throws', async () => { - const ctx = new Context() - await ctx.plugin(LlmService) - ctx.llm.registerAdapter(['m'], new MultiScriptAdapter([() => { throw new Error('boom') }])) - installLlmReplay(ctx, { mode: 'record', file }) - - await expect(drain(ctx.llm.stream({ model: 'm', messages: [] }))).rejects.toThrow('boom') - expect(loadFixture(file)).toEqual([{ kind: 'throw', chunks: [], message: 'boom', code: 'UNKNOWN' }]) - }) - - it('records the partial chunks emitted before a mid-stream throw', async () => { - const partial: StreamChunk[] = [ - { type: 'block-start', index: 0, blockType: 'text' }, - { type: 'text-delta', index: 0, text: 'par' }, - ] - function* chunkThenThrow(): Generator { - yield* partial - throw new LlmError('connection dropped', 'STREAM_CLOSED') - } - // An adapter that streams two chunks, then throws mid-stream. - class MidStreamThrowAdapter extends LlmAdapter { - async * stream(_options: GenerateOptions): AsyncIterable { - yield* chunkThenThrow() - } - } - const ctx = new Context() - await ctx.plugin(LlmService) - ctx.llm.registerAdapter(['m'], new MidStreamThrowAdapter()) - installLlmReplay(ctx, { mode: 'record', file }) - - const seen: StreamChunk[] = [] - await expect((async () => { - for await (const c of ctx.llm.stream({ model: 'm', messages: [] })) seen.push(c) - })()).rejects.toThrow('connection dropped') - expect(seen).toEqual(partial) // consumer saw the partial output before the throw - expect(loadFixture(file)).toEqual([ - { kind: 'throw', chunks: partial, message: 'connection dropped', code: 'STREAM_CLOSED' }, - ]) + it('ignores blank lines', () => { + const header = JSON.stringify({ type: 'session', version: 1, id: 's1', createdAt: 0 }) + const ev = chunkEvent(1, 1, 1, TEXT_CHUNKS[0] as StreamChunk) + expect(parseSessionLog(`${header}\n\n${JSON.stringify(ev)}\n\n`)).toEqual([ev]) }) }) -describe('llm-replay replay mode', () => { - function writeFixture(entries: ReplayEntry[]): void { - writeFileSync(file, JSON.stringify(entries), 'utf8') +describe('deriveReplayScript', () => { + it('groups assistant/chunk by (turn, step) into one entry per stream() call', () => { + const events: SessionEvent[] = TEXT_CHUNKS.map((c, i) => chunkEvent(i + 1, 1, 1, c)) + expect(deriveReplayScript(events)).toEqual([{ kind: 'chunks', chunks: TEXT_CHUNKS }]) + }) + + it('produces one entry per distinct (turn, step), in log order', () => { + const callA = TEXT_CHUNKS + const callB: StreamChunk[] = [ + { type: 'block-start', index: 0, blockType: 'text' }, + { type: 'text-delta', index: 0, text: 'two' }, + { type: 'finish', reason: { kind: 'stop' } }, + ] + let seq = 1 + const events: SessionEvent[] = [ + ...callA.map(c => chunkEvent(seq++, 1, 1, c)), + ...callB.map(c => chunkEvent(seq++, 1, 2, c)), // same turn, next step + ] + expect(deriveReplayScript(events)).toEqual([ + { kind: 'chunks', chunks: callA }, + { kind: 'chunks', chunks: callB }, + ]) + }) + + it('separates calls across turns too', () => { + let seq = 1 + const events: SessionEvent[] = [ + ...TEXT_CHUNKS.map(c => chunkEvent(seq++, 1, 1, c)), + ...TEXT_CHUNKS.map(c => chunkEvent(seq++, 2, 1, c)), // new turn, step resets to 1 + ] + expect(deriveReplayScript(events)).toHaveLength(2) + }) + + it('ignores non-assistant/chunk events', () => { + let seq = 1 + const events: SessionEvent[] = [ + { type: 'turn/start', seq: seq++, time: 0, data: { turn: 1, trigger: { kind: 'continuation' } } }, + ...TEXT_CHUNKS.map(c => chunkEvent(seq++, 1, 1, c)), + { type: 'turn/end', seq: seq++, time: 0, data: { turn: 1, reason: { kind: 'completed' } } }, + ] + expect(deriveReplayScript(events)).toEqual([{ kind: 'chunks', chunks: TEXT_CHUNKS }]) + }) + + it('returns an empty script for a log with no assistant/chunk events', () => { + expect(deriveReplayScript([])).toEqual([]) + }) + + it('keeps a finish-error chunk in the derived entry (replays naturally)', () => { + const errChunks: StreamChunk[] = [ + { type: 'block-start', index: 0, blockType: 'text' }, + { type: 'finish', reason: { kind: 'error', message: 'boom', code: 'X' } }, + ] + const events = errChunks.map((c, i) => chunkEvent(i + 1, 1, 1, c)) + expect(deriveReplayScript(events)).toEqual([{ kind: 'chunks', chunks: errChunks }]) + }) + + it('throws on a group that lacks a terminal finish chunk (a thrown stream)', () => { + // A thrown stream(): prefix chunks logged, then error/turn/end, NO finish. + const events: SessionEvent[] = [ + chunkEvent(1, 1, 1, { type: 'block-start', index: 0, blockType: 'text' }), + chunkEvent(2, 1, 1, { type: 'text-delta', index: 0, text: 'par' }), + { type: 'turn/end', seq: 3, time: 0, data: { turn: 1, reason: { kind: 'error', message: 'x' } } }, + ] + expect(() => deriveReplayScript(events)).toThrow(/without a finish chunk.*replay\.override\.json/s) + }) + + it('names the offending (turn, step) when a group is incomplete', () => { + const events: SessionEvent[] = [ + chunkEvent(1, 2, 3, { type: 'block-start', index: 0, blockType: 'text' }), + ] + expect(() => deriveReplayScript(events)).toThrow(/2\/3/) + }) +}) + +describe('loadReplayScript', () => { + it('derives from the session JSONL when no override is present', () => { + writeFileSync(file, sessionJsonl(TEXT_CHUNKS.map((c, i) => chunkEvent(i + 1, 1, 1, c))), 'utf8') + expect(loadReplayScript({ file })).toEqual([{ kind: 'chunks', chunks: TEXT_CHUNKS }]) + }) + + it('uses the sidecar override when present, ignoring the JSONL', () => { + writeFileSync(file, sessionJsonl([]), 'utf8') + const overrideFile = join(dir, 'replay.override.json') + const override: ReplayEntry[] = [{ kind: 'throw', chunks: [], message: '401', code: 'AUTH', status: 401 }] + writeFileSync(overrideFile, JSON.stringify(override), 'utf8') + expect(loadReplayScript({ file, overrideFile })).toEqual(override) + }) + + it('falls back to the JSONL when the override path is set but absent', () => { + writeFileSync(file, sessionJsonl(TEXT_CHUNKS.map((c, i) => chunkEvent(i + 1, 1, 1, c))), 'utf8') + expect(loadReplayScript({ file, overrideFile: join(dir, 'nope.json') })) + .toEqual([{ kind: 'chunks', chunks: TEXT_CHUNKS }]) + }) + + it('fails loud when the fixture is missing', () => { + expect(() => loadReplayScript({ file: join(dir, 'absent.jsonl') })).toThrow(/fixture not found/) + }) + + it('throws when the override is not a JSON array', () => { + writeFileSync(file, sessionJsonl([]), 'utf8') + const overrideFile = join(dir, 'replay.override.json') + writeFileSync(overrideFile, '{"not":"array"}', 'utf8') + expect(() => loadReplayScript({ file, overrideFile })).toThrow(/not a JSON array/) + }) +}) + +describe('installLlmReplay (through the real waterfall)', () => { + function writeLog(...calls: StreamChunk[][]): void { + let seq = 1 + const events: SessionEvent[] = [] + calls.forEach((chunks, step) => { + for (const c of chunks) events.push(chunkEvent(seq++, 1, step + 1, c)) + }) + writeFileSync(file, sessionJsonl(events), 'utf8') } - it('serves recorded chunks back in order, short-circuiting the adapter', async () => { - writeFixture([{ kind: 'chunks', chunks: TEXT_SCRIPT }]) + it('serves derived chunks back, short-circuiting the adapter', async () => { + writeLog(TEXT_CHUNKS) const ctx = new Context() await ctx.plugin(LlmService) // No adapter registered for 'm' — replay must not reach it. - installLlmReplay(ctx, { mode: 'replay', file }) - - const seen = await drain(ctx.llm.stream({ model: 'm', messages: [] })) - expect(seen).toEqual(TEXT_SCRIPT) + installLlmReplay(ctx, { file }) + expect(await drain(ctx.llm.stream({ model: 'm', messages: [] }))).toEqual(TEXT_CHUNKS) }) - it('serves the Nth call the Nth entry (positional)', async () => { + it('serves the Nth call the Nth derived entry (positional)', async () => { const second: StreamChunk[] = [ { type: 'block-start', index: 0, blockType: 'text' }, { type: 'text-delta', index: 0, text: 'two' }, { type: 'finish', reason: { kind: 'stop' } }, ] - writeFixture([{ kind: 'chunks', chunks: TEXT_SCRIPT }, { kind: 'chunks', chunks: second }]) + writeLog(TEXT_CHUNKS, second) const ctx = new Context() await ctx.plugin(LlmService) - installLlmReplay(ctx, { mode: 'replay', file }) - - expect(await drain(ctx.llm.stream({ model: 'm', messages: [] }))).toEqual(TEXT_SCRIPT) + installLlmReplay(ctx, { file }) + expect(await drain(ctx.llm.stream({ model: 'm', messages: [] }))).toEqual(TEXT_CHUNKS) expect(await drain(ctx.llm.stream({ model: 'm', messages: [] }))).toEqual(second) }) - it('replays a throw-entry as an LlmError with the recorded code/status', async () => { - writeFixture([{ kind: 'throw', chunks: [], message: 'unauthorized', code: 'AUTH', status: 401 }]) + it('replays a sidecar throw-entry as an LlmError with code/status, after its prefix chunks', async () => { + writeFileSync(file, sessionJsonl([]), 'utf8') + const overrideFile = join(dir, 'replay.override.json') + const partial: StreamChunk[] = [{ type: 'block-start', index: 0, blockType: 'text' }] + writeFileSync(overrideFile, JSON.stringify([ + { kind: 'throw', chunks: partial, message: 'unauthorized', code: 'AUTH', status: 401 }, + ]), 'utf8') const ctx = new Context() await ctx.plugin(LlmService) - installLlmReplay(ctx, { mode: 'replay', file }) - - await expect(drain(ctx.llm.stream({ model: 'm', messages: [] }))).rejects.toMatchObject({ - message: 'unauthorized', - code: 'AUTH', - status: 401, - }) - }) - - it('replays a throw-entry preceded by its partial chunks', async () => { - const partial: StreamChunk[] = [ - { type: 'block-start', index: 0, blockType: 'text' }, - { type: 'text-delta', index: 0, text: 'par' }, - ] - writeFixture([{ kind: 'throw', chunks: partial, message: 'dropped', code: 'STREAM_CLOSED' }]) - const ctx = new Context() - await ctx.plugin(LlmService) - installLlmReplay(ctx, { mode: 'replay', file }) + installLlmReplay(ctx, { file, overrideFile }) const seen: StreamChunk[] = [] await expect((async () => { for await (const c of ctx.llm.stream({ model: 'm', messages: [] })) seen.push(c) - })()).rejects.toThrow('dropped') - expect(seen).toEqual(partial) // partial output replayed before the throw + })()).rejects.toMatchObject({ message: 'unauthorized', code: 'AUTH', status: 401 }) + expect(seen).toEqual(partial) }) - it('replays a hang-entry that surfaces abort when the signal fires', async () => { - writeFixture([{ kind: 'hang' }]) + it('replays a sidecar hang-entry that surfaces abort when the signal fires', async () => { + writeFileSync(file, sessionJsonl([]), 'utf8') + const overrideFile = join(dir, 'replay.override.json') + writeFileSync(overrideFile, JSON.stringify([{ kind: 'hang' }]), 'utf8') const ctx = new Context() await ctx.plugin(LlmService) - installLlmReplay(ctx, { mode: 'replay', file }) + installLlmReplay(ctx, { file, overrideFile }) const controller = new AbortController() const iterator = ctx.llm.stream({ model: 'm', messages: [], signal: controller.signal })[Symbol.asyncIterator]() @@ -197,66 +245,49 @@ describe('llm-replay replay mode', () => { await expect(iterator.next()).rejects.toThrow('aborted') }) - it('fails loud when the fixture is missing', () => { - const ctx = new Context() - expect(() => installLlmReplay(ctx, { mode: 'replay', file: join(dir, 'absent.json') })) - .toThrow(/fixture not found/) - }) - - it('fails loud when the fixture is exhausted', async () => { - writeFixture([{ kind: 'chunks', chunks: TEXT_SCRIPT }]) + it('fails loud when the script is exhausted', async () => { + writeLog(TEXT_CHUNKS) const ctx = new Context() await ctx.plugin(LlmService) - installLlmReplay(ctx, { mode: 'replay', file }) - + installLlmReplay(ctx, { file }) await drain(ctx.llm.stream({ model: 'm', messages: [] })) await expect(drain(ctx.llm.stream({ model: 'm', messages: [] }))).rejects.toThrow(/exhausted/) }) it('aborts mid-replay when the signal is already set', async () => { - writeFixture([{ kind: 'chunks', chunks: TEXT_SCRIPT }]) + writeLog(TEXT_CHUNKS) const ctx = new Context() await ctx.plugin(LlmService) - installLlmReplay(ctx, { mode: 'replay', file }) - + installLlmReplay(ctx, { file }) const controller = new AbortController() controller.abort() await expect(drain(ctx.llm.stream({ model: 'm', messages: [], signal: controller.signal }))) .rejects.toThrow('aborted') }) -}) -describe('llm-replay HMR safety', () => { - it('removes the waterfall listener when the owning fiber is disposed', async () => { - writeFileSync(file, JSON.stringify([{ kind: 'chunks', chunks: TEXT_SCRIPT }]), 'utf8') + it('removes the waterfall listener when the owning fiber is disposed (HMR safety)', async () => { + writeLog(TEXT_CHUNKS, TEXT_CHUNKS) const ctx = new Context() await ctx.plugin(LlmService) - ctx.llm.registerAdapter(['m'], new MultiScriptAdapter([TEXT_SCRIPT, TEXT_SCRIPT])) + + // A real adapter to fall through to AFTER dispose, proving the listener is gone. + class FallthroughAdapter extends LlmAdapter { + async * stream(_options: GenerateOptions): AsyncIterable { + yield { type: 'finish', reason: { kind: 'stop' } } + } + } + ctx.llm.registerAdapter(['m'], new FallthroughAdapter()) const fiber = await ctx.plugin(Object.assign((inner: Context) => { - installLlmReplay(inner, { mode: 'replay', file }) + installLlmReplay(inner, { file }) }, { inject: ['llm'] })) - // While installed, replay short-circuits to the fixture ('hi'). - expect(await drain(ctx.llm.stream({ model: 'm', messages: [] }))).toEqual(TEXT_SCRIPT) + // While installed, replay short-circuits to the derived fixture ('hi'). + expect(await drain(ctx.llm.stream({ model: 'm', messages: [] }))).toEqual(TEXT_CHUNKS) await fiber.dispose() - // After dispose, the listener is gone and the call reaches the real adapter - // (also TEXT_SCRIPT here) — proving the waterfall no longer intercepts. - const afterDispose = await drain(ctx.llm.stream({ model: 'm', messages: [] })) - expect(afterDispose).toEqual(TEXT_SCRIPT) - }) -}) - -describe('loadFixture', () => { - it('throws on a non-array JSON fixture', () => { - writeFileSync(file, '{"not":"an array"}', 'utf8') - expect(() => loadFixture(file)).toThrow(/not a JSON array/) - }) - - it('reads back what was written', () => { - const entries: ReplayEntry[] = [{ kind: 'chunks', chunks: TEXT_SCRIPT }] - writeFileSync(file, JSON.stringify(entries), 'utf8') - expect(loadFixture(file)).toEqual(entries) + // After dispose the listener is gone; the call reaches the real adapter. + expect(await drain(ctx.llm.stream({ model: 'm', messages: [] }))) + .toEqual([{ type: 'finish', reason: { kind: 'stop' } }]) }) })