/** * Keyless snapshot-test LLM replay. It derives one model-call script per * recorded session from `assistant/chunk` events and explicitly marked local * compaction calls, then binds fresh live sessions to parent/child scripts by * first-call order. Throw and hang cases require an explicit override because * a session log cannot reconstruct them alone. * @module @deepseek-ai/dsh-llm-replay */ import { existsSync, readFileSync, writeFileSync } from 'node:fs' import { delimiter as pathDelimiter } from 'node:path' import type { Context } from '@deepseek-ai/cordis' import type {} from '@deepseek-ai/dsh-compaction' import { decodeStorageRecord } from '@deepseek-ai/dsh-session' import type { SessionEvent } from '@deepseek-ai/dsh-session' import type { ContentBlock, GenerateOptions, LlmModelInfo, LlmProviderInfo, LlmResolvedModelInfo, ModelModality, ResolvedRetryPolicy, RetryPolicyConfig, StreamChunk, TokenUsage, } from '@deepseek-ai/dsh-llm' import { LlmAdapter, LlmError, ReasoningEffortId, assertNever, resolveRetryPolicy } from '@deepseek-ai/dsh-llm' /** * One recorded model call. `throw` may replay prefix chunks before failing; * `hang` models cancellation. Derived chunk entries come from ordinary model * streams and complete outputs of explicitly marked local compaction calls; * an override sidecar can supply any variant. */ export type ReplayEntry = | { kind: 'chunks'; chunks: StreamChunk[] } | { kind: 'throw'; chunks: StreamChunk[]; message: string; code: string } | { kind: 'hang' /** Optional marker written after the prefix chunks are consumed and before the stream waits for cancellation. */ readyFile?: string } /** One model exposed by a replay-only provider catalog. */ export interface ReplayModelConfig { /** Model id used for replay requests. */ id: string /** Selector label; defaults to {@link id}. */ name?: string /** Optional selector description. */ description?: string /** Optional positive integer context capacity published by the replay adapter. */ contextWindow?: number /** Optional declared input modalities, so a scenario can exercise capability gates (e.g. image-capable `read_image`). */ inputModalities?: readonly ModelModality[] /** * Optional per-request output cap the replay route materializes when callers * omit one, so replay reconstructs the request header a live catalog produced. */ defaultMaxTokens?: number /** Optional reasoning-effort ids the replay route accepts, in display order. */ reasoningEfforts?: string[] /** * Optional effort materialized when callers omit one; must appear in * {@link reasoningEfforts} or call resolution rejects the route. */ defaultReasoningEffort?: string } /** One provider route exposed by the replay adapter. */ export interface ReplayProviderConfig { /** Provider route used for replay requests. */ id: string /** Selector label; defaults to {@link id}. */ name?: string /** Advisory models exposed to replay scenarios that exercise discovery. */ models?: ReplayModelConfig[] /** Optional provider-owned retry policy used by assembled recovery snapshots. */ retryPolicy?: RetryPolicyConfig } /** Resolved plugin configuration. */ export interface ReplayConfig { /** * Path to the PRIMARY (parent) `session.jsonl` fixture. For a single-session * scenario this is the only log; for a nested-agent scenario it is the parent, * and the child logs ride in {@link childFiles}. */ file: string /** * Optional sidecar for the PRIMARY session: a bare `ReplayEntry[]` replaces * the derived script; `{ patches }` keeps it and swaps the named call * indexes ({@link ReplayOverrideDoc}). Used by single-session scenarios not * expressible as `assistant/chunk` (throw-before-chunk, cancel/hang, * injected transient failures). Absent for normal and nested scenarios. */ overrideFile?: string /** * Additional recorded child-session logs (a nested-agent scenario's subagent * sessions). Each is derived independently; the full set is ordered by * `createdAt` so the parent (earliest) binds to the first live session. Empty * for a single-session scenario. */ childFiles?: string[] /** * Optional provider catalog. When non-empty, replay registers an adapter for * these routes; when absent or empty, it retains the catch-all waterfall used * by tests that do not need discovery. */ providers?: ReplayProviderConfig[] /** * Optional per-chunk pacing delay in milliseconds: each replayed chunk waits * this long before yielding, so a downstream transport (e.g. the web SSE * mux observed by a browser) sees genuinely incremental delivery. A realism * knob only — correctness must never depend on it. Absent or `0` keeps * today's synchronous burst yield. Must be a non-negative finite integer; * aborting mid-wait cancels the stream like any other abort. */ paceMs?: number } /** * Handle returned by {@link installLlmReplay}: removal plus the end-of-run * consumption check that turns silent fixture underruns (a scenario that * issued fewer calls than recorded, or never bound a recorded child script) * into a crisp diagnostic at teardown. */ export interface ReplayHandle { /** Remove the registered adapter or waterfall listener (HMR safety). Freestanding closure — safe to destructure. */ dispose(this: void): void /** * Throw unless every recorded script was bound to a live session and every * bound cursor consumed its full entry list. Call at scenario teardown. * Freestanding closure — safe to destructure. */ assertConsumed(this: void): void } /** * Recorded calls plus header facts used to order parent and child scripts. * Recorded ids are diagnostic; fresh live ids bind by ordered first use. */ export interface SessionScript { /** The recorded session id (diagnostics only — the live id differs). */ recordedId: string /** Session creation time; the deterministic ordering key (parent < child). */ createdAt: number /** The per-`stream()`-call replay entries, in recorded call order. */ entries: ReplayEntry[] /** * Whether this is the PRIMARY (parent) session. Breaks a `createdAt` tie in * favor of the parent, which always issues the first model call. */ primary: boolean } /** * 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} or a packed chunk row (expanded back into its events, so * a fixture recorded with `packChunks` on derives the same script). The header * is skipped; malformed lines fail loud. * @param text - the raw `.jsonl` file contents. * @returns every event after the header, in log order. */ export function parseSessionLog(text: string): SessionEvent[] { const lines = text.split('\n').filter(line => line.trim().length > 0) const events: SessionEvent[] = [] // The JSONL backend guarantees line 0 is the session header. for (let i = 1; i < lines.length; i++) { events.push(...decodeStorageRecord(JSON.parse(lines[i] as string))) } return events } /** * Read replay identity, ordering, and fork-seed facts from the JSONL header. * * @param text - the raw `.jsonl` file contents (only the header line is read). * @returns the header's `id`, `createdAt`, and `seedLength`, defaulted when absent. */ export function parseSessionHeader(text: string): { id: string; createdAt: number; seedLength: number } { const firstLine = text.split('\n').find(line => line.trim().length > 0) ?? '{}' const parsed = JSON.parse(firstLine) as { id?: unknown; createdAt?: unknown; seedLength?: unknown } return { id: typeof parsed.id === 'string' ? parsed.id : '', createdAt: typeof parsed.createdAt === 'number' ? parsed.createdAt : 0, seedLength: typeof parsed.seedLength === 'number' ? parsed.seedLength : 0, } } /** * Reconstruct the per-`stream()` replay script from a recorded session log. * * Splits `assistant/chunk` events at every `finish`, using turn and step changes * to detect an unterminated prior call. A `compaction/summary` explicitly marked * as one local LLM-stream call becomes a canonical successful stream from its * complete `rawOutput` at the summary's log position. A * missing assistant terminator means the live stream threw, so derivation * rejects and the scenario must provide an explicit override. Multiple calls * may share one turn and step when the loop retries. * @param events - the recorded session's events. * @returns one `chunks` entry per recorded model call, in call order. */ 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 === 'compaction/summary') { close(currentKey, current) currentKey = undefined current = [] // JSONL decoding crosses an untyped durable boundary, so retain its wider // shape even though current in-process producers enforce this correlation. const persisted: { readonly llmStreamCall?: true readonly rawOutput?: ContentBlock[] readonly usage?: TokenUsage } = event.data if (persisted.llmStreamCall === true) { if (persisted.rawOutput === undefined) { throw new Error('llm-replay: compaction/summary marks an LLM stream call without rawOutput') } const chunks: StreamChunk[] = [] for (const [index, block] of persisted.rawOutput.entries()) { chunks.push({ type: 'block-start', index, blockType: block.type }) chunks.push({ type: 'block-end', index, block }) } if (persisted.usage !== undefined) chunks.push({ type: 'usage', usage: persisted.usage }) chunks.push({ type: 'finish', reason: { kind: 'stop' } }) script.push({ kind: 'chunks', chunks }) } continue } if (event.type !== 'assistant/chunk') continue const { turn, step, chunk } = event.data const key = `${turn}/${step}` if (current.length > 0 && key !== currentKey) { close(currentKey, current) } if (current.length === 0) currentKey = key current.push(chunk) if (chunk.type === 'finish') { close(currentKey, current) currentKey = undefined current = [] } } close(currentKey, current) return script } /** * One positional patch in an augmentation sidecar: replaces the derived * entry at call index `at` (0-based) with `entry`, or appends when `at` * equals the derived length (an extra recorded-after-the-fact call, e.g. the * retry attempt following an injected transient throw). */ export interface ReplayOverridePatch { /** 0-based call index into the derived script; == length appends. */ at: number /** The replacement (or appended) entry at that call position. */ entry: ReplayEntry } /** * Override sidecar document: either a whole-script replacement (a * bare `ReplayEntry[]`) or the augmentation form `{ patches }`, which keeps * the JSONL-derived script and swaps only the named call indexes — the shape * for "turn N errors, everything else replays as recorded". */ export type ReplayOverrideDoc = ReplayEntry[] | { patches: ReplayOverridePatch[] } const REPLAY_CHUNK_TYPES = new Set([ 'block-start', 'text-delta', 'reasoning-delta', 'tool-call-delta', 'block-end', 'usage', 'finish', ]) const FROM_REQUEST_OPEN = '{{fromRequest:' const FROM_REQUEST_CLOSE = '}}' /** Collect every string leaf of one JSON-compatible value, in traversal order. */ function collectStrings(value: unknown, out: string[]): void { if (typeof value === 'string') { out.push(value) return } if (Array.isArray(value)) { for (const item of value) collectStrings(item, out) return } if (value !== null && typeof value === 'object') { for (const item of Object.values(value)) collectStrings(item, out) } } /** Resolve one placeholder pattern against the request corpus; the LAST match wins. */ function resolveFromRequest(pattern: string, corpus: string): string { let regex: RegExp try { regex = new RegExp(pattern, 'g') } catch (error) { // RegExp construction only throws SyntaxError; String() carries its message. throw new Error(`llm-replay: fromRequest has an invalid pattern ${JSON.stringify(pattern)}: ${String(error)}`) } let last: RegExpExecArray | undefined for (const match of corpus.matchAll(regex)) last = match if (last === undefined) { throw new Error(`llm-replay: fromRequest pattern ${JSON.stringify(pattern)} matched nothing in the request`) } return last[1] ?? last[0] } /** Replace every `{{fromRequest:}}` occurrence in one scripted string. */ function substituteString(text: string, corpus: string): string { let result = '' let cursor = 0 while (true) { const open = text.indexOf(FROM_REQUEST_OPEN, cursor) if (open === -1) return result + text.slice(cursor) let close = text.indexOf(FROM_REQUEST_CLOSE, open + FROM_REQUEST_OPEN.length) if (close === -1) { throw new Error(`llm-replay: fromRequest placeholder is unterminated in ${JSON.stringify(text)}`) } // The last two braces of a consecutive `}` run terminate the placeholder, // so a pattern may end with a brace quantifier like `[0-9a-f]{4}`. while (text[close + FROM_REQUEST_CLOSE.length] === '}') close += 1 const pattern = text.slice(open + FROM_REQUEST_OPEN.length, close) result += text.slice(cursor, open) + resolveFromRequest(pattern, corpus) cursor = close + FROM_REQUEST_CLOSE.length } } /** Deep-copy one JSON-compatible value with scripted placeholders resolved. */ function substituteValue(value: unknown, corpus: string): unknown { if (typeof value === 'string') { return value.includes(FROM_REQUEST_OPEN) ? substituteString(value, corpus) : value } if (Array.isArray(value)) return value.map(item => substituteValue(item, corpus)) if (value !== null && typeof value === 'object') { return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, substituteValue(item, corpus)])) } return value } /** * Resolve every `{{fromRequest:}}` placeholder in one scripted entry * against the live request. The corpus is every string leaf of the request * messages joined by newlines; the pattern's LAST corpus match wins and its * first capture group (or, without one, the whole match) substitutes in place. * Scenario sidecars use this to script arguments no static file can know, * such as a randomly minted goal id the model must echo back. A pattern that * matches nothing, an invalid pattern, and an unterminated placeholder each * fail loud. The last two braces of a consecutive `}` run terminate the * placeholder, so a pattern may end with a brace quantifier but cannot * contain `}}` followed by further pattern content. Derived entries pass * through the same resolution as sidecar entries. * @param entry - the scripted entry about to replay. * @param messages - the live request messages searched by the placeholders. * @returns the entry itself when no placeholder appears, else a resolved deep copy. */ export function resolveScriptedEntry(entry: ReplayEntry, messages: GenerateOptions['messages']): ReplayEntry { if (!JSON.stringify(entry).includes(FROM_REQUEST_OPEN)) return entry const leaves: string[] = [] collectStrings(messages, leaves) return substituteValue(entry, leaves.join('\n')) as ReplayEntry } function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value) } function hasExactKeys(value: Record, keys: readonly string[]): boolean { return Object.keys(value).length === keys.length && keys.every(key => Object.hasOwn(value, key)) } function invalidOverride(file: string, location: string, detail: string): never { throw new Error(`llm-replay: invalid override ${file}: ${location} ${detail}`) } function readChunks(value: unknown, file: string, location: string): StreamChunk[] { if (!Array.isArray(value)) invalidOverride(file, location, 'chunks must be an array') for (const [index, chunk] of value.entries()) { if (!isRecord(chunk) || typeof chunk['type'] !== 'string' || !REPLAY_CHUNK_TYPES.has(chunk['type'] as StreamChunk['type'])) { invalidOverride(file, `${location}.chunks[${index}]`, 'must have a known StreamChunk type') } } return value as StreamChunk[] } function readReplayEntry(value: unknown, file: string, location: string): ReplayEntry { if (!isRecord(value)) invalidOverride(file, location, 'must be an object') switch (value['kind']) { case 'chunks': { if (!hasExactKeys(value, ['kind', 'chunks'])) invalidOverride(file, location, 'has invalid chunks-entry fields') return { kind: 'chunks', chunks: readChunks(value['chunks'], file, location) } } case 'throw': { if (!hasExactKeys(value, ['kind', 'chunks', 'message', 'code'])) { invalidOverride(file, location, 'has invalid throw-entry fields') } if (typeof value['message'] !== 'string' || value['message'].length === 0) { invalidOverride(file, location, 'message must be a non-empty string') } if (typeof value['code'] !== 'string' || value['code'].length === 0) { invalidOverride(file, location, 'code must be a non-empty string') } return { kind: 'throw', chunks: readChunks(value['chunks'], file, location), message: value['message'], code: value['code'], } } case 'hang': { const readyFile = value['readyFile'] const keys = readyFile === undefined ? ['kind'] : ['kind', 'readyFile'] if (!hasExactKeys(value, keys)) invalidOverride(file, location, 'has invalid hang-entry fields') if (readyFile !== undefined && (typeof readyFile !== 'string' || readyFile.length === 0)) { invalidOverride(file, location, 'readyFile must be a non-empty string') } return { kind: 'hang', ...(readyFile === undefined ? {} : { readyFile }) } } default: return invalidOverride(file, location, `has unknown kind ${JSON.stringify(value['kind'])}`) } } function readOverrideDoc(value: unknown, file: string): ReplayOverrideDoc { if (Array.isArray(value)) return value.map((entry, index) => readReplayEntry(entry, file, `entry ${index}`)) if (!isRecord(value) || !hasExactKeys(value, ['patches']) || !Array.isArray(value['patches'])) { return invalidOverride(file, 'document', 'must be a ReplayEntry[] or { patches: [...] }') } return { patches: value['patches'].map((value, index): ReplayOverridePatch => { const location = `patch ${index}` if (!isRecord(value) || !hasExactKeys(value, ['at', 'entry'])) { return invalidOverride(file, location, 'must contain exactly at and entry') } const at = value['at'] if (typeof at !== 'number' || !Number.isSafeInteger(at) || at < 0) { return invalidOverride(file, location, 'at must be a non-negative safe integer') } return { at, entry: readReplayEntry(value['entry'], file, `${location}.entry`) } }), } } /** * Load the PRIMARY session's replay script: the sidecar override when present * (whole-script replacement or `{ patches }` augmentation over the derived * script), else the script derived from the session JSONL (fail-loud when the * fixture is missing). * @param config - the fixture paths; only `file` and `overrideFile` are consulted. * @returns the resolved primary-session script. */ export function loadReplayScript(config: ReplayConfig): ReplayEntry[] { if (config.overrideFile !== undefined && existsSync(config.overrideFile)) { const doc = readOverrideDoc(JSON.parse(readFileSync(config.overrideFile, 'utf8')) as unknown, config.overrideFile) if (Array.isArray(doc)) return doc const script = deriveScriptFromFile(config.file) const derivedLength = script.length const seenIndexes = new Set() for (const patch of doc.patches) { if (patch.at > derivedLength) { throw new Error( `llm-replay: override patch index ${String(patch.at)} out of range ` + `(derived script has ${derivedLength} call(s); == length appends): ${config.overrideFile}`, ) } if (seenIndexes.has(patch.at)) { throw new Error(`llm-replay: duplicate override patch index ${patch.at}: ${config.overrideFile}`) } seenIndexes.add(patch.at) script[patch.at] = patch.entry } return script } return deriveScriptFromFile(config.file) } /** Derive the primary script from the session JSONL, failing loud on a missing fixture. */ function deriveScriptFromFile(file: string): ReplayEntry[] { if (!existsSync(file)) { throw new Error(`llm-replay: fixture not found: ${file} — run \`pnpm run test:snapshot:record\` first`) } return deriveReplayScript(parseSessionLog(readFileSync(file, 'utf8'))) } /** * Load the primary and child scripts in bind order. Child derivation begins at * `seedLength` so inherited parent chunks are never replayed as child calls. * * @param config - the fixture paths: the primary log plus any recorded child logs. * @returns the primary script first, then the child scripts in bind order. */ export function loadSessionScripts(config: ReplayConfig): SessionScript[] { const primaryEntries = loadReplayScript(config) // The override path replaces the derived script but carries no header; read // the header off the JSONL when it exists, else use a stable default so an // override-only fixture (header-less) still orders first as the primary. const primaryHeader = existsSync(config.file) ? parseSessionHeader(readFileSync(config.file, 'utf8')) : { id: '', createdAt: 0 } const primary: SessionScript = { recordedId: primaryHeader.id, createdAt: primaryHeader.createdAt, entries: primaryEntries, primary: true, } const children: SessionScript[] = [] for (const childFile of config.childFiles ?? []) { if (!existsSync(childFile)) { throw new Error(`llm-replay: child fixture not found: ${childFile} — re-record the scenario`) } const text = readFileSync(childFile, 'utf8') const header = parseSessionHeader(text) // Derive the child's script from its own events only — events AT OR after the seed // boundary. const ownEvents = parseSessionLog(text).slice(header.seedLength) children.push({ recordedId: header.id, createdAt: header.createdAt, entries: deriveReplayScript(ownEvents), primary: false, }) } // Synchronous children start in creation order; the id only stabilizes timestamp ties. // XXX(concurrent-subagents): concurrent children need an explicit first-call ordinal. children.sort((a, b) => a.createdAt - b.createdAt || a.recordedId.localeCompare(b.recordedId)) return [primary, ...children] } /** Replay adapter that makes a configured provider catalog discoverable without provider I/O. */ class ReplayAdapter extends LlmAdapter { private readonly providers: ReadonlyMap constructor( providers: readonly ReplayProviderConfig[], private readonly replay: (options: GenerateOptions) => AsyncIterable, ) { super() this.providers = new Map(providers.map(provider => [provider.id, provider])) } override providerInfo(provider: string): LlmProviderInfo { const configured = this.providers.get(provider) /* v8 ignore next -- LlmRuntime only asks about routes registered from this same map. */ if (configured === undefined) return super.providerInfo(provider) return { id: provider, name: configured.name ?? provider } } override providerRetryPolicy(provider: string): ResolvedRetryPolicy | undefined { const configured = this.providers.get(provider) /* v8 ignore next -- LlmRuntime only asks about routes registered from this same map. */ if (configured === undefined) return super.providerRetryPolicy(provider) return configured.retryPolicy === undefined ? undefined : resolveRetryPolicy(configured.retryPolicy, `llm-replay: provider "${provider}" retryPolicy`) } override listModels(provider: string): Promise { const configured = this.providers.get(provider) /* v8 ignore next -- LlmRuntime only asks about routes registered from this same map. */ if (configured === undefined) return Promise.resolve([]) return Promise.resolve((configured.models ?? []).map(model => ({ provider, id: model.id, name: model.name ?? model.id, ...model.description === undefined ? {} : { description: model.description }, ...model.inputModalities === undefined ? {} : { inputModalities: [...model.inputModalities] }, }))) } override resolveModel(provider: string, model: string): Promise { const configured = this.providers.get(provider) /* v8 ignore next -- LlmRuntime only asks about routes registered from this same map. */ if (configured === undefined) return Promise.resolve({ provider, id: model, name: model }) const configuredModel = configured.models?.find(candidate => candidate.id === model) return Promise.resolve({ provider, id: model, name: configuredModel?.name ?? model, ...configuredModel?.description === undefined ? {} : { description: configuredModel.description }, ...configuredModel?.inputModalities === undefined ? {} : { inputModalities: [...configuredModel.inputModalities] }, ...configuredModel?.contextWindow === undefined ? {} : { context: { contextWindow: configuredModel.contextWindow } }, ...configuredModel?.defaultMaxTokens === undefined ? {} : { defaultMaxTokens: configuredModel.defaultMaxTokens }, ...configuredModel?.reasoningEfforts === undefined ? {} : { reasoning: { efforts: configuredModel.reasoningEfforts.map(id => ({ id: ReasoningEffortId(id), name: id })), ...configuredModel.defaultReasoningEffort === undefined ? {} : { defaultEffort: ReasoningEffortId(configuredModel.defaultReasoningEffort) }, }, }, }) } override stream(options: GenerateOptions): AsyncIterable { return this.replay(options) } } /** * Wait `paceMs` between chunk yields, aborting the wait (and the stream) the * moment the signal fires — a paced replay must cancel as promptly as a burst * one. */ function paceDelay(paceMs: number, signal: AbortSignal | undefined): Promise { return new Promise((resolve, reject) => { const timer = setTimeout(() => { signal?.removeEventListener('abort', onAbort) resolve() }, paceMs) const onAbort = (): void => { clearTimeout(timer) reject(new Error('aborted')) } signal?.addEventListener('abort', onAbort, { once: true }) }) } /** Yield a recorded stream back, honoring abort like a real adapter. */ async function* replayEntry(entry: ReplayEntry, signal: AbortSignal | undefined, paceMs: number): AsyncIterable { switch (entry.kind) { case 'chunks': for (const chunk of entry.chunks) { if (signal?.aborted) throw new Error('aborted') if (paceMs > 0) await paceDelay(paceMs, signal) 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') if (paceMs > 0) await paceDelay(paceMs, signal) yield chunk } throw new LlmError(entry.message, entry.code) 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' } if (entry.readyFile !== undefined) writeFileSync(entry.readyFile, '') await new Promise((_resolve, reject) => { if (signal?.aborted) { reject(new Error('aborted')); return } signal?.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true }) }) /* v8 ignore next -- unreachable: the hang promise only ever rejects (on abort), never resolves; control never reaches here */ return /* v8 ignore next -- sidecar entries are validated before they reach the closed local union. */ default: return assertNever(entry, 'llm-replay replay entry') } } /** * Install per-session positional replay. A newly seen live session takes the * next ordered recorded script, then advances its own cursor synchronously at * invocation time; calls without `sessionId` share one anonymous session. A * non-empty provider catalog registers a routed replay adapter; otherwise a * catch-all waterfall intercepts requests. * * @param ctx - the context whose LLM service receives the replay route or waterfall. * @param config - the resolved fixture paths (env-var defaulting is `apply`'s job). * @returns the {@link ReplayHandle} carrying the disposer and the teardown consumption check. */ export function installLlmReplay(ctx: Context, config: ReplayConfig): ReplayHandle { const paceMs = config.paceMs ?? 0 if (!Number.isInteger(paceMs) || paceMs < 0) { throw new Error(`llm-replay: paceMs must be a non-negative integer, got ${String(config.paceMs)}`) } const scripts = loadSessionScripts(config) // Live-session → its bound script + cursor. A new live session id claims the // next not-yet-bound script (scripts are in bind order); `nextScript` is the // index of the next unclaimed one. const bound = new Map() let nextScript = 0 const ANON = '\0anon\0' // the key for a call that carries no sessionId const replay = (options: GenerateOptions): AsyncIterable => { const key = options.sessionId ?? ANON let state = bound.get(key) let unrecorded = false if (state === undefined) { const script = scripts[nextScript] if (script === undefined) { // More distinct live sessions made calls than the scenario recorded — // an unrecorded subagent appeared. Defer the throw into the returned // generator (the listener must return an AsyncIterable, not throw). unrecorded = true state = { entries: [], cursor: 0 } } else { nextScript++ state = { entries: script.entries, cursor: 0 } bound.set(key, state) } } const boundState = state const seenSessions = nextScript const totalScripts = scripts.length const index = boundState.cursor++ const entry: ReplayEntry | undefined = boundState.entries[index] return (async function* () { if (unrecorded) { throw new Error( `llm-replay: a model call arrived from an unrecorded session (#${seenSessions + 1}); ` + `the scenario recorded only ${totalScripts} session(s) — re-record it`, ) } if (entry === undefined) { throw new Error( `llm-replay: script exhausted — session requested model call #${index + 1} ` + `but its script has only ${boundState.entries.length}; re-record the scenario`, ) } yield* replayEntry(resolveScriptedEntry(entry, options.messages), options.signal, paceMs) })() } const providers = config.providers ?? [] const dispose = providers.length > 0 ? ctx.llm.registerAdapter(providers.map(provider => provider.id), new ReplayAdapter(providers, replay)) : ctx.on('llm/stream', (options: GenerateOptions, _next) => replay(options)) return { dispose, assertConsumed(): void { const problems: string[] = [] if (nextScript < scripts.length) { problems.push(`${scripts.length - nextScript} recorded script(s) never bound to a live session`) } for (const [key, state] of bound) { if (state.cursor < state.entries.length) { const who = key === ANON ? 'the anonymous session' : `session ${key}` problems.push(`${who} consumed ${state.cursor}/${state.entries.length} recorded call(s)`) } } if (problems.length > 0) { throw new Error(`llm-replay: fixture not fully consumed — ${problems.join('; ')}; the scenario drove fewer model calls than recorded`) } }, } } export const name = 'llm-replay' export const inject = ['llm'] /** Plugin config: the {@link ReplayConfig} inputs, each defaulting to its `DSH_SNAPSHOT_*` env var in `apply`. */ 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 /** * Override the child-log paths; defaults to `$DSH_SNAPSHOT_CHILD_FILES` (a * path-separator-delimited list). Each is a recorded subagent session log for * a nested-agent scenario; absent/empty for a single-session scenario. */ childFiles?: string[] /** Optional replay-only provider catalog; absent or empty selects catch-all waterfall replay. */ providers?: ReplayProviderConfig[] /** Optional per-chunk pacing delay in ms (see {@link ReplayConfig.paceMs}); absent keeps burst yield. */ paceMs?: number } function validateConfiguredModalities(providers: ReplayProviderConfig[] | undefined): void { for (const provider of providers ?? []) { for (const model of provider.models ?? []) { const modalities: unknown = model.inputModalities if (modalities === undefined) continue if (!Array.isArray(modalities) || !modalities.every((modality: unknown) => modality === 'text' || modality === 'image')) { throw new Error( `llm-replay: provider "${provider.id}" model "${model.id}" inputModalities ` + 'must be an array containing only "text" and "image"', ) } } } } 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)') } validateConfiguredModalities(config.providers) const overrideFile = config.overrideFile ?? process.env.DSH_SNAPSHOT_OVERRIDE const childEnv = process.env.DSH_SNAPSHOT_CHILD_FILES const childFiles = config.childFiles ?? (childEnv !== undefined && childEnv.length > 0 ? childEnv.split(pathDelimiter) : []) installLlmReplay(ctx, { file, ...overrideFile !== undefined && overrideFile.length > 0 ? { overrideFile } : {}, ...childFiles.length > 0 ? { childFiles } : {}, ...config.providers !== undefined ? { providers: config.providers } : {}, ...config.paceMs !== undefined ? { paceMs: config.paceMs } : {}, }) }