Merge remote-tracking branch 'origin/master' into jsonl-packed-chunk-rows
Conflicts: the four generated catalog docs (regenerated over merged sources), session index.ts exports (keep chunk-rows exports + master's SessionSurface re-export), stdio/acp demo config schema and persistence wiring (thread packChunks through master's DEFAULT_PERSISTENCE_ROOT/UI shape), stdio README config table, and the jsonl spec import line. The packed-chunk fixture also gains the provenance field master made required on assistant/message.
This commit is contained in:
kingwl
@@ -1,65 +1,46 @@
|
||||
/**
|
||||
* Shared ACP snapshot subprocess harness. It boots the real agent bin through the Cordis
|
||||
* loader, drives deterministic ACP JSON-RPC over stdio, captures protocol-pure stdout, and
|
||||
* harvests persisted session logs after graceful shutdown. Normalization stays in
|
||||
* `normalize.ts`; suite registration stays in `suite.ts`.
|
||||
* Shared subprocess harness for ACP snapshot suites. A library module driven by
|
||||
* the suite factory in ./suite.ts (and directly by harness-level specs); each
|
||||
* example's `*.snapshot.ts` names its own agent-under-test paths.
|
||||
*
|
||||
* It boots the REAL agent bin subprocess via the cordis Loader (so the
|
||||
* export-shape bug class stays guarded — see docs/postmortem/0001), drives it
|
||||
* over real ACP JSON-RPC stdio with a deterministic input script, tees raw
|
||||
* stdout (for the expected-output and purity checks) into an SDK `ClientSideConnection`,
|
||||
* and — in record mode — harvests the persisted session JSONL after a graceful
|
||||
* shutdown flush. The pure normalizers in ./normalize.ts turn the captured
|
||||
* stdout frames and the session-log events into stable, snapshot-able text.
|
||||
*
|
||||
* See .agents/notes/implemented/testing/2026-06-19-acp-snapshot-tests.md.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-acp-snapshot/harness
|
||||
*/
|
||||
|
||||
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
|
||||
import { cp, mkdtemp, readFile, readdir, rm } from 'node:fs/promises'
|
||||
import { existsSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join, delimiter } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { Readable, Writable } from 'node:stream'
|
||||
import {
|
||||
ClientSideConnection,
|
||||
ndJsonStream,
|
||||
PROTOCOL_VERSION,
|
||||
type Agent as AcpAgent,
|
||||
type Client,
|
||||
type RequestPermissionRequest,
|
||||
type RequestPermissionResponse,
|
||||
type SessionNotification,
|
||||
} from '@agentclientprotocol/sdk'
|
||||
import { launchAcpTestAgent, type AgentUnderTest, type LaunchedAcpTestAgent } from './launcher.ts'
|
||||
|
||||
// Resolve tsx's ESM loader to an ABSOLUTE path once: the child runs with its
|
||||
// cwd in a temp dir OUTSIDE the repo, where a bare `--import tsx` would not
|
||||
// resolve from node_modules. import.meta.resolve gives this package's tsx
|
||||
// regardless of the child cwd.
|
||||
const tsxLoader = fileURLToPath(import.meta.resolve('tsx'))
|
||||
export type { AgentUnderTest } from './launcher.ts'
|
||||
|
||||
/**
|
||||
* The agent composition a scenario runs against: which bin to boot and which
|
||||
* leaf config it loads. All paths are ABSOLUTE — the subprocess cwd is a temp
|
||||
* dir outside the repo, so relative resolution would miss; a suite resolves
|
||||
* them from its own `import.meta.url`.
|
||||
*/
|
||||
export interface AgentUnderTest {
|
||||
/** The agent bin entry (e.g. `packages/examples/acp-demo/src/bin.ts`), run unbuilt via tsx. */
|
||||
binScript: string
|
||||
/**
|
||||
* The example's live `cordis.yml`. Under `DSH_SNAPSHOT=replay` the bin swaps
|
||||
* it for the sibling `cordis.snapshot.yml` (the keyless replay overlay), so
|
||||
* one path serves both modes.
|
||||
*/
|
||||
configPath: string
|
||||
/**
|
||||
* The repo-root tsconfig whose `paths` map resolves the unbuilt workspace
|
||||
* imports. Passed to the child as `TSX_TSCONFIG_PATH`: tsx finds a tsconfig
|
||||
* by searching UP from the child's cwd — a temp dir outside the repo — so
|
||||
* without the explicit pin the dsh-* imports fail before the bin writes a
|
||||
* byte.
|
||||
*/
|
||||
tsconfigPath: string
|
||||
}
|
||||
|
||||
/**
|
||||
* One step of a scenario's deterministic input script (`input.json`). The harness interprets
|
||||
* these in order. `newSession` captures the server-issued (random) session id into a
|
||||
* `{{sessionId}}` variable that later steps reference. `promptAndCancel` sends without awaiting,
|
||||
* waits for the first streamed message, then cancels, making transcript order deterministic.
|
||||
* One step of a scenario's deterministic input script (`input.json`). The
|
||||
* harness interprets these in order. `newSession` captures the server-issued
|
||||
* (random) session id into a `{{sessionId}}` variable that later steps
|
||||
* reference, since a committed file cannot know the id in advance.
|
||||
*
|
||||
* `promptAndCancel` starts a prompt without awaiting completion, waits until
|
||||
* the client observes the selected update (`agent_message_chunk` by default),
|
||||
* then cancels and awaits completion. A named `waitForToolCallUpdate` keeps the
|
||||
* step open for a terminal tool update that may follow the prompt response.
|
||||
*/
|
||||
export type InputStep =
|
||||
| { op: 'initialize'; terminalOutput?: boolean }
|
||||
@@ -67,7 +48,12 @@ export type InputStep =
|
||||
| { op: 'newSessionExpectError'; additionalDirectories?: string[] }
|
||||
| { op: 'prompt'; text: string }
|
||||
| { op: 'promptExpectError'; text: string }
|
||||
| { op: 'promptAndCancel'; text: string }
|
||||
| {
|
||||
op: 'promptAndCancel'
|
||||
text: string
|
||||
afterUpdate?: 'agent_message_chunk' | 'tool_call'
|
||||
waitForToolCallUpdate?: string
|
||||
}
|
||||
| { op: 'cancel' }
|
||||
| { op: 'setConfigOption'; configId: string; value: string }
|
||||
| { op: 'setConfigOptionExpectError'; configId: string; value: string }
|
||||
@@ -76,9 +62,16 @@ export type InputStep =
|
||||
export interface InputScript {
|
||||
steps: InputStep[]
|
||||
/**
|
||||
* FIFO permission answers selected by stable option kind; the harness maps each kind to the
|
||||
* agent-issued option id. Exhaustion cancels, while a kind the agent did not offer fails the
|
||||
* scenario.
|
||||
* Ordered answers for the agent's `session/request_permission` round-trips,
|
||||
* consumed FIFO — the Nth request gets the Nth answer. Each answer selects
|
||||
* by option KIND: option ids are agent-issued randoms a committed script
|
||||
* cannot know, while kinds are the ACP-stable vocabulary, so the client maps
|
||||
* kind → the offered `optionId` at answer time. A request beyond the queue
|
||||
* (or with no queue at all) is answered `cancelled` — the stub behavior a
|
||||
* scenario without approvals relies on. A scripted kind the request does
|
||||
* not offer REJECTS the run: the scenario scripted an impossible click,
|
||||
* and {@link runScenario} throws once the in-flight step settles (the
|
||||
* agent itself just sees `cancelled`, so it cannot absorb the bug).
|
||||
*/
|
||||
permissionAnswers?: PermissionAnswer[]
|
||||
}
|
||||
@@ -168,25 +161,26 @@ export interface RunOptions {
|
||||
export async function runScenario(input: InputScript, opts: RunOptions): Promise<RunResult> {
|
||||
const cwd = await mkdtemp(join(tmpdir(), 'acp-snap-cwd-'))
|
||||
const sessionsRoot = await mkdtemp(join(tmpdir(), 'acp-snap-sessions-'))
|
||||
// Everything past the temp-dir creation runs under a try/finally that always
|
||||
// removes both dirs — so a failure in workspace seeding, spawn, or any step
|
||||
// never leaks them (the "e2e tests own their resources" rule).
|
||||
let child: ChildProcessWithoutNullStreams | undefined
|
||||
// Fixed path length: spill-policy budgets the preview against the REAL path
|
||||
// before stdout normalization, so tmpdir() length differences churn expected outputs.
|
||||
const spillRoot = '/tmp/dsh-acp-snapshot-spill'
|
||||
// Everything past the temp-dir creation is followed by failure-safe cleanup,
|
||||
// so a failure in workspace seeding, spawn, or any step never leaks resources.
|
||||
let launched: LaunchedAcpTestAgent | undefined
|
||||
let sessionId: string | undefined
|
||||
let sessionLogs: HarvestedLog[] = []
|
||||
const rawBuffers: Buffer[] = []
|
||||
const stderrChunks: string[] = []
|
||||
try {
|
||||
const outcome = await (async (): Promise<RunResult> => {
|
||||
// Seed the workspace if the scenario ships one (a file the agent reads/edits).
|
||||
// Copied into the temp cwd so the agent's bash tools see it; the expected outputs
|
||||
// normalize the cwd, so the seeded paths stay stable across runs.
|
||||
if (opts.workspaceDir !== undefined && existsSync(opts.workspaceDir)) {
|
||||
await cp(opts.workspaceDir, cwd, { recursive: true })
|
||||
}
|
||||
const env: NodeJS.ProcessEnv = {
|
||||
...process.env,
|
||||
TSX_TSCONFIG_PATH: opts.agent.tsconfigPath,
|
||||
DSH_SNAPSHOT: opts.mode,
|
||||
DSH_SNAPSHOT_FILE: opts.fixtureFile,
|
||||
DSH_SNAPSHOT_SESSIONS_ROOT: sessionsRoot,
|
||||
DSH_SNAPSHOT_SPILL_ROOT: spillRoot,
|
||||
DSH_HOME: join(cwd, '.dsh'),
|
||||
DSH_AGENTS_HOME: join(cwd, '.agents'),
|
||||
...opts.overrideFile !== undefined ? { DSH_SNAPSHOT_OVERRIDE: opts.overrideFile } : {},
|
||||
@@ -195,58 +189,22 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
|
||||
: {},
|
||||
}
|
||||
|
||||
child = spawn(
|
||||
process.execPath,
|
||||
['--import', tsxLoader, opts.agent.binScript, '--config', opts.configPath ?? opts.agent.configPath],
|
||||
{ cwd, env, stdio: ['pipe', 'pipe', 'pipe'] },
|
||||
)
|
||||
|
||||
child.stderr.setEncoding('utf8')
|
||||
child.stderr.on('data', (c: string) => stderrChunks.push(c))
|
||||
|
||||
// Tee the same raw bytes to the golden and SDK client. Decode once at the end so a UTF-8
|
||||
// sequence split across stream chunks cannot corrupt the transcript.
|
||||
const passthrough = new Readable({ read() {} })
|
||||
child.stdout.on('data', (buf: Buffer) => {
|
||||
rawBuffers.push(buf)
|
||||
passthrough.push(buf)
|
||||
})
|
||||
child.stdout.on('end', () => passthrough.push(null))
|
||||
|
||||
const stream = ndJsonStream(
|
||||
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
|
||||
Readable.toWeb(passthrough) as ReadableStream<Uint8Array>,
|
||||
)
|
||||
// Watcher so a step can block until the client OBSERVES a particular
|
||||
// session/update — used by promptAndCancel to pin frame order (send cancel
|
||||
// only after the streamed agent_message_chunk has arrived, so those frames
|
||||
// deterministically precede the cancelled prompt response).
|
||||
const updateWaiters: { match: (u: SessionNotification['update']) => boolean; resolve: () => void }[] = []
|
||||
const waitForUpdate = (match: (u: SessionNotification['update']) => boolean): Promise<void> =>
|
||||
new Promise<void>(resolve => updateWaiters.push({ match, resolve }))
|
||||
|
||||
// Permission answers are consumed FIFO across the whole run; exhaustion
|
||||
// falls back to `cancelled` so approval-free scenarios keep the plain stub.
|
||||
const permissionQueue = [...input.permissionAnswers ?? []]
|
||||
// A callback throw would become only an RPC error the agent could absorb. Record an
|
||||
// impossible permission choice here, answer cancelled, and fail the outer scenario.
|
||||
// A scenario bug detected inside a client callback (a scripted permission
|
||||
// kind the agent never offered). It cannot fail the run from in there: a
|
||||
// callback throw only becomes a JSON-RPC error RESPONSE to the agent, and
|
||||
// a tolerant agent treats that as a denial and carries on — the run (or
|
||||
// worse, a record) would absorb the impossible click silently. So the
|
||||
// callback answers `cancelled` (a well-defined path for the agent),
|
||||
// captures the error here, and the step loop fails the run on it.
|
||||
let scriptError: Error | undefined
|
||||
const makeClient = (_agent: AcpAgent): Client => ({
|
||||
sessionUpdate(params: SessionNotification): Promise<void> {
|
||||
for (let i = updateWaiters.length - 1; i >= 0; i--) {
|
||||
const waiter = updateWaiters[i]
|
||||
// The index is always in-bounds (i only decreases; splice removes at
|
||||
// i, so lower entries stay valid); the guard satisfies
|
||||
// noUncheckedIndexedAccess.
|
||||
/* v8 ignore next 1 -- unreachable in-bounds guard, see above */
|
||||
if (waiter === undefined) continue
|
||||
if (waiter.match(params.update)) {
|
||||
updateWaiters.splice(i, 1)
|
||||
waiter.resolve()
|
||||
}
|
||||
}
|
||||
return Promise.resolve()
|
||||
},
|
||||
launched = launchAcpTestAgent({
|
||||
agent: opts.agent,
|
||||
cwd,
|
||||
...opts.configPath !== undefined ? { configPath: opts.configPath } : {},
|
||||
env,
|
||||
requestPermission(params: RequestPermissionRequest): Promise<RequestPermissionResponse> {
|
||||
const answer = permissionQueue.shift()
|
||||
if (answer === undefined) return Promise.resolve({ outcome: { outcome: 'cancelled' } })
|
||||
@@ -264,10 +222,12 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
|
||||
return Promise.resolve({ outcome: { outcome: 'selected', optionId: option.optionId } })
|
||||
},
|
||||
})
|
||||
const client = new ClientSideConnection(makeClient, stream)
|
||||
const active = launched
|
||||
await active.spawned
|
||||
const { client } = active
|
||||
|
||||
for (const step of input.steps) {
|
||||
await runStep(client, step, cwd, waitForUpdate, () => sessionId, (id) => { sessionId = id })
|
||||
await runStep(client, step, cwd, match => active.waitForUpdate(match), () => sessionId, (id) => { sessionId = id })
|
||||
// A permission exchange happens while a step's request is in flight, so
|
||||
// by the time the step settles any script bug it exposed is captured —
|
||||
// fail the run HERE, as a harness error, rather than hoping the agent's
|
||||
@@ -276,30 +236,57 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
|
||||
}
|
||||
// Done driving: close stdin so the server disposes gracefully (flushing
|
||||
// persistence) and exits. Then await exit so the harvested log is complete.
|
||||
child.stdin.end()
|
||||
await waitForExit(child)
|
||||
await active.close()
|
||||
// Harvest EVERY persisted log (parent + any subagent children) while the
|
||||
// temp dirs still exist, ordered primary-first.
|
||||
sessionLogs = await harvestSessionLogs(sessionsRoot)
|
||||
} finally {
|
||||
// Failure-safe teardown: kill a still-running child and drop the temp dirs
|
||||
// even if seeding/spawn/a step/harvest threw, so a flaky run never leaks a
|
||||
// process or dir. `child` is undefined only if spawn itself threw.
|
||||
if (child !== undefined && child.exitCode === null && child.signalCode === null) {
|
||||
child.kill('SIGKILL')
|
||||
await waitForExit(child)
|
||||
return {
|
||||
rawStdout: launched.rawStdout(),
|
||||
stderr: launched.stderr(),
|
||||
cwd,
|
||||
...sessionId !== undefined ? { sessionId } : {},
|
||||
sessionLogs,
|
||||
}
|
||||
await rm(cwd, { recursive: true, force: true })
|
||||
await rm(sessionsRoot, { recursive: true, force: true })
|
||||
}
|
||||
})().then(
|
||||
value => ({ status: 'fulfilled', value } as const),
|
||||
(error: unknown) => {
|
||||
const stderr = launched?.stderr() ?? ''
|
||||
return {
|
||||
status: 'rejected',
|
||||
error: stderr === ''
|
||||
? error
|
||||
: new Error(`snapshot-harness: scenario failed: ${String(error)}\nagent stderr:\n${stderr}`, { cause: error }),
|
||||
} as const
|
||||
},
|
||||
)
|
||||
|
||||
return {
|
||||
rawStdout: Buffer.concat(rawBuffers).toString('utf8'),
|
||||
stderr: stderrChunks.join(''),
|
||||
cwd,
|
||||
...sessionId !== undefined ? { sessionId } : {},
|
||||
sessionLogs,
|
||||
// Failure-safe teardown: wait for a still-running child, then attempt every
|
||||
// owned-path removal even when an earlier cleanup rejects. Report every
|
||||
// teardown failure alongside a scenario failure so neither orthogonal
|
||||
// outcome hides the other.
|
||||
const cleanupResults: PromiseSettledResult<unknown>[] = []
|
||||
const cleanup = async (action: () => Promise<unknown>): Promise<void> => {
|
||||
cleanupResults.push(...await Promise.allSettled([action()]))
|
||||
}
|
||||
/* v8 ignore next 1 -- launch itself can only throw on a defensive synchronous spawn API failure */
|
||||
await cleanup(() => launched?.close('SIGKILL') ?? Promise.resolve())
|
||||
await cleanup(() => rm(cwd, { recursive: true, force: true }))
|
||||
await cleanup(() => rm(sessionsRoot, { recursive: true, force: true }))
|
||||
await cleanup(() => rm(spillRoot, { recursive: true, force: true }))
|
||||
|
||||
const cleanupFailures = cleanupResults
|
||||
.filter((result): result is PromiseRejectedResult => result.status === 'rejected')
|
||||
.map(result => result.reason as unknown)
|
||||
if (cleanupFailures.length > 0) {
|
||||
throw new AggregateError(
|
||||
outcome.status === 'rejected' ? [outcome.error, ...cleanupFailures] : cleanupFailures,
|
||||
outcome.status === 'rejected'
|
||||
? 'snapshot scenario and cleanup failed'
|
||||
: 'snapshot cleanup failed',
|
||||
)
|
||||
}
|
||||
if (outcome.status === 'rejected') throw outcome.error
|
||||
return outcome.value
|
||||
}
|
||||
|
||||
/** Drive one input step over the client connection. */
|
||||
@@ -307,7 +294,7 @@ async function runStep(
|
||||
client: ClientSideConnection,
|
||||
step: InputStep,
|
||||
cwd: string,
|
||||
waitForUpdate: (match: (u: SessionNotification['update']) => boolean) => Promise<void>,
|
||||
waitForUpdate: (match: (u: SessionNotification['update']) => boolean) => Promise<SessionNotification['update']>,
|
||||
getSessionId: () => string | undefined,
|
||||
setSessionId: (id: string) => void,
|
||||
): Promise<void> {
|
||||
@@ -324,8 +311,10 @@ async function runStep(
|
||||
return
|
||||
}
|
||||
case 'newSessionExpectError': {
|
||||
// The bridge rejects a session/new that widens the workspace scope (non-empty
|
||||
// additionalDirectories / mcpServers — unimplemented).
|
||||
// The bridge rejects a session/new that widens the workspace scope
|
||||
// (non-empty additionalDirectories / mcpServers — unimplemented). The SDK
|
||||
// surfaces that as a rejected RPC; swallow it so the run completes and the
|
||||
// error frame is captured in the transcript.
|
||||
await client.newSession({
|
||||
cwd,
|
||||
mcpServers: [],
|
||||
@@ -345,8 +334,10 @@ async function runStep(
|
||||
case 'promptExpectError': {
|
||||
const sessionId = getSessionId()
|
||||
if (sessionId === undefined) throw new Error('snapshot-harness: promptExpectError before newSession')
|
||||
// The model fails this turn (a recorded provider error), so the bridge answers the prompt
|
||||
// with a JSON-RPC error and the SDK rejects.
|
||||
// The model fails this turn (a recorded provider error), so the bridge
|
||||
// answers the prompt with a JSON-RPC error and the SDK rejects. That
|
||||
// rejection IS the expected editor experience — swallow it so the run
|
||||
// completes and the stdout transcript (the error frame) is captured.
|
||||
await client.prompt({ sessionId, prompt: [{ type: 'text', text: step.text }] })
|
||||
.then(() => { throw new Error('snapshot-harness: expected the prompt to fail but it succeeded') },
|
||||
() => { /* expected: the turn failed and the bridge returned an error */ })
|
||||
@@ -355,12 +346,19 @@ async function runStep(
|
||||
case 'promptAndCancel': {
|
||||
const sessionId = getSessionId()
|
||||
if (sessionId === undefined) throw new Error('snapshot-harness: promptAndCancel before newSession')
|
||||
// A hang fixture never resolves alone. Wait for its streamed chunk before cancellation
|
||||
// so updates deterministically precede the cancelled prompt response.
|
||||
// Dispatch without awaiting because the fixture does not settle on its
|
||||
// own. Waiting for the selected update pins it before cancellation and
|
||||
// the cancelled prompt response in the transcript.
|
||||
const promptDone = client.prompt({ sessionId, prompt: [{ type: 'text', text: step.text }] })
|
||||
await waitForUpdate(u => u.sessionUpdate === 'agent_message_chunk')
|
||||
const afterUpdate = step.afterUpdate ?? 'agent_message_chunk'
|
||||
await waitForUpdate(u => u.sessionUpdate === afterUpdate)
|
||||
// Arm this before cancellation so a fast tool drain cannot outrun the waiter.
|
||||
const toolCallUpdateDone = step.waitForToolCallUpdate === undefined
|
||||
? undefined
|
||||
: waitForUpdate(u => u.sessionUpdate === 'tool_call_update' && u.toolCallId === step.waitForToolCallUpdate)
|
||||
await client.cancel({ sessionId })
|
||||
await promptDone
|
||||
if (toolCallUpdateDone !== undefined) await toolCallUpdateDone
|
||||
return
|
||||
}
|
||||
case 'cancel': {
|
||||
@@ -392,16 +390,6 @@ async function runStep(
|
||||
}
|
||||
}
|
||||
|
||||
/** Resolve once the child process exits (any code/signal). */
|
||||
function waitForExit(child: ChildProcessWithoutNullStreams): Promise<void> {
|
||||
// Race guard: both call sites run within one synchronous frame of
|
||||
// stdin.end()/kill(), so the exit event cannot have been delivered yet;
|
||||
// kept for any future caller that awaits in between.
|
||||
/* v8 ignore next 1 -- unreachable race guard, see above */
|
||||
if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve()
|
||||
return new Promise<void>(resolve => child.once('exit', () => { resolve() }))
|
||||
}
|
||||
|
||||
/**
|
||||
* Harvest EVERY persisted `.jsonl` session log under a sessions root, parse each
|
||||
* header line, and return them ordered primary-first: the top-level session (no
|
||||
@@ -442,8 +430,14 @@ async function harvestSessionLogs(root: string): Promise<HarvestedLog[]> {
|
||||
})
|
||||
}
|
||||
}
|
||||
// Match replay fixture assignment: primary first, then children by creation time, with id as
|
||||
// a deterministic collision tiebreaker.
|
||||
// Primary (no parentSession) first, then children by ascending createdAt. A
|
||||
// scenario has exactly one top-level session. In the synchronous cut sibling
|
||||
// children are created strictly sequentially, so their createdAt values are
|
||||
// strictly ordered; the recordedId tiebreak only keeps a degenerate
|
||||
// same-millisecond collision (unreachable here) deterministic. This harvest
|
||||
// order must match the replay load order in dsh-llm-replay's loadSessionScripts
|
||||
// so session.<n>.jsonl maps to the same child on record and replay — replay
|
||||
// re-sorts childFiles by the same key, so the two stay consistent.
|
||||
logs.sort((a, b) => {
|
||||
const ap = a.parentSession === undefined ? 0 : 1
|
||||
const bp = b.parentSession === undefined ? 0 : 1
|
||||
|
||||
@@ -1,13 +1,23 @@
|
||||
/**
|
||||
* ACP snapshot suite kit: subprocess scenario harness, pure golden normalizers, and the Vitest
|
||||
* suite factory behind `pnpm run test:snapshot`. Because this entry exports `suite.ts`, importing
|
||||
* it requires a Vitest run.
|
||||
* ACP snapshot suite kit — the shared machinery behind the keyless snapshot
|
||||
* tier (`pnpm run test:snapshot`). Four layers, composable per example: the
|
||||
* shared subprocess/client launcher ({@link launchAcpTestAgent}), the scripted
|
||||
* scenario harness ({@link runScenario}), the pure expected-output normalizers
|
||||
* ({@link normalizeStdout} / {@link normalizeSessionLog} /
|
||||
* {@link scrubRequestHeaders} / {@link scrubSystemPrompts}), and the suite
|
||||
* factory ({@link defineAcpSnapshotSuite}) that registers a scenario table as a
|
||||
* full describe/it tree. Ordinary ACP e2e tests can use the launcher directly;
|
||||
* an example's `*.snapshot.ts` supplies only its {@link AgentUnderTest} paths,
|
||||
* snapshots directory, and {@link Scenario} table.
|
||||
*
|
||||
* NOTE: ./suite.ts imports vitest, so this package is importable only inside a
|
||||
* vitest run — a support-tier constraint stated in the README.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-acp-snapshot
|
||||
*/
|
||||
|
||||
export {
|
||||
runScenario,
|
||||
type AgentUnderTest,
|
||||
type HarvestedLog,
|
||||
type InputScript,
|
||||
type InputStep,
|
||||
@@ -15,6 +25,12 @@ export {
|
||||
type RunOptions,
|
||||
type RunResult,
|
||||
} from './harness.ts'
|
||||
export {
|
||||
launchAcpTestAgent,
|
||||
type AcpTestLaunchOptions,
|
||||
type AgentUnderTest,
|
||||
type LaunchedAcpTestAgent,
|
||||
} from './launcher.ts'
|
||||
export {
|
||||
normalizeSessionLog,
|
||||
normalizeStdout,
|
||||
|
||||
276
packages/support/acp-snapshot/src/launcher.ts
Normal file
276
packages/support/acp-snapshot/src/launcher.ts
Normal file
@@ -0,0 +1,276 @@
|
||||
/**
|
||||
* Shared launcher for ACP tests that drive an agent subprocess over JSON-RPC
|
||||
* stdio. It owns source-or-built launch resolution, workspace environment,
|
||||
* stdout tee, SDK client, update collection, permission fallback, and process
|
||||
* shutdown so e2e and snapshot suites do not each reconstruct that boundary.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-acp-snapshot/launcher
|
||||
*/
|
||||
|
||||
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
|
||||
import { join } from 'node:path'
|
||||
import { Readable, Writable } from 'node:stream'
|
||||
import {
|
||||
ClientSideConnection,
|
||||
ndJsonStream,
|
||||
type Agent as AcpAgent,
|
||||
type Client,
|
||||
type RequestPermissionRequest,
|
||||
type RequestPermissionResponse,
|
||||
type SessionNotification,
|
||||
} from '@agentclientprotocol/sdk'
|
||||
import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
|
||||
|
||||
/** The source/built agent entry, leaf config, and workspace tsconfig an ACP test boots. */
|
||||
export interface AgentUnderTest {
|
||||
/** The agent source bin entry (for example `packages/examples/acp-demo/src/bin.ts`). */
|
||||
binScript: string
|
||||
/** Explicit built-mode entry for fixtures whose source path is not under `src/`. */
|
||||
libBinScript?: string | undefined
|
||||
/** The leaf `cordis.yml` loaded by the bin. */
|
||||
configPath: string
|
||||
/** The repo tsconfig whose paths resolve unbuilt workspace imports. */
|
||||
tsconfigPath: string
|
||||
}
|
||||
|
||||
/** Options for one ACP test subprocess. */
|
||||
export interface AcpTestLaunchOptions {
|
||||
/** The agent composition to boot. */
|
||||
agent: AgentUnderTest
|
||||
/** Process cwd and default session-home root. */
|
||||
cwd: string
|
||||
/** Alternate leaf config for this launch. */
|
||||
configPath?: string
|
||||
/** Extra environment values layered over the parent environment. */
|
||||
env?: NodeJS.ProcessEnv
|
||||
/** Permission handler; omitted requests fail closed as `cancelled`. */
|
||||
requestPermission?: (params: RequestPermissionRequest) => Promise<RequestPermissionResponse>
|
||||
}
|
||||
|
||||
/** A running ACP test process and its captured client-side surfaces. */
|
||||
export interface LaunchedAcpTestAgent {
|
||||
/** The child process, exposed for process-level assertions. */
|
||||
child: ChildProcessWithoutNullStreams
|
||||
/** Resolve when the OS spawns the child; reject with its asynchronous spawn failure. */
|
||||
spawned: Promise<void>
|
||||
/** The SDK connection backed by the child's stdio. */
|
||||
client: ClientSideConnection
|
||||
/** Session updates in receive order. */
|
||||
updates: SessionNotification['update'][]
|
||||
/** Decode all stdout bytes captured so far. */
|
||||
rawStdout(): string
|
||||
/** Decode all stderr chunks captured so far. */
|
||||
stderr(): string
|
||||
/** Resolve when a future session update matches the predicate. */
|
||||
waitForUpdate(match: (update: SessionNotification['update']) => boolean): Promise<SessionNotification['update']>
|
||||
/** Close the process and drain its streams and callbacks; rejects promptly if fallback termination is refused. */
|
||||
close(signal?: NodeJS.Signals): Promise<void>
|
||||
}
|
||||
|
||||
/**
|
||||
* Boot an ACP agent subprocess and connect an SDK client to its stdio.
|
||||
*
|
||||
* @param options Agent paths, cwd, environment, and optional permission handler.
|
||||
* @returns The running process, connected client, captures, and shutdown handle.
|
||||
*/
|
||||
export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTestAgent {
|
||||
const { agent, cwd } = options
|
||||
const launch = resolveExampleLaunch({
|
||||
srcBin: agent.binScript,
|
||||
libBin: agent.libBinScript,
|
||||
configArgs: ['--config', options.configPath ?? agent.configPath],
|
||||
tsconfigPath: agent.tsconfigPath,
|
||||
env: {
|
||||
...options.env,
|
||||
DSH_HOME: join(cwd, '.dsh'),
|
||||
DSH_AGENTS_HOME: join(cwd, '.agents'),
|
||||
},
|
||||
})
|
||||
const child = spawn(
|
||||
launch.command,
|
||||
launch.args,
|
||||
{
|
||||
cwd,
|
||||
env: { ...process.env, ...launch.env },
|
||||
stdio: ['pipe', 'pipe', 'pipe'],
|
||||
},
|
||||
)
|
||||
// A spawn-level failure is an asynchronous `error` event. Observe it in the
|
||||
// same tick as spawn so a missing cwd or OS rejection cannot crash the test
|
||||
// runner, then make startup and shutdown surface the original error.
|
||||
// Keep observing after the first error: a fallback kill attempted during
|
||||
// shutdown may itself report another process error, which must not become an
|
||||
// unhandled EventEmitter error after the promise has already settled.
|
||||
const childFailure = new Promise<Error>(resolve => child.on('error', resolve))
|
||||
const spawned = Promise.race([
|
||||
new Promise<void>(resolve => child.once('spawn', resolve)),
|
||||
childFailure.then((error): never => { throw error }),
|
||||
])
|
||||
// `spawned` is public and close() also awaits it, but a caller may ignore both.
|
||||
// Keep that misuse from turning the already-observed child error into an
|
||||
// unhandled promise rejection.
|
||||
void spawned.catch(() => undefined)
|
||||
|
||||
const stderrChunks: string[] = []
|
||||
child.stderr.setEncoding('utf8')
|
||||
child.stderr.on('data', (chunk: string) => stderrChunks.push(chunk))
|
||||
|
||||
const rawBuffers: Buffer[] = []
|
||||
const passthrough = new Readable({ read() {} })
|
||||
const updates: SessionNotification['update'][] = []
|
||||
const updateWaiters: {
|
||||
match: (update: SessionNotification['update']) => boolean
|
||||
resolve: (update: SessionNotification['update']) => void
|
||||
reject: (reason: unknown) => void
|
||||
}[] = []
|
||||
let updateStreamFailure: Error | undefined
|
||||
const closeUpdateStream = (): void => {
|
||||
if (updateStreamFailure !== undefined) return
|
||||
updateStreamFailure = new Error('ACP test agent update stream closed before a matching session update arrived')
|
||||
for (const waiter of updateWaiters.splice(0)) waiter.reject(updateStreamFailure)
|
||||
}
|
||||
child.stdout.on('data', (buffer: Buffer) => {
|
||||
rawBuffers.push(buffer)
|
||||
passthrough.push(buffer)
|
||||
})
|
||||
child.stdout.on('end', () => {
|
||||
passthrough.push(null)
|
||||
})
|
||||
const stream = ndJsonStream(
|
||||
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
|
||||
Readable.toWeb(passthrough) as ReadableStream<Uint8Array>,
|
||||
)
|
||||
const inFlightClientCallbacks = new Set<Promise<unknown>>()
|
||||
const trackClientCallback = <T>(callback: () => T | PromiseLike<T>): Promise<T> => {
|
||||
const pending = Promise.resolve().then(callback)
|
||||
inFlightClientCallbacks.add(pending)
|
||||
const untrack = (): void => { inFlightClientCallbacks.delete(pending) }
|
||||
void pending.then(untrack, untrack)
|
||||
return pending
|
||||
}
|
||||
const requestPermission = options.requestPermission
|
||||
?? (() => Promise.resolve({ outcome: { outcome: 'cancelled' as const } }))
|
||||
const makeClient = (_agent: AcpAgent): Client => ({
|
||||
sessionUpdate(params: SessionNotification): Promise<void> {
|
||||
return trackClientCallback(() => {
|
||||
updates.push(params.update)
|
||||
for (let index = updateWaiters.length - 1; index >= 0; index--) {
|
||||
const waiter = updateWaiters[index]
|
||||
/* v8 ignore next 1 -- index is bounded by the array length */
|
||||
if (waiter === undefined) continue
|
||||
let matches: boolean
|
||||
try {
|
||||
matches = waiter.match(params.update)
|
||||
} catch (error: unknown) {
|
||||
updateWaiters.splice(index, 1)
|
||||
waiter.reject(error)
|
||||
continue
|
||||
}
|
||||
if (!matches) continue
|
||||
updateWaiters.splice(index, 1)
|
||||
waiter.resolve(params.update)
|
||||
}
|
||||
})
|
||||
},
|
||||
requestPermission: params => trackClientCallback(() => requestPermission(params)),
|
||||
})
|
||||
const client = new ClientSideConnection(makeClient, stream)
|
||||
// `exit` only reports the parent process's status. Descendants may retain
|
||||
// inherited stdout/stderr handles and buffered ACP frames may still be
|
||||
// crossing the SDK parser. Node's `close` follows stdio closure; the SDK's
|
||||
// `closed` follows parser exhaustion. Capture both eagerly so a caller that
|
||||
// invokes close after process exit still joins the complete drain boundary.
|
||||
const stdioClosed = new Promise<void>(resolve => child.once('close', () => { resolve() }))
|
||||
const drained = Promise.all([stdioClosed, client.closed]).then(async () => {
|
||||
// The ACP SDK's readable loop dispatches client callbacks without awaiting
|
||||
// them. Once `closed` settles no new callbacks can start, but callbacks
|
||||
// already in flight still belong to this launch's teardown boundary.
|
||||
while (inFlightClientCallbacks.size > 0) {
|
||||
await Promise.allSettled([...inFlightClientCallbacks])
|
||||
}
|
||||
})
|
||||
// A caller may await a pending update without calling close(). Make natural
|
||||
// stream exhaustion terminal for those waiters too, but only after the
|
||||
// parser has dispatched every buffered frame.
|
||||
void client.closed.then(closeUpdateStream)
|
||||
|
||||
return {
|
||||
child,
|
||||
spawned,
|
||||
client,
|
||||
updates,
|
||||
rawStdout: () => Buffer.concat(rawBuffers).toString('utf8'),
|
||||
stderr: () => stderrChunks.join(''),
|
||||
waitForUpdate(match): Promise<SessionNotification['update']> {
|
||||
if (updateStreamFailure !== undefined) return Promise.reject(updateStreamFailure)
|
||||
return new Promise((resolve, reject) => updateWaiters.push({ match, resolve, reject }))
|
||||
},
|
||||
async close(signal?: NodeJS.Signals): Promise<void> {
|
||||
try {
|
||||
await spawned
|
||||
} catch (error: unknown) {
|
||||
await drained
|
||||
closeUpdateStream()
|
||||
throw error
|
||||
}
|
||||
if (!isRunning(child)) {
|
||||
await drained
|
||||
closeUpdateStream()
|
||||
return
|
||||
}
|
||||
const exited = waitForExit(child)
|
||||
if (signal === undefined) child.stdin.end()
|
||||
else child.kill(signal)
|
||||
const failure = await Promise.race([
|
||||
exited.then((): undefined => undefined),
|
||||
childFailure,
|
||||
])
|
||||
if (failure === undefined) {
|
||||
await drained
|
||||
closeUpdateStream()
|
||||
return
|
||||
}
|
||||
|
||||
// An `error` after spawn is not an exit edge: in particular, a failed
|
||||
// signal can leave the subprocess live. Force termination, await the
|
||||
// already-observed exit edge, and only then propagate the child error so
|
||||
// callers may safely remove cwd/session resources after close rejects.
|
||||
const fallbackError = Promise.withResolvers<Error>()
|
||||
const observeFallbackError = (error: Error): void => { fallbackError.resolve(error) }
|
||||
child.once('error', observeFallbackError)
|
||||
if (!child.kill('SIGKILL')) {
|
||||
child.off('error', observeFallbackError)
|
||||
closeUpdateStream()
|
||||
throw new AggregateError(
|
||||
[failure, new Error('Fallback SIGKILL was not accepted by the child process')],
|
||||
'ACP test agent failed and fallback termination was refused',
|
||||
)
|
||||
}
|
||||
const fallbackFailure = await Promise.race([
|
||||
exited.then((): undefined => undefined),
|
||||
fallbackError.promise,
|
||||
])
|
||||
child.off('error', observeFallbackError)
|
||||
if (fallbackFailure !== undefined) {
|
||||
closeUpdateStream()
|
||||
throw new AggregateError(
|
||||
[failure, fallbackFailure],
|
||||
'ACP test agent failed and fallback termination was refused',
|
||||
)
|
||||
}
|
||||
await drained
|
||||
closeUpdateStream()
|
||||
throw failure
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/** Resolve once a running child exits. */
|
||||
function waitForExit(child: ChildProcessWithoutNullStreams): Promise<void> {
|
||||
return new Promise<void>(resolve => child.once('exit', () => { resolve() }))
|
||||
}
|
||||
|
||||
/** Whether the child still lacks either OS termination marker. */
|
||||
function isRunning(child: ChildProcessWithoutNullStreams): boolean {
|
||||
return child.exitCode === null && child.signalCode === null
|
||||
}
|
||||
@@ -14,6 +14,16 @@ const MESSAGE_PREFIX = '{{messagePrefix}}'
|
||||
|
||||
/** A UUID v4 string, the shape `randomUUID()` produces for session ids. */
|
||||
const UUID_RE = /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/gi
|
||||
const LOCAL_SPILL_PATH_RE = new RegExp(
|
||||
String.raw`\{\{cwd\}\}/\.spill/session-[0-9a-f]{12}/[0-9a-f]{12}-([A-Za-z0-9._~-]+?)`
|
||||
+ String.raw`(?=\. Use read with offset/limit|[\s)]|$)`,
|
||||
'g',
|
||||
)
|
||||
const SNAPSHOT_SPILL_PATH_RE = new RegExp(
|
||||
String.raw`/tmp/dsh-acp-snapshot-spill/session-[0-9a-f]{12}/[0-9a-f]{12}-([A-Za-z0-9._~-]+?)`
|
||||
+ String.raw`(?=\. Use read with offset/limit|[\s)]|$)`,
|
||||
'g',
|
||||
)
|
||||
|
||||
/** Inputs the normalizers need to recognize a run's volatile values. */
|
||||
export interface NormalizeContext {
|
||||
@@ -29,6 +39,9 @@ function scrubString(value: string, ctx: NormalizeContext): string {
|
||||
// cwd first (longest, most specific), then explicit session ids, then any
|
||||
// residual UUID (covers ids that appear in places we didn't enumerate).
|
||||
out = out.split(ctx.cwd).join(CWD)
|
||||
out = out.split(`/private${CWD}`).join(CWD)
|
||||
out = out.replace(LOCAL_SPILL_PATH_RE, (_match, name: string) => `{{spillLocator:${name}}}`)
|
||||
out = out.replace(SNAPSHOT_SPILL_PATH_RE, (_match, name: string) => `{{spillLocator:${name}}}`)
|
||||
for (const id of ctx.sessionIds) out = out.split(id).join(SESSION_ID)
|
||||
out = out.replace(UUID_RE, SESSION_ID)
|
||||
return out
|
||||
@@ -47,7 +60,7 @@ function scrubValue(value: unknown, ctx: NormalizeContext): unknown {
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalize a raw stdout transcript (newline-delimited JSON-RPC frames) into a stable golden
|
||||
* Normalize a raw stdout transcript (newline-delimited JSON-RPC frames) into a stable expected output
|
||||
* in the same shape as the wire: one compact JSON frame per line (NDJSON), with the JSON-RPC
|
||||
* `id` rewritten to a per-transcript sequence (1, 2, 3, …) and all volatile strings scrubbed.
|
||||
* Invalid JSON throws, doubling as a protocol-stdout purity check.
|
||||
@@ -59,7 +72,7 @@ function scrubValue(value: unknown, ctx: NormalizeContext): unknown {
|
||||
export function normalizeStdout(rawStdout: string, ctx: NormalizeContext): string {
|
||||
const lines = rawStdout.split('\n').filter(line => line.trim().length > 0)
|
||||
// Map each distinct JSON-RPC id (request/response correlate by id) to a stable
|
||||
// sequence number, in first-seen order, so id churn doesn't perturb the golden.
|
||||
// sequence number, in first-seen order, so id churn doesn't perturb the expected output.
|
||||
const idSeq = new Map<string, number>()
|
||||
const stableId = (id: unknown): number => {
|
||||
const key = JSON.stringify(id)
|
||||
@@ -78,7 +91,7 @@ export function normalizeStdout(rawStdout: string, ctx: NormalizeContext): strin
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalize a session JSONL log into a stable golden: the header line's
|
||||
* Normalize a session JSONL log into a stable expected output: the header line's
|
||||
* volatile fields (`createdAt`, `id`, `cwd`) and every event's `time` are
|
||||
* zeroed/scrubbed, all volatile strings scrubbed, and `seq` is LEFT INTACT
|
||||
* (deterministic by contract). A packed chunk row's timing (`time0`, the `dt`
|
||||
@@ -108,7 +121,7 @@ export function normalizeSessionLog(rawLog: string, ctx: NormalizeContext): stri
|
||||
// Event line: zero the epoch-ms timestamp; keep seq (deterministic).
|
||||
record.time = 0
|
||||
// A hook/result carries the hook's wall-clock runtime (`data.durationMs`),
|
||||
// which is run-to-run noise like `time` — zero it so the golden reflects
|
||||
// which is run-to-run noise like `time` — zero it so the expected output reflects
|
||||
// the hook's decision/exit, not how long the shell took.
|
||||
if (record.type === 'hook/result' && record.data !== null && typeof record.data === 'object') {
|
||||
const data = record.data as Record<string, unknown>
|
||||
@@ -121,8 +134,8 @@ export function normalizeSessionLog(rawLog: string, ctx: NormalizeContext): stri
|
||||
}
|
||||
|
||||
/**
|
||||
* Replace system-prompt content in request headers and header deltas with
|
||||
* `{{system}}` tokens while retaining field presence and delta structure.
|
||||
* Replace system-prompt content in request headers with `{{system}}` tokens
|
||||
* while retaining field presence.
|
||||
* Other header content stays verbatim, so a header-pinning fixture can keep
|
||||
* its complete tool schemas while every JSONL fixture omits the prompt text.
|
||||
* Lines without a system payload pass through byte-for-byte; the transform is
|
||||
@@ -136,11 +149,11 @@ export function scrubSystemPrompts(rawLog: string): string {
|
||||
}
|
||||
|
||||
/**
|
||||
* Replace tool schemas in request headers and header deltas with `{{tools}}`
|
||||
* tokens while retaining field presence, tool names, and delta structure.
|
||||
* System prompts and session-prefix messages stay verbatim so pinning fixtures
|
||||
* can move only schema bulk into their dedicated JSON sidecar. Lines without a
|
||||
* tool payload pass through byte-for-byte; the transform is idempotent.
|
||||
* Replace tool schemas in full request-header snapshots with `{{tools}}`
|
||||
* tokens while retaining field presence. System prompts and session-prefix
|
||||
* messages stay verbatim so pinning fixtures can move only schema bulk into
|
||||
* their dedicated JSON sidecar. Lines without a tool payload pass through
|
||||
* byte-for-byte; the transform is idempotent.
|
||||
*
|
||||
* @param rawLog The raw session `.jsonl` content.
|
||||
* @returns The JSONL with tool-schema content tokenized.
|
||||
@@ -153,9 +166,9 @@ export function scrubToolSchemas(rawLog: string): string {
|
||||
* Replace all bulky request-header content in a session JSONL with stable
|
||||
* tokens. This includes the system-prompt fields handled by
|
||||
* {@link scrubSystemPrompts}, tool schemas, and session-prefix messages. It
|
||||
* keeps system-delta line positions and arity, tool-delta names, prefix
|
||||
* message counts, field presence, config, and reason. Lines without content
|
||||
* to scrub pass through byte-for-byte, and the transform is idempotent.
|
||||
* keeps prefix message counts, field presence, config, and reason. Lines
|
||||
* without content to scrub pass through byte-for-byte, and the transform is
|
||||
* idempotent.
|
||||
*
|
||||
* @param rawLog The raw session `.jsonl` content.
|
||||
* @returns The JSONL with all header bulk tokenized, other lines byte-identical.
|
||||
@@ -191,33 +204,7 @@ function scrubHeaderContent(rawLog: string, options: HeaderScrubOptions): string
|
||||
}
|
||||
return touched ? JSON.stringify(record) : line
|
||||
}
|
||||
if (record.type === 'request/header-delta') {
|
||||
let touched = false
|
||||
const system = data.system as Record<string, unknown> | null | undefined
|
||||
if (options.system === true && system !== null && typeof system === 'object' && Array.isArray(system.insert)) {
|
||||
system.insert = system.insert.map(() => SYSTEM)
|
||||
touched = true
|
||||
}
|
||||
const tools = data.tools as Record<string, unknown> | null | undefined
|
||||
if (options.tools === true && tools !== null && typeof tools === 'object') {
|
||||
if (Array.isArray(tools.added)) { tools.added = tools.added.map(scrubToolSchema); touched = true }
|
||||
if (Array.isArray(tools.changed)) { tools.changed = tools.changed.map(scrubToolSchema); touched = true }
|
||||
}
|
||||
if (options.prefix === true && Array.isArray(data.messagePrefix)) {
|
||||
data.messagePrefix = data.messagePrefix.map(() => MESSAGE_PREFIX)
|
||||
touched = true
|
||||
}
|
||||
return touched ? JSON.stringify(record) : line
|
||||
}
|
||||
return line
|
||||
})
|
||||
return out.join('\n')
|
||||
}
|
||||
|
||||
/** Tokenize one tool schema's bulk (description, parameters, anything else), keeping its identifying `name`. */
|
||||
function scrubToolSchema(tool: unknown): unknown {
|
||||
if (tool === null || typeof tool !== 'object' || Array.isArray(tool)) return tool
|
||||
const out: Record<string, unknown> = {}
|
||||
for (const [k, v] of Object.entries(tool)) out[k] = k === 'name' ? v : TOOLS
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -1,16 +1,21 @@
|
||||
/**
|
||||
* Keyless-by-default ACP snapshot suite factory. Each scenario drives the real subprocess and
|
||||
* compares normalized stdout; comparable session fixtures are both replay input and expected
|
||||
* output. Record mode refreshes reproducible model scenarios from the live API, while refresh
|
||||
* mode replays committed scripts and rewrites derived artifacts without a key.
|
||||
* Keyless-by-default ACP snapshot suite factory. Each scenario drives the real
|
||||
* subprocess and compares normalized stdout; comparable session fixtures are
|
||||
* both replay input and expected output. Record mode refreshes reproducible
|
||||
* model scenarios from the live API, while refresh mode replays committed
|
||||
* scripts and rewrites derived artifacts without a key.
|
||||
* Replay scenarios run concurrently because each subprocess owns unique temp
|
||||
* cwd and persistence roots and reads only committed fixtures. Record and
|
||||
* refresh stay serial while writing.
|
||||
*
|
||||
* Exactly one scenario per header-composition class pins the system prompt and tool schemas in
|
||||
* dedicated sidecars. Every live header is checked against that pin, so session-dependent
|
||||
* composition must declare a separate class instead of escaping coverage.
|
||||
* Exactly one scenario per header-composition class pins the full prompt and
|
||||
* tool-schema sequences in dedicated sidecars. Every live header is checked
|
||||
* against that pin, so session-dependent composition must declare a separate
|
||||
* class instead of escaping coverage.
|
||||
* @module @deepseek-ai/dsh-acp-snapshot/suite
|
||||
*/
|
||||
|
||||
import { readFile, readdir, writeFile } from 'node:fs/promises'
|
||||
import { readFile, readdir, rm, writeFile } from 'node:fs/promises'
|
||||
import { existsSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
@@ -25,10 +30,10 @@ import {
|
||||
} from './normalize.ts'
|
||||
|
||||
/** The readable system-prompt snapshot beside each header-pinning fixture. */
|
||||
const SYSTEM_PROMPT_SNAPSHOT = 'system-prompt.golden.md'
|
||||
const SYSTEM_PROMPT_SNAPSHOT = 'system-prompt.expected.md'
|
||||
|
||||
/** The structured tool-schema snapshot beside each header-pinning fixture. */
|
||||
const TOOL_SCHEMAS_SNAPSHOT = 'tool-schemas.golden.json'
|
||||
const TOOL_SCHEMAS_SNAPSHOT = 'tool-schemas.expected.json'
|
||||
|
||||
/** Stable session-log token standing in for the sidecar's initial schemas. */
|
||||
const TOOLS_TOKEN = '{{tools}}'
|
||||
@@ -36,7 +41,7 @@ const TOOLS_TOKEN = '{{tools}}'
|
||||
/** A snapshot scenario and how its fixtures are produced. */
|
||||
export interface Scenario {
|
||||
name: string
|
||||
/** Whether the scenario drives at least one model turn (so a JSONL golden applies). */
|
||||
/** Whether the scenario drives at least one model turn (so a JSONL expected output applies). */
|
||||
hasModelTurn: boolean
|
||||
/**
|
||||
* Whether the run persists a comparable session log to diff against the
|
||||
@@ -66,28 +71,17 @@ export interface Scenario {
|
||||
* false (replay derives from the fixture's `assistant/chunk` events).
|
||||
*/
|
||||
overridden?: boolean
|
||||
/**
|
||||
* How many SUBAGENT child sessions this scenario records beyond the top-level
|
||||
* one (0 for a single-session scenario). Each child rides in a sibling fixture
|
||||
* `session.<n>.jsonl` (1-based); replay forwards them to `dsh-llm-replay` so
|
||||
* each child session replays from its own script, and record mode writes the
|
||||
* harvested child logs back to those files. Defaults to 0.
|
||||
*/
|
||||
childSessions?: number
|
||||
/**
|
||||
* Whether this scenario is its header class's sole request-header pin. Dedicated sidecars own
|
||||
* the prompt and tool schemas, while every classmate is checked for equality.
|
||||
*/
|
||||
pinsHeader?: boolean
|
||||
/**
|
||||
* How many `request/header-delta` events this PINNING scenario's fixture
|
||||
* legitimately carries (default 0). A recorded mid-run header change — a
|
||||
* config-option switch rewriting a prompt section — is part of the pinned
|
||||
* surface, with readable prompt text in Markdown; any OTHER count
|
||||
* still fails, so fixture rot stays caught. Meaningless off the pin (the
|
||||
* live uniformity guard keeps non-pinning scenarios delta-free).
|
||||
* How many changed `request/header` snapshots this PINNING scenario's primary
|
||||
* fixture legitimately carries (default 0). Their full prompt text is kept in
|
||||
* the readable Markdown pin; any other count fails. Meaningless off the pin.
|
||||
*/
|
||||
expectedHeaderDeltas?: number
|
||||
expectedHeaderChanges?: number
|
||||
/**
|
||||
* Which header-composition class this scenario belongs to. Scenarios that
|
||||
* boot the same config compose the same header; each class has exactly one
|
||||
@@ -118,8 +112,8 @@ export interface SnapshotSuiteOptions {
|
||||
scenarios: Scenario[]
|
||||
/**
|
||||
* `replay` (keyless, the default tier), `record` (live API; re-records the
|
||||
* `recorded` scenarios' fixtures and refreshes the Vitest goldens under
|
||||
* `--update`), or `refresh` (keyless replay that rewrites stdout goldens and
|
||||
* `recorded` scenarios' fixtures and refreshes the Vitest expected outputs under
|
||||
* `--update`), or `refresh` (keyless replay that rewrites stdout expected outputs and
|
||||
* comparable session fixtures from the replay run). The caller derives this
|
||||
* from `$DSH_SNAPSHOT` — env reading stays outside this library.
|
||||
*/
|
||||
@@ -127,14 +121,40 @@ export interface SnapshotSuiteOptions {
|
||||
}
|
||||
|
||||
/**
|
||||
* The sibling child-fixture paths for a scenario (`session.1.jsonl` …).
|
||||
* Validate and order a scenario directory's session-fixture filenames.
|
||||
*
|
||||
* @param dir The scenario's snapshots directory (`<snapshotsDir>/<name>`).
|
||||
* @param childSessions How many subagent child sessions the scenario records.
|
||||
* @returns One path per child, 1-based, in fixture order.
|
||||
* The primary fixture is always `session.jsonl`; child sessions are discovered
|
||||
* from contiguous `session.1.jsonl` … filenames. The directory is the source of
|
||||
* truth, so scenario tables do not duplicate a child count that can drift from
|
||||
* the files. A session-like JSONL with any other suffix fails loud.
|
||||
*
|
||||
* @param names File names in one scenario directory.
|
||||
* @returns The primary and child fixture names in replay/harvest order.
|
||||
*/
|
||||
export function childFixturePaths(dir: string, childSessions: number): string[] {
|
||||
return Array.from({ length: childSessions }, (_, i) => join(dir, `session.${i + 1}.jsonl`))
|
||||
export function sessionFixtureNames(names: readonly string[]): string[] {
|
||||
if (!names.includes('session.jsonl')) throw new Error('missing session.jsonl')
|
||||
const children: { name: string; index: number }[] = []
|
||||
for (const name of names) {
|
||||
if (name === 'session.jsonl') continue
|
||||
if (!name.startsWith('session.') || !name.endsWith('.jsonl')) continue
|
||||
const match = /^session\.([1-9]\d*)\.jsonl$/.exec(name)
|
||||
if (match === null) throw new Error(`invalid child session fixture name: ${name}`)
|
||||
children.push({ name, index: Number(match[1]) })
|
||||
}
|
||||
children.sort((a, b) => a.index - b.index)
|
||||
for (const [offset, child] of children.entries()) {
|
||||
const expected = offset + 1
|
||||
if (child.index !== expected) {
|
||||
throw new Error(`child session fixtures must be contiguous: expected session.${expected}.jsonl, found ${child.name}`)
|
||||
}
|
||||
}
|
||||
return ['session.jsonl', ...children.map(child => child.name)]
|
||||
}
|
||||
|
||||
/** Read one scenario directory's validated session-fixture inventory. */
|
||||
async function sessionFixtures(dir: string): Promise<string[]> {
|
||||
const entries = await readdir(dir, { withFileTypes: true })
|
||||
return sessionFixtureNames(entries.filter(entry => entry.isFile()).map(entry => entry.name))
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -208,114 +228,58 @@ export function normalizedToolSchemas(rawLog: string, ctx: NormalizeContext): un
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract normalized tool-schema edits from request-header deltas in log order.
|
||||
* Deltas without an object-valued tools edit are omitted; their remaining
|
||||
* structure stays pinned in the session JSONL.
|
||||
*
|
||||
* @param rawLog The session `.jsonl` content to inspect.
|
||||
* @param ctx The volatile values of the run that produced it.
|
||||
* @returns The normalized tool-schema edits, in event order.
|
||||
*/
|
||||
export function normalizedToolSchemaDeltas(rawLog: string, ctx: NormalizeContext): unknown[] {
|
||||
return normalizeSessionLog(rawLog, ctx)
|
||||
.split('\n')
|
||||
.filter(line => line.trim().length > 0)
|
||||
.map(line => JSON.parse(line) as { type?: unknown; data?: { tools?: unknown } })
|
||||
.filter(record => record.type === 'request/header-delta')
|
||||
.flatMap((record) => {
|
||||
const tools = record.data?.tools
|
||||
return tools !== null && typeof tools === 'object' && !Array.isArray(tools) ? [tools] : []
|
||||
})
|
||||
}
|
||||
|
||||
/** The structured contents of a tool-schema sidecar. */
|
||||
export interface ToolSchemasSnapshot {
|
||||
/** The complete tool schemas from the pinned request header. */
|
||||
initial: unknown[]
|
||||
/** Complete tool-schema edits from subsequent request-header deltas. */
|
||||
deltas: unknown[]
|
||||
/** Complete tool schemas from subsequent changed-header snapshots. */
|
||||
changes: unknown[][]
|
||||
}
|
||||
|
||||
/**
|
||||
* Render tool schemas and later schema edits as canonical, readable JSON.
|
||||
* Render the full tool-schema sequence as canonical, readable JSON.
|
||||
*
|
||||
* @param initial The pinned request header's complete tool schemas.
|
||||
* @param deltas Complete tool-schema edits from request-header deltas.
|
||||
* @param changes Complete tool schemas from later changed headers.
|
||||
* @returns A pretty-printed JSON snapshot ending in one newline.
|
||||
*/
|
||||
export function formatToolSchemasSnapshot(initial: readonly unknown[], deltas: readonly unknown[] = []): string {
|
||||
return `${JSON.stringify({ initial, deltas }, null, 2)}\n`
|
||||
export function formatToolSchemasSnapshot(initial: readonly unknown[], changes: readonly unknown[][] = []): string {
|
||||
return `${JSON.stringify({ initial, changes }, null, 2)}\n`
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse and validate the stable top-level shape of a tool-schema sidecar.
|
||||
*
|
||||
* @param snapshot The JSON sidecar text.
|
||||
* @returns Its initial schemas and schema deltas.
|
||||
* @returns Its initial and changed-header schema sets.
|
||||
*/
|
||||
export function parseToolSchemasSnapshot(snapshot: string): ToolSchemasSnapshot {
|
||||
const parsed = JSON.parse(snapshot) as unknown
|
||||
if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) {
|
||||
throw new Error('acp-snapshot: tool-schema snapshot must be an object')
|
||||
}
|
||||
const { initial, deltas } = parsed as { initial?: unknown; deltas?: unknown }
|
||||
if (!Array.isArray(initial) || !Array.isArray(deltas)) {
|
||||
throw new Error('acp-snapshot: tool-schema snapshot must carry array-valued initial and deltas fields')
|
||||
const { initial, changes } = parsed as { initial?: unknown; changes?: unknown }
|
||||
if (!Array.isArray(initial) || !Array.isArray(changes) || !changes.every(Array.isArray)) {
|
||||
throw new Error('acp-snapshot: tool-schema snapshot must carry array-valued initial and changes fields')
|
||||
}
|
||||
return { initial, deltas }
|
||||
return { initial, changes }
|
||||
}
|
||||
|
||||
/**
|
||||
* Restore a sidecar's initial schemas into a tokenized pinned header.
|
||||
* Restore one sidecar schema set into a tokenized pinned header.
|
||||
*
|
||||
* @param header The parsed request header carrying `tools: "{{tools}}"`.
|
||||
* @param snapshot The parsed tool-schema sidecar.
|
||||
* @returns A copy of the header with its complete initial schemas restored.
|
||||
* @param schemas The complete schemas for this full header snapshot.
|
||||
* @returns A copy of the header with its complete schemas restored.
|
||||
*/
|
||||
export function restorePinnedToolSchemas(header: unknown, snapshot: ToolSchemasSnapshot): unknown {
|
||||
export function restorePinnedToolSchemas(header: unknown, schemas: readonly unknown[]): unknown {
|
||||
if (header === null || typeof header !== 'object' || Array.isArray(header)) {
|
||||
throw new Error('acp-snapshot: pinned request header must be an object')
|
||||
}
|
||||
if ((header as { tools?: unknown }).tools !== TOOLS_TOKEN) {
|
||||
throw new Error(`acp-snapshot: pinned request header tools must equal ${TOOLS_TOKEN}`)
|
||||
}
|
||||
return { ...header, tools: snapshot.initial }
|
||||
}
|
||||
|
||||
/** One normalized system-prompt edit carried by a `request/header-delta`. */
|
||||
export interface SystemPromptDeltaSnapshot {
|
||||
/** How many leading lines remain from the prior prompt. */
|
||||
keepStart: number
|
||||
/** How many trailing lines remain from the prior prompt. */
|
||||
keepEnd: number
|
||||
/** The normalized replacement lines inserted between the retained ranges. */
|
||||
insert: string[]
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract normalized system-prompt edits from request-header deltas in log
|
||||
* order. Deltas without a well-formed system edit are omitted; their non-prompt
|
||||
* structure remains pinned in JSONL.
|
||||
*
|
||||
* @param rawLog The session `.jsonl` content to inspect.
|
||||
* @param ctx The volatile values of the run that produced it.
|
||||
* @returns The normalized system-prompt edits, in event order.
|
||||
*/
|
||||
export function normalizedSystemPromptDeltas(rawLog: string, ctx: NormalizeContext): SystemPromptDeltaSnapshot[] {
|
||||
return normalizeSessionLog(rawLog, ctx)
|
||||
.split('\n')
|
||||
.filter(line => line.trim().length > 0)
|
||||
.map(line => JSON.parse(line) as { type?: unknown; data?: { system?: unknown } })
|
||||
.filter(record => record.type === 'request/header-delta')
|
||||
.flatMap((record) => {
|
||||
const system = record.data?.system
|
||||
if (system === null || typeof system !== 'object') return []
|
||||
const { keepStart, keepEnd, insert } = system as { keepStart?: unknown; keepEnd?: unknown; insert?: unknown }
|
||||
if (typeof keepStart !== 'number' || typeof keepEnd !== 'number' || !Array.isArray(insert)) return []
|
||||
if (!insert.every(line => typeof line === 'string')) return []
|
||||
return [{ keepStart, keepEnd, insert: insert }]
|
||||
})
|
||||
return { ...header, tools: schemas }
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -324,38 +288,40 @@ export function normalizedSystemPromptDeltas(rawLog: string, ctx: NormalizeConte
|
||||
* the committed file follows the repository newline contract.
|
||||
*
|
||||
* @param prompt The normalized system prompt.
|
||||
* @param deltas Normalized prompt edits to append as readable sections.
|
||||
* @param changes Full normalized prompts from later changed-header snapshots.
|
||||
* @returns Markdown snapshot text ending in a newline.
|
||||
*/
|
||||
export function formatSystemPromptSnapshot(
|
||||
prompt: string,
|
||||
deltas: readonly SystemPromptDeltaSnapshot[] = [],
|
||||
changes: readonly string[] = [],
|
||||
): string {
|
||||
let snapshot = prompt.endsWith('\n') ? prompt : `${prompt}\n`
|
||||
for (const [index, delta] of deltas.entries()) {
|
||||
snapshot += `\n<!-- request/header-delta ${index + 1}: keepStart=${delta.keepStart}, keepEnd=${delta.keepEnd} -->\n\n`
|
||||
const insert = delta.insert.join('\n')
|
||||
snapshot += insert.endsWith('\n') ? insert : `${insert}\n`
|
||||
for (const [index, change] of changes.entries()) {
|
||||
snapshot += `\n<!-- request/header change ${index + 1} -->\n\n`
|
||||
snapshot += change.endsWith('\n') ? change : `${change}\n`
|
||||
}
|
||||
return snapshot
|
||||
}
|
||||
|
||||
/** Return the initial-prompt portion of a possibly delta-bearing snapshot. */
|
||||
/** Return the initial-prompt portion of a possibly multi-header snapshot. */
|
||||
function initialSystemPromptSnapshot(snapshot: string): string {
|
||||
const marker = snapshot.indexOf('\n<!-- request/header-delta ')
|
||||
const marker = snapshot.indexOf('\n<!-- request/header change ')
|
||||
return marker < 0 ? snapshot : snapshot.slice(0, marker)
|
||||
}
|
||||
|
||||
/**
|
||||
* Count the `request/header-delta` events in a session JSONL.
|
||||
* Count changed `request/header` snapshots in a session JSONL.
|
||||
*
|
||||
* @param rawLog The session `.jsonl` content.
|
||||
* @returns How many `request/header-delta` events the log carries.
|
||||
* @returns How many headers carry reason `change`.
|
||||
*/
|
||||
export function headerDeltaCount(rawLog: string): number {
|
||||
export function headerChangeCount(rawLog: string): number {
|
||||
return rawLog.split('\n')
|
||||
.filter(line => line.trim().length > 0)
|
||||
.filter(line => (JSON.parse(line) as { type?: unknown }).type === 'request/header-delta')
|
||||
.filter((line) => {
|
||||
const record = JSON.parse(line) as { type?: unknown; data?: { reason?: unknown } }
|
||||
return record.type === 'request/header' && record.data?.reason === 'change'
|
||||
})
|
||||
.length
|
||||
}
|
||||
|
||||
@@ -461,7 +427,7 @@ export function stabilizeRefreshLog(fresh: string, existing: string, replacement
|
||||
}
|
||||
|
||||
/**
|
||||
* Register the suite: one `describe` per scenario (the golden/log compares and
|
||||
* Register the suite: one test per scenario (the expected-output and log comparisons and
|
||||
* the header-uniformity guard) plus the fixture guard block (no orphan
|
||||
* scenario dirs, required files present, exactly one pin per header class,
|
||||
* pinning fixtures well-formed, every JSONL prompt-scrubbed, non-pinning
|
||||
@@ -477,6 +443,7 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
const RECORDING = mode === 'record'
|
||||
const REFRESHING = mode === 'refresh'
|
||||
const childMode: 'replay' | 'record' = RECORDING ? 'record' : 'replay'
|
||||
const scenarioSuite = mode === 'replay' ? describe.concurrent : describe
|
||||
|
||||
/** The class a scenario's header composition belongs to (see {@link Scenario.headerClass}). */
|
||||
const classOf = (scenario: Scenario): string => scenario.headerClass ?? 'default'
|
||||
@@ -496,16 +463,21 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
}
|
||||
}
|
||||
|
||||
for (const scenario of scenarios) {
|
||||
describe(`snapshot: ${scenario.name}`, () => {
|
||||
scenarioSuite('snapshot scenarios', () => {
|
||||
for (const scenario of scenarios) {
|
||||
// In RECORD mode, only re-run the `recorded` (live-API) scenarios; the `authored` ones
|
||||
// (sidecar-driven errors/cancel) are never re-recorded.
|
||||
it.skipIf(RECORDING && !scenario.recorded)('matches the goldens', async () => {
|
||||
it.skipIf(RECORDING && !scenario.recorded)(`snapshot: ${scenario.name} matches the expected outputs`, async ({ expect }) => {
|
||||
const dir = join(snapshotsDir, scenario.name)
|
||||
const input = JSON.parse(await readFile(join(dir, 'input.json'), 'utf8')) as InputScript
|
||||
const overrideFile = join(dir, 'replay.override.json')
|
||||
const workspaceDir = join(dir, 'workspace')
|
||||
const childSessions = scenario.childSessions ?? 0
|
||||
// Replay/refresh need the committed inventory up front because those
|
||||
// files drive the model scripts. Record mode creates that inventory
|
||||
// from the harvested live logs, so it must also work for a brand-new
|
||||
// scenario with no session.jsonl yet.
|
||||
let fixtureFiles = RECORDING ? [] : await sessionFixtures(dir)
|
||||
const childFixtureFiles = fixtureFiles.slice(1)
|
||||
const comparesLog = scenario.comparesLog ?? scenario.hasModelTurn
|
||||
const result = await runScenario(input, {
|
||||
agent,
|
||||
@@ -514,7 +486,7 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
...existsSync(overrideFile) ? { overrideFile } : {},
|
||||
// In REPLAY, forward the recorded child fixtures so each subagent session
|
||||
// replays from its own script. In RECORD they are harvested, not read.
|
||||
...!RECORDING && childSessions > 0 ? { childFiles: childFixturePaths(dir, childSessions) } : {},
|
||||
...!RECORDING && childFixtureFiles.length > 0 ? { childFiles: childFixtureFiles.map(file => join(dir, file)) } : {},
|
||||
...existsSync(workspaceDir) ? { workspaceDir } : {},
|
||||
// A scenario booting an overlay tree passes its own live config; the
|
||||
// bin's replay swap derives the sibling `*cordis.snapshot.yml` from it.
|
||||
@@ -542,7 +514,6 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
const scrub = scenario.pinsHeader === true
|
||||
? (log: string): string => scrubToolSchemas(scrubSystemPrompts(log))
|
||||
: scrubRequestHeaders
|
||||
const fixtureFiles = ['session.jsonl', ...Array.from({ length: childSessions }, (_, i) => `session.${i + 1}.jsonl`)]
|
||||
const existingFixtures = REFRESHING
|
||||
? await Promise.all(fixtureFiles.map(file => readFile(join(dir, file), 'utf8')))
|
||||
: []
|
||||
@@ -551,58 +522,66 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
|| (REFRESHING && comparesLog)
|
||||
if (writesSessionFixtures) {
|
||||
expect(result.sessionLogs.length, `${mode} produced no session log to harvest`).toBeGreaterThan(0)
|
||||
expect(result.sessionLogs.length, `expected ${childSessions + 1} session logs (parent + children)`)
|
||||
.toBe(childSessions + 1)
|
||||
if (REFRESHING) {
|
||||
expect(result.sessionLogs.length, `expected ${fixtureFiles.length} session logs (parent + children)`)
|
||||
.toBe(fixtureFiles.length)
|
||||
}
|
||||
const outputFixtureFiles = [
|
||||
'session.jsonl',
|
||||
...Array.from({ length: result.sessionLogs.length - 1 }, (_, i) => `session.${i + 1}.jsonl`),
|
||||
]
|
||||
const primary = (result.sessionLogs[0] as HarvestedLog).content
|
||||
await writeFile(join(dir, 'session.jsonl'), scrub(
|
||||
await writeFile(join(dir, outputFixtureFiles[0] as string), scrub(
|
||||
REFRESHING ? stabilizeRefreshLog(primary, existingFixtures[0] as string, replacements) : primary,
|
||||
))
|
||||
for (let i = 1; i < result.sessionLogs.length; i++) {
|
||||
const child = (result.sessionLogs[i] as HarvestedLog).content
|
||||
await writeFile(join(dir, `session.${i}.jsonl`), scrub(
|
||||
await writeFile(join(dir, outputFixtureFiles[i] as string), scrub(
|
||||
REFRESHING ? stabilizeRefreshLog(child, existingFixtures[i] as string, replacements) : child,
|
||||
))
|
||||
}
|
||||
if (RECORDING) {
|
||||
const outputNames = new Set(outputFixtureFiles)
|
||||
const entries = await readdir(dir, { withFileTypes: true })
|
||||
await Promise.all(entries
|
||||
.filter(entry => entry.isFile()
|
||||
// Only valid numbered children are record-owned stale output.
|
||||
// Malformed session-like names stay for the inventory guard to
|
||||
// reject instead of being silently deleted during mutation.
|
||||
&& /^session\.[1-9]\d*\.jsonl$/.test(entry.name)
|
||||
&& !outputNames.has(entry.name))
|
||||
.map(entry => rm(join(dir, entry.name))))
|
||||
fixtureFiles = outputFixtureFiles
|
||||
}
|
||||
if (scenario.pinsHeader === true) {
|
||||
const prompts = result.sessionLogs.flatMap(log => normalizedSystemPrompts(log.content, ctx))
|
||||
expect(prompts.length, `${mode} produced no system prompt to snapshot`).toBeGreaterThan(0)
|
||||
const initialSnapshot = formatSystemPromptSnapshot(prompts[0] as string)
|
||||
for (const prompt of prompts) {
|
||||
expect(formatSystemPromptSnapshot(prompt), 'the pinning run produced divergent system prompts')
|
||||
.toEqual(initialSnapshot)
|
||||
}
|
||||
const primary = result.sessionLogs[0] as HarvestedLog
|
||||
const snapshot = formatSystemPromptSnapshot(
|
||||
prompts[0] as string,
|
||||
normalizedSystemPromptDeltas(primary.content, ctx),
|
||||
)
|
||||
const prompts = normalizedSystemPrompts(primary.content, ctx)
|
||||
expect(prompts.length, `${mode} produced no system prompt to snapshot`).toBeGreaterThan(0)
|
||||
const snapshot = formatSystemPromptSnapshot(prompts[0] as string, prompts.slice(1))
|
||||
await writeFile(join(dir, SYSTEM_PROMPT_SNAPSHOT), snapshot)
|
||||
|
||||
const schemaSets = result.sessionLogs.flatMap(log => normalizedToolSchemas(log.content, ctx))
|
||||
const schemaSets = normalizedToolSchemas(primary.content, ctx)
|
||||
expect(schemaSets.length, `${mode} produced no tool schemas to snapshot`).toBeGreaterThan(0)
|
||||
const initialSchemaSnapshot = formatToolSchemasSnapshot(schemaSets[0] as unknown[])
|
||||
for (const schemas of schemaSets) {
|
||||
expect(formatToolSchemasSnapshot(schemas), 'the pinning run produced divergent tool schemas')
|
||||
.toEqual(initialSchemaSnapshot)
|
||||
}
|
||||
expect(schemaSets.length, `${mode} produced a tool-schema sequence that differs from its prompt sequence`)
|
||||
.toBe(prompts.length)
|
||||
await writeFile(join(dir, TOOL_SCHEMAS_SNAPSHOT), formatToolSchemasSnapshot(
|
||||
schemaSets[0] as unknown[],
|
||||
normalizedToolSchemaDeltas(primary.content, ctx),
|
||||
schemaSets.slice(1),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
const stdout = normalizeStdout(result.rawStdout, ctx)
|
||||
if (REFRESHING) {
|
||||
await writeFile(join(dir, 'stdout.golden.jsonl'), stdout)
|
||||
await writeFile(join(dir, 'stdout.expected.jsonl'), stdout)
|
||||
}
|
||||
await expect(stdout).toMatchFileSnapshot(join(dir, 'stdout.golden.jsonl'))
|
||||
await expect(stdout).toMatchFileSnapshot(join(dir, 'stdout.expected.jsonl'))
|
||||
|
||||
// A model turn always produces a log worth comparing; a hook scenario can
|
||||
// produce one without a model turn (a `rejected` turn carrying `hook/*`).
|
||||
if (comparesLog) {
|
||||
// The harvested logs (primary-first) must match their committed fixtures 1:1.
|
||||
expect(result.sessionLogs.length, 'this scenario must persist a session log').toBe(childSessions + 1)
|
||||
expect(result.sessionLogs.length, 'this scenario must persist one log per session fixture').toBe(fixtureFiles.length)
|
||||
for (let i = 0; i < fixtureFiles.length; i++) {
|
||||
const harvested = scrub((result.sessionLogs[i] as HarvestedLog).content)
|
||||
const fixture = scrub(await readFile(join(dir, fixtureFiles[i] as string), 'utf8'))
|
||||
@@ -611,8 +590,8 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
}
|
||||
}
|
||||
|
||||
// Header-uniformity guard: every live header in a class must equal the class pin split
|
||||
// across tokenized JSONL plus readable prompt and structured schema sidecars.
|
||||
// Every live full header must equal its class pin reconstructed from
|
||||
// tokenized JSONL plus readable prompt and structured schema sidecars.
|
||||
/* v8 ignore next -- construction guarantees the pin exists; a miss would fail the one-header assertion loudly. */
|
||||
const pinningScenario = pinningByClass.get(classOf(scenario)) ?? scenario
|
||||
const pinningDir = join(snapshotsDir, pinningScenario.name)
|
||||
@@ -620,17 +599,23 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
const pinned = normalizedHeaders(pinnedFixture, fixtureContext(pinnedFixture))
|
||||
const promptSnapshot = await readFile(join(pinningDir, SYSTEM_PROMPT_SNAPSHOT), 'utf8')
|
||||
const initialPromptSnapshot = initialSystemPromptSnapshot(promptSnapshot)
|
||||
expect(pinned.length, `the pinning fixture (${pinningScenario.name}) has an unexpected request/header count`)
|
||||
.toBe(1 + (pinningScenario.expectedHeaderChanges ?? 0))
|
||||
const toolSchemasSnapshot = await readFile(join(pinningDir, TOOL_SCHEMAS_SNAPSHOT), 'utf8')
|
||||
const toolSchemas = parseToolSchemasSnapshot(toolSchemasSnapshot)
|
||||
expect(pinned.length, `the pinning fixture (${pinningScenario.name}) must carry exactly one request/header`)
|
||||
.toBe(1)
|
||||
const pinnedHeader = restorePinnedToolSchemas(pinned[0], toolSchemas)
|
||||
const pinnedSchemaSets = [toolSchemas.initial, ...toolSchemas.changes]
|
||||
expect(pinnedSchemaSets.length, `the pinning fixture (${pinningScenario.name}) has an unexpected tool-schema count`)
|
||||
.toBe(pinned.length)
|
||||
const pinnedHeaders = pinned.map((header, index) => restorePinnedToolSchemas(
|
||||
header,
|
||||
pinnedSchemaSets[index] as unknown[],
|
||||
))
|
||||
for (const [logIndex, log] of result.sessionLogs.entries()) {
|
||||
const expectedDeltas = scenario.pinsHeader === true && logIndex === 0
|
||||
? scenario.expectedHeaderDeltas ?? 0
|
||||
const expectedChanges = scenario.pinsHeader === true && logIndex === 0
|
||||
? scenario.expectedHeaderChanges ?? 0
|
||||
: 0
|
||||
expect(headerDeltaCount(log.content), `session ${log.id}: request/header-delta count`)
|
||||
.toBe(expectedDeltas)
|
||||
expect(headerChangeCount(log.content), `session ${log.id}: changed request/header count`)
|
||||
.toBe(expectedChanges)
|
||||
const headers = normalizedHeaders(scrubSystemPrompts(log.content), ctx)
|
||||
const prompts = normalizedSystemPrompts(log.content, ctx)
|
||||
const schemaSets = normalizedToolSchemas(log.content, ctx)
|
||||
@@ -639,31 +624,34 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
expect(schemaSets.length, `session ${log.id}: every request/header must carry an array-valued tools field`)
|
||||
.toBe(headers.length)
|
||||
for (const [k, header] of headers.entries()) {
|
||||
const expected = expectedChanges > 0 ? pinnedHeaders[k] : pinnedHeaders[0]
|
||||
expect(header, `session ${log.id}: request/header #${k + 1} diverged from the pinned (${pinningScenario.name}) header`)
|
||||
.toEqual(pinnedHeader)
|
||||
expect(formatSystemPromptSnapshot(prompts[k] as string), `session ${log.id}: initial system prompt #${k + 1} diverged from ${pinningScenario.name}/${SYSTEM_PROMPT_SNAPSHOT}`)
|
||||
.toEqual(initialPromptSnapshot)
|
||||
.toEqual(expected)
|
||||
if (expectedChanges === 0) {
|
||||
expect(formatSystemPromptSnapshot(prompts[k] as string), `session ${log.id}: initial system prompt #${k + 1} diverged from ${pinningScenario.name}/${SYSTEM_PROMPT_SNAPSHOT}`)
|
||||
.toEqual(initialPromptSnapshot)
|
||||
}
|
||||
}
|
||||
if (scenario.pinsHeader === true && logIndex === 0) {
|
||||
expect(formatSystemPromptSnapshot(
|
||||
prompts[0] as string,
|
||||
normalizedSystemPromptDeltas(log.content, ctx),
|
||||
), `session ${log.id}: system-prompt deltas diverged from ${pinningScenario.name}/${SYSTEM_PROMPT_SNAPSHOT}`)
|
||||
prompts.slice(1),
|
||||
), `session ${log.id}: changed system prompts diverged from ${pinningScenario.name}/${SYSTEM_PROMPT_SNAPSHOT}`)
|
||||
.toEqual(promptSnapshot)
|
||||
expect(formatToolSchemasSnapshot(
|
||||
schemaSets[0] as unknown[],
|
||||
normalizedToolSchemaDeltas(log.content, ctx),
|
||||
), `session ${log.id}: tool-schema deltas diverged from ${pinningScenario.name}/${TOOL_SCHEMAS_SNAPSHOT}`)
|
||||
schemaSets.slice(1),
|
||||
), `session ${log.id}: changed tool schemas diverged from ${pinningScenario.name}/${TOOL_SCHEMAS_SNAPSHOT}`)
|
||||
.toEqual(toolSchemasSnapshot)
|
||||
}
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
describe('snapshot fixtures', () => {
|
||||
it('every scenario directory is registered (no orphans)', async () => {
|
||||
// toMatchFileSnapshot does not prune orphaned golden/fixture files, so a
|
||||
// toMatchFileSnapshot does not prune orphaned expected-output or fixture files, so a
|
||||
// renamed/removed scenario could leave a stale dir that nothing exercises.
|
||||
// Fail loud on any snapshots/<dir> not present in the scenario table.
|
||||
const entries = await readdir(snapshotsDir, { withFileTypes: true })
|
||||
@@ -672,12 +660,12 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
expect(onDisk).toEqual(registered)
|
||||
})
|
||||
|
||||
it('every registered scenario has its required fixture files', () => {
|
||||
// Every scenario has an input script and an stdout golden.
|
||||
for (const { name, overridden, childSessions, pinsHeader } of scenarios) {
|
||||
it('every registered scenario has its required fixture files', async () => {
|
||||
// Every scenario needs input, stdout, a primary session fixture, and matching optional sidecars.
|
||||
for (const { name, overridden, pinsHeader } of scenarios) {
|
||||
const dir = join(snapshotsDir, name)
|
||||
expect(existsSync(join(dir, 'input.json')), `${name}/input.json`).toBe(true)
|
||||
expect(existsSync(join(dir, 'stdout.golden.jsonl')), `${name}/stdout.golden.jsonl`).toBe(true)
|
||||
expect(existsSync(join(dir, 'stdout.expected.jsonl')), `${name}/stdout.expected.jsonl`).toBe(true)
|
||||
expect(existsSync(join(dir, 'session.jsonl')), `${name}/session.jsonl`).toBe(true)
|
||||
expect(existsSync(join(dir, 'replay.override.json')), `${name}/replay.override.json presence must match \`overridden\``)
|
||||
.toBe(overridden === true)
|
||||
@@ -685,11 +673,7 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
.toBe(pinsHeader === true)
|
||||
expect(existsSync(join(dir, TOOL_SCHEMAS_SNAPSHOT)), `${name}/${TOOL_SCHEMAS_SNAPSHOT} presence must match \`pinsHeader\``)
|
||||
.toBe(pinsHeader === true)
|
||||
// A nested-agent scenario ships one child fixture per recorded subagent
|
||||
// session (`session.1.jsonl` …), the replay source for that child session.
|
||||
for (const childFixture of childFixturePaths(dir, childSessions ?? 0)) {
|
||||
expect(existsSync(childFixture), childFixture).toBe(true)
|
||||
}
|
||||
await expect(sessionFixtures(dir), `${name}: session fixture inventory`).resolves.toBeDefined()
|
||||
}
|
||||
})
|
||||
|
||||
@@ -708,25 +692,30 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
}
|
||||
})
|
||||
|
||||
it('every pinning fixture carries one tokenized request/header, two sidecars, and its declared deltas', async () => {
|
||||
// The live uniformity guard runs only in NON-pinning scenarios, so a class made of just
|
||||
// its pinning scenario would otherwise accept a re-recorded pin with several headers or
|
||||
// an undeclared mid-run header-delta — shapes the pin design cannot represent.
|
||||
it('every pinning fixture carries one tokenized header sequence and two sidecars', async () => {
|
||||
// Assert the committed pin directly because a class containing only its
|
||||
// pinning scenario has no non-pinning live run to catch undeclared changes.
|
||||
for (const scenario of pinningByClass.values()) {
|
||||
const fixture = await readFile(join(snapshotsDir, scenario.name, 'session.jsonl'), 'utf8')
|
||||
const headers = normalizedHeaders(fixture, fixtureContext(fixture))
|
||||
const promptSnapshot = await readFile(join(snapshotsDir, scenario.name, SYSTEM_PROMPT_SNAPSHOT), 'utf8')
|
||||
expect(headers.length, `${scenario.name}: unexpected request/header count`)
|
||||
.toBe(1 + (scenario.expectedHeaderChanges ?? 0))
|
||||
const toolSchemasSnapshot = await readFile(join(snapshotsDir, scenario.name, TOOL_SCHEMAS_SNAPSHOT), 'utf8')
|
||||
const toolSchemas = parseToolSchemasSnapshot(toolSchemasSnapshot)
|
||||
expect(headers.length, `${scenario.name}: a pinning fixture must carry exactly one request/header`).toBe(1)
|
||||
expect(() => restorePinnedToolSchemas(headers[0], toolSchemas), `${scenario.name}: tools must use the sidecar token`)
|
||||
.not.toThrow()
|
||||
const schemaSets = [toolSchemas.initial, ...toolSchemas.changes]
|
||||
expect(schemaSets.length, `${scenario.name}: tool-schema sequence must match the header sequence`)
|
||||
.toBe(headers.length)
|
||||
for (const [index, header] of headers.entries()) {
|
||||
expect(() => restorePinnedToolSchemas(header, schemaSets[index] as unknown[]), `${scenario.name}: tools must use the sidecar token`)
|
||||
.not.toThrow()
|
||||
}
|
||||
expect(promptSnapshot.length, `${scenario.name}/${SYSTEM_PROMPT_SNAPSHOT} must not be empty`).toBeGreaterThan(0)
|
||||
expect(promptSnapshot.endsWith('\n'), `${scenario.name}/${SYSTEM_PROMPT_SNAPSHOT} must end in a newline`).toBe(true)
|
||||
expect(toolSchemasSnapshot, `${scenario.name}/${TOOL_SCHEMAS_SNAPSHOT} must use canonical JSON formatting`)
|
||||
.toBe(formatToolSchemasSnapshot(toolSchemas.initial, toolSchemas.deltas))
|
||||
expect(headerDeltaCount(fixture), `${scenario.name}: a pinning fixture must carry exactly its declared request/header-deltas`)
|
||||
.toBe(scenario.expectedHeaderDeltas ?? 0)
|
||||
.toBe(formatToolSchemasSnapshot(toolSchemas.initial, toolSchemas.changes))
|
||||
expect(headerChangeCount(fixture), `${scenario.name}: a pinning fixture must carry exactly its declared changed headers`)
|
||||
.toBe(scenario.expectedHeaderChanges ?? 0)
|
||||
}
|
||||
})
|
||||
|
||||
@@ -736,10 +725,7 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
|
||||
// storage rules fail loud.
|
||||
for (const scenario of scenarios) {
|
||||
const dir = join(snapshotsDir, scenario.name)
|
||||
const files = [
|
||||
'session.jsonl',
|
||||
...Array.from({ length: scenario.childSessions ?? 0 }, (_, i) => `session.${i + 1}.jsonl`),
|
||||
]
|
||||
const files = await sessionFixtures(dir)
|
||||
for (const file of files) {
|
||||
const fixture = await readFile(join(dir, file), 'utf8')
|
||||
expect(unknownToolCallIds(fixture), `${scenario.name}/${file} contains UNKNOWN_TOOL`)
|
||||
|
||||
Reference in New Issue
Block a user