feat(acp-example): add record/replay llm/stream plugin for snapshot tests

Introduces examples/acp-agent/src/llm-replay.ts, a function/namespace plugin
that installs a single llm/stream waterfall listener. In record mode it tees
the real model's StreamChunks into a per-scenario llm.json (flushed atomically
after EACH stream, since the snapshot subprocess is SIGKILLed and start.ts has
no disposal path). In replay mode it short-circuits the waterfall and serves
recorded streams back positionally — the Nth stream() call gets the Nth entry —
so a snapshot test can drive the real agent with no API key.

Each fixture entry is a discriminated record {chunks|throw|hang} so it can
replay BOTH branches of the LLM failure contract (throw from stream() vs a
finish-error chunk) plus cancellation. A throw entry carries the prefix chunks
emitted before the throw, replayed before the error, so a mid-stream failure
(partial output then STREAM_CLOSED) reproduces what the loop saw live.

Fail-loud on a missing or exhausted fixture (never a silent skip). Unit tests
drive the real LlmService waterfall (record tee, ordered replay, the three
entry kinds, partial-then-throw, fail-loud, event-driven abort, HMR-safety).
Broadens the unit vitest include to examples/*/tests and registers the plugin
+ snapshot tests as knip entries. Per docs/rfc/implemented/2026-06-19.

examples/acp-agent/README.md

@@ -30,6 +30,10 @@ Add to your Zed `settings.json` under `agent_servers`:
 
 The editor sets each session's `cwd` to the project it opens; the agent's bash tools run there (see the per-session `cwd` note in `packages/acp`), so the server does not need to be launched in the workspace.
 
+## 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.
+
 ## MVP limitations
 
 The bridge supports N concurrent sessions per connection, each in its own workspace `cwd` (RFC 011). Remaining limits: text-only prompts, `additionalDirectories` rejected (a session operates in its single `cwd`), and the tool-permission gate is deferred (`TODO(rfc010-permission-gate)` — tools run with the executor's full authority). See `packages/acp/README.md` for the full contract.

examples/acp-agent/src/llm-replay.ts

@@ -0,0 +1,195 @@
+/**
+ * Record/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.
+ *
+ * It lives in the example (not packages/) because it is example/test
+ * infrastructure with one consumer, exactly like echo-agent's `mock-llm.ts`;
+ * the capability-seams rule says not to split into a published package
+ * preemptively.
+ *
+ * Plugin export shape: named `name`/`inject`/`Config`/`apply`, NO default
+ * export (the cordis Loader's `unwrapExports` does `exports.default ?? exports`,
+ * so a stray default would drop the namespace — see docs/postmortem/0001).
+ */
+
+import { existsSync, readFileSync, renameSync, writeFileSync } from 'node:fs'
+import type { Context } from 'cordis'
+import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
+import { LlmError, assertNever } from '@deepseek-ai/dsh-llm'
+
+/**
+ * One recorded model call. A discriminated union (not a bare `StreamChunk[]`)
+ * so it can faithfully replay BOTH branches of the documented LLM failure
+ * contract — an adapter may THROW from `stream()` or end with a `finish` error
+ * chunk — plus a `hang` marker for cancellation scenarios (mirrors the
+ * `MockAdapter` `hang` support in packages/agent-loop/tests).
+ *
+ * A `throw` entry carries any `chunks` the adapter emitted BEFORE it threw, so
+ * 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).
+ */
+export type ReplayEntry =
+  | { kind: 'chunks'; chunks: StreamChunk[] }
+  | { kind: 'throw'; chunks: StreamChunk[]; message: string; code: string; status?: number }
+  | { kind: 'hang' }
+
+/** 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. */
+  file: 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
+ * 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`)
+  }
+  const parsed: unknown = JSON.parse(readFileSync(file, 'utf8'))
+  if (!Array.isArray(parsed)) {
+    throw new Error(`llm-replay: fixture is not a JSON array: ${file}`)
+  }
+  // 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)
+}
+
+/** Yield a recorded stream back, honoring abort like a real adapter. */
+async function* replayEntry(entry: ReplayEntry, signal: AbortSignal | undefined): AsyncIterable<StreamChunk> {
+  switch (entry.kind) {
+    case 'chunks':
+      for (const chunk of entry.chunks) {
+        if (signal?.aborted) throw new Error('aborted')
+        yield chunk
+      }
+      return
+    case 'throw':
+      // Replay the THROW branch of the LLM contract: emit whatever the adapter
+      // streamed before it threw (so the loop sees the same partial output it
+      // saw live), then throw the recorded error (e.g. a provider 401, or a
+      // mid-stream STREAM_CLOSED after partial chunks).
+      for (const chunk of entry.chunks) {
+        if (signal?.aborted) throw new Error('aborted')
+        yield chunk
+      }
+      throw new LlmError(entry.message, entry.code, entry.status)
+    case 'hang':
+      // Replay a stream that stalls until cancelled (mirrors MockAdapter): one
+      // chunk, then wait for abort and surface it as the consumer expects.
+      yield { type: 'block-start', index: 0, blockType: 'text' }
+      yield { type: 'text-delta', index: 0, text: 'partial' }
+      await new Promise<void>((_resolve, reject) => {
+        if (signal?.aborted) { reject(new Error('aborted')); return }
+        signal?.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
+      })
+      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')
+  }
+}
+
+/**
+ * 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.
+ *
+ * Replay is POSITIONAL: the Nth `stream()` call serves the Nth fixture 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()
+    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
+      }
+      recorded.push({ kind: 'chunks', chunks })
+      flushFixture(config.file, recorded)
+    })()
+  })
+}
+
+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
+}
+
+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 })
+}

examples/acp-agent/tests/llm-replay.spec.ts

@@ -0,0 +1,262 @@
+import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
+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'
+
+/**
+ * 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.
+ */
+
+const TEXT_SCRIPT: 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' } },
+  { type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } },
+  { 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()
+  }
+
+  async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
+    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
+  }
+}
+
+let dir: string
+let file: string
+
+beforeEach(() => {
+  dir = mkdtempSync(join(tmpdir(), 'llm-replay-spec-'))
+  file = join(dir, 'llm.json')
+})
+
+afterEach(() => {
+  rmSync(dir, { recursive: true, force: true })
+})
+
+async function drain(iter: AsyncIterable<StreamChunk>): Promise<StreamChunk[]> {
+  const out: StreamChunk[] = []
+  for await (const chunk of iter) out.push(chunk)
+  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 }])
+  })
+
+  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<StreamChunk> {
+      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<StreamChunk> {
+        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' },
+    ])
+  })
+})
+
+describe('llm-replay replay mode', () => {
+  function writeFixture(entries: ReplayEntry[]): void {
+    writeFileSync(file, JSON.stringify(entries), 'utf8')
+  }
+
+  it('serves recorded chunks back in order, short-circuiting the adapter', async () => {
+    writeFixture([{ kind: 'chunks', chunks: TEXT_SCRIPT }])
+    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)
+  })
+
+  it('serves the Nth call the Nth 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 }])
+    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)
+    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 }])
+    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 })
+
+    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
+  })
+
+  it('replays a hang-entry that surfaces abort when the signal fires', async () => {
+    writeFixture([{ kind: 'hang' }])
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    installLlmReplay(ctx, { mode: 'replay', file })
+
+    const controller = new AbortController()
+    const iterator = ctx.llm.stream({ model: 'm', messages: [], signal: controller.signal })[Symbol.asyncIterator]()
+    // Deterministically consume the two pre-hang chunks (no sleep), then abort
+    // and assert the next pull rejects — event-driven, per the no-sleeps rule.
+    expect((await iterator.next()).value).toMatchObject({ type: 'block-start' })
+    expect((await iterator.next()).value).toMatchObject({ type: 'text-delta' })
+    controller.abort()
+    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 }])
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    installLlmReplay(ctx, { mode: 'replay', 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 }])
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    installLlmReplay(ctx, { mode: 'replay', 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')
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    ctx.llm.registerAdapter(['m'], new MultiScriptAdapter([TEXT_SCRIPT, TEXT_SCRIPT]))
+
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      installLlmReplay(inner, { mode: 'replay', file })
+    }, { inject: ['llm'] }))
+
+    // While installed, replay short-circuits to the fixture ('hi').
+    expect(await drain(ctx.llm.stream({ model: 'm', messages: [] }))).toEqual(TEXT_SCRIPT)
+
+    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)
+  })
+})