Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(examples): extract reusable logic into tested packages

Browse Source 
Logic that lived under examples/ was outside the per-file 100% coverage
gate (examples/ are not workspaces) and, in the stdio-UI case, duplicated
across two examples. Move it into packages/ so it is gated and de-duped.

- packages/ui-stdio (new): unify the two diverged stdio-chat.ts copies into
  one @deepseek-ai/dsh-ui-stdio plugin (welcome/agent Config). A test-only
  I/O seam (createStdioChat(ctx, config, runtime)) keeps process streams out
  of the serializable config and makes every render/EOF/disposal branch
  unit-testable. Per-file 100%. echo/coding cordis.yml now load the package;
  both src/stdio-chat.ts deleted.
- packages/llm-replay (new): move examples/acp-agent/src/llm-replay.ts (+ its
  spec) here so its derive/parse/replay branches fall under the coverage gate.
  cordis.snapshot.yml + README rewired to the package name; added apply/env
  /assertNever/abort tests to reach per-file 100%.
- examples/{echo,coding}-agent: keyless Loader-path e2e smokes that boot the
  real cordis.yml (no key) — the guard a hand-mounted unit test cannot be for
  the unwrapExports/export-shape class (postmortem 0001). examples/AGENTS.md
  codifies the keyless+with-key smoke convention (keyless-by-nature exception
  for echo-agent).
- AGENTS.md: a scoped, removal-triggered pre-release stance (foundation over
  blast radius). packages/README.md: new rows + a FIXME to later regroup ALL
  packages into a hierarchy. Wiring: tsconfig paths/refs, publint, knip,
  module-graph.

Verified: typecheck, lint, test:coverage (887 tests, 100%), build, hygiene,
doc-sync, test:snapshot (10), test:e2e (6 keyless pass, with-key self-skip).
This commit is contained in:
Tianyi Cui
2026-06-19 12:42:28 +08:00
parent 1e09ab7204
commit 072f97c184
29 changed files with 1121 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
examples/acp-agent/README.md
View File

@@ -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 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). A scenario that needs the agent to operate on existing files ships an optional `<scenario>/workspace/` directory — the harness copies its contents into the temp cwd before the run (see `workspace-edit`). 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 `@deepseek-ai/dsh-llm-replay`, 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). A scenario that needs the agent to operate on existing files ships an optional `<scenario>/workspace/` directory — the harness copies its contents into the temp cwd before the run (see `workspace-edit`). 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

2
examples/acp-agent/cordis.snapshot.yml
View File

@@ -23,7 +23,7 @@
# The replay adapter: short-circuits llm/stream with the recorded log's chunks,
# in place of llm-deepseek.
- id: llm-replay
name: './src/llm-replay.ts'
name: '@deepseek-ai/dsh-llm-replay'
# agent-loop + persistence + the ACP bridge — shared with cordis.yml.
- id: acp-tail

244
examples/acp-agent/src/llm-replay.ts
View File

@@ -1,244 +0,0 @@
/**
* Replay LLM plugin for snapshot tests.
*
* 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`;
* 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 } 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'
/**
* 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).
*
* 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[] }
| { kind: 'throw'; chunks: StreamChunk[]; message: string; code: string; status?: number }
| { kind: 'hang' }
/** Resolved plugin configuration. */
export interface ReplayConfig {
/** 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
}
/**
* 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 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[]
}
if (!existsSync(config.file)) {
throw new Error(`llm-replay: fixture not found: ${config.file} — run \`pnpm run test:snapshot:record\` first`)
}
return deriveReplayScript(parseSessionLog(readFileSync(config.file, 'utf8')))
}
/** 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) sidecar data — fail loud with a runtime diagnostic.
return assertNever(entry, 'llm-replay replay entry')
}
}
/**
* 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 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 {
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* () {
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`,
)
}
yield* replayEntry(entry, options.signal)
})()
})
}
export const name = 'llm-replay'
export const inject = ['llm']
export interface Config {
/** 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 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)')
}
const overrideFile = config.overrideFile ?? process.env.DSH_SNAPSHOT_OVERRIDE
installLlmReplay(ctx, overrideFile === undefined || overrideFile.length === 0 ? { file } : { file, overrideFile })
}

293
examples/acp-agent/tests/llm-replay.spec.ts
View File

@@ -1,293 +0,0 @@
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 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 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_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' } },
{ type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } },
{ type: 'finish', reason: { kind: 'stop' } },
]
/** 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'
}
/** 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
let file: string
beforeEach(() => {
dir = mkdtempSync(join(tmpdir(), 'llm-replay-spec-'))
file = join(dir, 'session.jsonl')
})
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('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('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('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 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, { file })
expect(await drain(ctx.llm.stream({ model: 'm', messages: [] }))).toEqual(TEXT_CHUNKS)
})
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' } },
]
writeLog(TEXT_CHUNKS, second)
const ctx = new Context()
await ctx.plugin(LlmService)
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 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, { file, overrideFile })
const seen: StreamChunk[] = []
await expect((async () => {
for await (const c of ctx.llm.stream({ model: 'm', messages: [] })) seen.push(c)
})()).rejects.toMatchObject({ message: 'unauthorized', code: 'AUTH', status: 401 })
expect(seen).toEqual(partial)
})
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, { file, overrideFile })
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 script is exhausted', async () => {
writeLog(TEXT_CHUNKS)
const ctx = new Context()
await ctx.plugin(LlmService)
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 () => {
writeLog(TEXT_CHUNKS)
const ctx = new Context()
await ctx.plugin(LlmService)
installLlmReplay(ctx, { file })
const controller = new AbortController()
controller.abort()
await expect(drain(ctx.llm.stream({ model: 'm', messages: [], signal: controller.signal })))
.rejects.toThrow('aborted')
})
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)
// 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, { file })
}, { inject: ['llm'] }))
// 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; the call reaches the real adapter.
expect(await drain(ctx.llm.stream({ model: 'm', messages: [] })))
.toEqual([{ type: 'finish', reason: { kind: 'stop' } }])
})
})