refactor(acp-example): derive llm-replay script from the session JSONL
Per a design revision, the per-scenario snapshot fixture becomes EXACTLY the persisted session JSONL (<scenario>/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.
This commit is contained in:
Tianyi Cui
@@ -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 (`<scenario>/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 `<scenario>/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
|
||||
|
||||
|
||||
@@ -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 (`<scenario>/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
|
||||
* (`<scenario>/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 })
|
||||
}
|
||||
|
||||
@@ -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<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
|
||||
}
|
||||
/** 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<StreamChunk>): Promise<StreamChunk[]> {
|
||||
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<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' },
|
||||
])
|
||||
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<StreamChunk> {
|
||||
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' } }])
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user