From 1a1ce734ba158d344c534d2e017eb0180ba99cd4 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Fri, 19 Jun 2026 01:10:30 +0800 Subject: [PATCH] feat(acp-example): add record/replay llm/stream plugin for snapshot tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 4 + examples/acp-agent/src/llm-replay.ts | 195 +++++++++++++++ examples/acp-agent/tests/llm-replay.spec.ts | 262 ++++++++++++++++++++ knip.json | 1 + vitest.config.ts | 2 +- 5 files changed, 463 insertions(+), 1 deletion(-) create mode 100644 examples/acp-agent/src/llm-replay.ts create mode 100644 examples/acp-agent/tests/llm-replay.spec.ts diff --git a/examples/acp-agent/README.md b/examples/acp-agent/README.md index 846e49fe78..31c81dccee 100644 --- a/examples/acp-agent/README.md +++ b/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. diff --git a/examples/acp-agent/src/llm-replay.ts b/examples/acp-agent/src/llm-replay.ts new file mode 100644 index 0000000000..ea1c13f186 --- /dev/null +++ b/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 { + 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((_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 }) +} diff --git a/examples/acp-agent/tests/llm-replay.spec.ts b/examples/acp-agent/tests/llm-replay.spec.ts new file mode 100644 index 0000000000..2da8981532 --- /dev/null +++ b/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 { + 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): Promise { + 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 { + 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' }, + ]) + }) +}) + +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) + }) +}) diff --git a/knip.json b/knip.json index 189da8d695..0693f71c02 100644 --- a/knip.json +++ b/knip.json @@ -7,6 +7,7 @@ "entry": [ "examples/echo-agent/src/*.ts", "examples/coding-agent/src/*.ts", + "examples/acp-agent/src/*.ts", "examples/acp-agent/tests/**/*.e2e.ts" ], "project": ["scripts/**/*.ts", "examples/**/*.ts"] diff --git a/vitest.config.ts b/vitest.config.ts index e8eb5e204e..0878477fb4 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -19,7 +19,7 @@ export default defineConfig({ // instead applies the one root map to every importer. plugins: [tsconfigPaths({ projects: ['./tsconfig.test.json'] })], test: { - include: ['packages/*/tests/**/*.spec.ts'], + include: ['packages/*/tests/**/*.spec.ts', 'examples/*/tests/**/*.spec.ts'], coverage: { provider: 'v8', // Coverage measures OUR runtime source. Types-only files carry no