Merge remote-tracking branch 'origin/master' into codex/simp-prune-llm-contract

2026-07-19 02:10:25 +08:00
parent 90cbe2573b 8518dd31c7
commit 7b076a1e01
72 changed files with 3147 additions and 792 deletions
--- a/packages/support/README.md
+++ b/packages/support/README.md
@@ -4,11 +4,11 @@ Packages that exist to serve development, testing, and the examples rather than

 | Package | Role | ctx key |
 |---|---|---|
-| `acp-snapshot/` | ACP snapshot suite kit: subprocess scenario harness + golden normalizers + the `defineAcpSnapshotSuite` factory | (library — imported by example `*.snapshot.ts` suites) |
+| `acp-snapshot/` | ACP test kit: shared subprocess/client launcher + snapshot harness, normalizers, and suite factory | (library — imported by ACP e2e and `*.snapshot.ts` suites) |
 | `agent-loop-testkit/` | Shared prerequisite mounting for tests that exercise the concrete agent loop | (library — imported by AgentLoop integration tests) |
 | `invariants/` | Runtime event-contract assertions for development diagnostics | (listens on `session/*`, `agent/*`) |
 | `loader-smoke/` | Shared real-Loader subprocess harness for keyless example smokes | (library — imported by example e2e suites) |
 | `llm-replay/` | Record/replay adapter: short-circuits `llm/stream` from a recorded session JSONL (keyless snapshot tests) | (listens on `llm/stream`) |
 | `subagent-mock/` | Scripted `SubagentProvider` for deterministic seam/tool tests | (registers on `ctx.subagents`) |

-`invariants` is development support but has no environment guard: it runs wherever registered, and the default `dsh-agent-spine-demo` bundle mounts it unconditionally. `agent-loop-testkit` centralizes the mandatory service spine for hand-built AgentLoop tests without owning their loop or scenario. `llm-replay` backs the demos and the snapshot test tier under the per-file coverage gate. `acp-snapshot` carries the snapshot tier's harness/normalizer/suite machinery, while `loader-smoke` owns the parallel stdio/Loader process boundary used by keyless example e2e suites. `subagent-mock` exercises the real `ctx.subagents` load path without a model or child agent. A package graduates OUT of `support/` into a product group only when it gains documented product consumers.
+`invariants` is development support but has no environment guard: it runs wherever registered, and the default `dsh-agent-spine-demo` bundle mounts it unconditionally. `agent-loop-testkit` centralizes the mandatory service spine for hand-built AgentLoop tests without owning their loop or scenario. `llm-replay` backs the demos and the snapshot test tier under the per-file coverage gate. `acp-snapshot` carries the ACP subprocess/client boundary plus the snapshot harness, normalizers, and suite machinery, while `loader-smoke` owns the parallel stdio/Loader process boundary used by keyless example e2e suites. `subagent-mock` exercises the real `ctx.subagents` load path without a model or child agent. A package graduates OUT of `support/` into a product group only when it gains documented product consumers.
--- a/packages/support/acp-snapshot/README.md
+++ b/packages/support/acp-snapshot/README.md
@@ -2,11 +2,12 @@

 The ACP snapshot suite kit: the shared machinery behind the keyless snapshot tier (`pnpm run test:snapshot`, [testing policy](../../../docs/testing.md)). An example gets a full snapshot suite from a scenario table plus a fixtures directory; every compare/guard mechanic lives here, under the per-file coverage gate, instead of being copied per example.

-Three layers, importable separately:
+Four layers, importable separately:

+- **`launchAcpTestAgent` (launcher)** — boots an unbuilt ACP agent from a temp cwd, pins tsx to the repo tsconfig, connects the SDK client over a raw-byte stdout tee, collects session updates and stderr, surfaces asynchronous spawn failures through its startup lifecycle, fails closed on unhandled permission requests, and owns graceful or signalled shutdown. Shutdown waits for process exit, inherited stdio closure, and ACP parser exhaustion before resolving or propagating a child error, so captures are complete and callers can remove owned paths after either outcome. Snapshot and ordinary e2e suites share this process boundary; a test supplies only agent paths, cwd, environment overrides, and any permission policy.
 - **`runScenario` (harness)** — boots the real agent bin as a subprocess via tsx (unbuilt, Loader path), drives it over ACP JSON-RPC stdio from a deterministic `input.json` script, tees raw stdout for the golden + purity check, and harvests every persisted session JSONL (parent + subagent children, primary-first) after a graceful stdin-EOF shutdown. Parameterized by `AgentUnderTest` (`binScript`, `configPath`, `tsconfigPath` — absolute paths; the subprocess cwd is a temp dir outside the repo). Startup failures preserve captured agent stderr in the rejected diagnostic.
 - **Normalizers** — pure functions turning the two captured surfaces into stable text: `normalizeStdout` (JSON-RPC ids → first-seen sequence; UUIDs/cwd → tokens; doubles as the stdout-purity check), `normalizeSessionLog` (times zeroed, `seq` kept), `scrubSystemPrompts` (prompt text → `{{system}}`), `scrubToolSchemas` (schema bulk → `{{tools}}`), and `scrubRequestHeaders` (all header bulk → `{{system}}`/`{{tools}}`/`{{messagePrefix}}` outside each pin, structure kept — [pinned-header RFC](../../../docs/rfc/implemented/testing/2026-07-06-pin-request-header-content-in-one-scenario.md)).
- **`defineAcpSnapshotSuite` (factory)** — registers the whole describe/it tree for a scenario table: per-scenario golden + re-persisted-log compares, record/refresh fixture write-back, rejection of structured `UNKNOWN_TOOL` results, the per-header-class pin (`system-prompt.golden.md` plus `tool-schemas.golden.json`) with its live uniformity guard, and the fixture guard block (no orphan scenario dirs, required files present, exactly one pin per class, every JSONL prompt/schema-scrubbed, non-pinning fixtures fully header-scrubbed). Must be called at vitest collection time.
+- **`defineAcpSnapshotSuite` (factory)** — registers the whole describe/it tree for a scenario table: per-scenario golden + re-persisted-log compares, record/refresh fixture write-back, rejection of structured `UNKNOWN_TOOL` results, the per-header-class pin (`system-prompt.golden.md` plus `tool-schemas.golden.json`) with its live uniformity guard, and the fixture guard block (no orphan scenario dirs, required files present, exactly one pin per class, every JSONL prompt/schema-scrubbed, non-pinning fixtures fully header-scrubbed). Each scenario directory's `session.jsonl` plus contiguous `session.<n>.jsonl` siblings are the ordered primary/child inventory; the scenario table does not duplicate their count. Must be called at vitest collection time.

 A consuming `*.snapshot.ts` is the scenario table plus one factory call:

@@ -37,9 +38,9 @@ defineAcpSnapshotSuite({

 A scenario booting a differently-composed tree sets its own `configPath` (an overlay whose basename still ends in `cordis.yml`, so the bin's replay swap finds the sibling `*cordis.snapshot.yml`) and, when that composition changes the request header, its own `headerClass` with its own pinning scenario — the acp-agent example's Code Mode and filesystem scenarios are templates. Each pinning directory stores the normalized full prompt sequence in generated `system-prompt.golden.md` and the corresponding full tool-schema sequence in generated `tool-schemas.golden.json`; `session.jsonl` stores `"system":"{{system}}","tools":"{{tools}}"` while retaining config, reason, and any model-visible prefix. A pin with legitimate mid-run header changes declares `expectedHeaderChanges`, which fixes the length of both sidecar sequences.

-Examples use a `cordis.snapshot.yml` overlay with [`dsh-llm-replay`](../llm-replay/README.md). Recording calls the live model and updates model fixtures; keyless refresh replays those fixtures and updates derived stdout, session-log, prompt, and tool-schema snapshots. See the [snapshot RFC](../../../docs/rfc/implemented/testing/2026-06-19-acp-snapshot-tests.md).
+The example also ships a `cordis.snapshot.yml` replay overlay next to its `cordis.yml` (the bin swaps them under `DSH_SNAPSHOT=replay` — [single-source replay config RFC](../../../docs/rfc/implemented/testing/2026-07-04-single-source-acp-replay-config.md)); replay fixtures are served by [`dsh-llm-replay`](../llm-replay/README.md), which this package points at via the `DSH_SNAPSHOT_*` env vars it sets on the child. `pnpm run test:snapshot:record` calls the live LLM and rewrites the recorded scenarios' model fixtures; `pnpm run test:snapshot:refresh` stays keyless, runs the replay overlay, and rewrites stdout, comparable session-log goldens, and each pin's prompt and tool-schema sidecars from the committed model scripts. Fixture roles, record/replay/refresh semantics, and scenario-table fields are documented on `Scenario` and in the [snapshot RFC](../../../docs/rfc/implemented/testing/2026-06-19-acp-snapshot-tests.md).

-`suite.ts` imports Vitest, so use this package only inside a Vitest run. The ACP-specific script queues permission answers by stable option kind and maps them to current option ids; a missing answer cancels, while an unavailable kind fails the scenario after cancelling the agent request. It can also set session config options or assert that unknown ids and values are rejected in the transcript.
+Constraints: `suite.ts` imports vitest, so the package entry is importable only inside a vitest run (the launcher, harness, and normalizers have no such dependency but ship from the same entry). ACP-specific by design — the launcher speaks the SDK's `ClientSideConnection`. Permission round-trips are scriptable: `InputScript.permissionAnswers` is a FIFO queue of option-kind selections (`allow_once`, `reject_once`, …) the client maps to the agent-issued `optionId` at answer time; an absent or exhausted queue answers `cancelled`, and a kind the request never offered rejects the run (the agent is answered `cancelled`, so a tolerant agent cannot absorb the scenario bug). Session config options are scriptable too: the `setConfigOption` step switches a knob over `session/set_config_option`, and `setConfigOptionExpectError` asserts the bridge rejects an unknown id or out-of-vocabulary value (the error frame stays in the transcript).

 ## Model Experience

--- a/packages/support/acp-snapshot/package.json
+++ b/packages/support/acp-snapshot/package.json
@@ -1,6 +1,6 @@
 {
  "name": "@deepseek-ai/dsh-acp-snapshot",
-  "description": "ACP snapshot suite kit: real-subprocess scenario harness, golden normalizers, and the suite factory behind the keyless snapshot tier",
+  "description": "ACP test kit: shared subprocess launcher, snapshot scenario harness, golden normalizers, and suite factory",
  "version": "0.0.1",
  "private": true,
  "type": "module",
--- a/packages/support/acp-snapshot/src/harness.ts
+++ b/packages/support/acp-snapshot/src/harness.ts
@@ -1,59 +1,47 @@
 /**
- * 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 golden + a purity check) 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 docs/rfc/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 { 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 { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
+import { launchAcpTestAgent, type AgentUnderTest, type LaunchedAcpTestAgent } from './launcher.ts'
+
+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's SOURCE entry (e.g. `packages/examples/acp-demo/src/bin.ts`); the `lib` bin is derived from it. */
-  binScript: string
-  /** Explicit plain-Node entry for `lib` mode; intended for test fixtures outside a package `src/` tree. */
-  libBinScript?: string | undefined
-  /**
-   * 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 in `src` mode (passed to the child as `TSX_TSCONFIG_PATH`). Ignored
-   * in `lib` mode, where the example resolves plugins through real `exports`.
-   */
-  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` sends a prompt WITHOUT awaiting its response, waits until
+ * the client observes the first streamed `agent_message_chunk` (so the emitted
+ * frames deterministically precede the cancellation), then cancels the turn —
+ * the only way to exercise a cancel deterministically (a plain `prompt` step
+ * awaits the response, which a cancel/hang scenario would block on forever).
 */
 export type InputStep =
  | { op: 'initialize'; terminalOutput?: boolean }
@@ -70,9 +58,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[]
 }
@@ -165,93 +160,47 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
  // Fixed path length: spill-policy budgets the preview against the REAL path
  // before stdout normalization, so tmpdir() length differences churn goldens.
  const spillRoot = '/tmp/dsh-acp-snapshot-spill'
-  // 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
+  // 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 goldens
+    // 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 })
    }
-    // Boot the agent in the environment's mode (DSH_EXAMPLE_MODE): `src` runs the
-    // source bin under tsx with the paths map; `lib` runs the built bin under plain
-    // Node, resolving plugins through the example's workspace node_modules → lib.
-    const launch = resolveExampleLaunch({
-      srcBin: opts.agent.binScript,
-      libBin: opts.agent.libBinScript,
-      configArgs: ['--config', opts.configPath ?? opts.agent.configPath],
-      tsconfigPath: opts.agent.tsconfigPath,
-      env: {
-        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 } : {},
-        ...opts.childFiles !== undefined && opts.childFiles.length > 0
-          ? { DSH_SNAPSHOT_CHILD_FILES: opts.childFiles.join(delimiter) }
-          : {},
-      },
-    })
-
-    child = spawn(
-      launch.command,
-      launch.args,
-      { cwd, env: { ...process.env, ...launch.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 }))
+    const env: NodeJS.ProcessEnv = {
+      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 } : {},
+      ...opts.childFiles !== undefined && opts.childFiles.length > 0
+        ? { DSH_SNAPSHOT_CHILD_FILES: opts.childFiles.join(delimiter) }
+        : {},
+    }

    // 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' } })
@@ -269,10 +218,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
@@ -281,35 +232,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)
-  } catch (error: unknown) {
-    const stderr = stderrChunks.join('')
-    if (stderr === '') throw error
-    throw new Error(`snapshot-harness: scenario failed: ${String(error)}\nagent stderr:\n${stderr}`, { cause: error })
-  } 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 })
-    await rm(spillRoot, { 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. */
@@ -317,7 +290,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> {
@@ -334,8 +307,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: [],
@@ -355,8 +330,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 */ })
@@ -365,8 +342,13 @@ 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 the prompt WITHOUT awaiting (a hang fixture never resolves on
+      // its own). To pin frame order deterministically, wait until the client
+      // has OBSERVED the hang's streamed agent_message_chunk before cancelling —
+      // so those update frames always precede the cancelled prompt response in
+      // the transcript (without this, the late chunk and the response race).
+      // Then cancel and await the prompt, which the bridge settles as
+      // `cancelled` once the abort propagates.
      const promptDone = client.prompt({ sessionId, prompt: [{ type: 'text', text: step.text }] })
      await waitForUpdate(u => u.sessionUpdate === 'agent_message_chunk')
      await client.cancel({ sessionId })
@@ -402,16 +384,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
@@ -452,8 +424,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
--- a/packages/support/acp-snapshot/src/index.ts
+++ b/packages/support/acp-snapshot/src/index.ts
@@ -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 golden 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,
--- a/packages/support/acp-snapshot/src/launcher.ts
+++ b/packages/support/acp-snapshot/src/launcher.ts
@@ -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
+}
--- a/packages/support/acp-snapshot/src/suite.ts
+++ b/packages/support/acp-snapshot/src/suite.ts
@@ -15,7 +15,7 @@
 * @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'
@@ -71,14 +71,6 @@ 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.
@@ -129,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))
 }

 /**
@@ -454,7 +472,12 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
        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,
@@ -463,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.
@@ -491,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')))
          : []
@@ -500,18 +522,37 @@ 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 primary = result.sessionLogs[0] as HarvestedLog
            const prompts = normalizedSystemPrompts(primary.content, ctx)
@@ -540,7 +581,7 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
        // 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'))
@@ -619,9 +660,9 @@ 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)
@@ -632,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()
      }
    })

@@ -688,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`)
--- a/packages/support/acp-snapshot/tests/fixtures/fake-acp-agent.ts
+++ b/packages/support/acp-snapshot/tests/fixtures/fake-acp-agent.ts
@@ -6,6 +6,7 @@

 import { mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
 import { readdirSync } from 'node:fs'
+import { spawn } from 'node:child_process'
 import { dirname, join } from 'node:path'
 import { randomUUID } from 'node:crypto'
 import { createInterface } from 'node:readline'
@@ -40,6 +41,8 @@ interface Behavior {
  echoWorkspace?: boolean
  /** Write a line to stderr on boot (spec-side stderr-capture assertions). */
  stderrNote?: string
+  /** Let a short-lived descendant retain stdio and emit one final ACP update plus stderr line after this parent exits. */
+  lateInheritedOutput?: boolean
  /** Session logs to persist on stdin EOF. */
  logs?: ScriptedLog[]
  /** Leave a stray FILE directly under the sessions root (harvest must skip it). */
@@ -247,6 +250,24 @@ function flushLogsAndExit(): void {
    writeFileSync(join(sessionsRoot, 'bucket-noise', 'notes.txt'), 'not a session log\n')
  }
  if (behavior.deleteSessionsRoot === true) rmSync(sessionsRoot, { recursive: true, force: true })
+  if (behavior.lateInheritedOutput === true) {
+    const frame = JSON.stringify({
+      jsonrpc: '2.0',
+      method: 'session/update',
+      params: {
+        sessionId,
+        update: {
+          sessionUpdate: 'agent_message_chunk',
+          content: { type: 'text', text: 'late inherited stdout' },
+        },
+      },
+    })
+    const code = [
+      `setTimeout(() => process.stdout.write(${JSON.stringify(`${frame}\n`)}), 50)`,
+      `setTimeout(() => process.stderr.write(${JSON.stringify('late inherited stderr\n')}), 75)`,
+    ].join(';')
+    spawn(process.execPath, ['-e', code], { stdio: ['ignore', 1, 2] }).unref()
+  }
  process.exit(0)
 }

--- a/packages/support/acp-snapshot/tests/harness.spec.ts
+++ b/packages/support/acp-snapshot/tests/harness.spec.ts
@@ -1,13 +1,34 @@
 import { mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { once } from 'node:events'
 import { tmpdir } from 'node:os'
 import { delimiter, join } from 'node:path'
 import { fileURLToPath } from 'node:url'
-import { afterAll, describe, expect, it } from 'vitest'
+import { afterAll, describe, expect, it, vi } from 'vitest'
+import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
 import { runScenario, type AgentUnderTest, type InputStep } from '../src/harness.ts'
+import { launchAcpTestAgent } from '../src/launcher.ts'
+
+const fsControl = vi.hoisted(() => ({ cleanupFailure: undefined as Error | undefined }))
+
+vi.mock('node:fs/promises', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('node:fs/promises')>()
+  return {
+    ...actual,
+    async rm(...args: Parameters<typeof actual.rm>): Promise<void> {
+      if (String(args[0]).includes('acp-snap-cwd-') && fsControl.cleanupFailure !== undefined) {
+        const failure = fsControl.cleanupFailure
+        fsControl.cleanupFailure = undefined
+        await actual.rm(...args)
+        throw failure
+      }
+      await actual.rm(...args)
+    },
+  }
+})

 /**
 * Unit tests for the subprocess harness, driven through the REAL spawn path
- * (tsx loader, temp cwd, env plumbing) against the scripted fake ACP bin in
+ * (mode-aware launcher, temp cwd, env plumbing) against the scripted fake ACP bin in
 * ./fixtures/fake-acp-agent.ts. Each case writes a `behavior.json` next to a
 * throwaway fixture path; the fake bin echoes observable facts (env, seeded
 * workspace, permission outcomes) into `agent_message_chunk` text, so the
@@ -40,6 +61,181 @@ async function scenario(behavior: object): Promise<{ dir: string; fixtureFile: s
 const boot: InputStep[] = [{ op: 'initialize' }, { op: 'newSession' }]

 describe('runScenario', () => {
+  it('surfaces an asynchronous child spawn failure through startup and close', async () => {
+    const { dir } = await scenario({})
+    const launched = launchAcpTestAgent({ agent: AGENT, cwd: join(dir, 'missing') })
+    let stdioClosed = false
+    let clientClosed = false
+    launched.child.once('close', () => { stdioClosed = true })
+    void launched.client.closed.then(
+      () => { clientClosed = true },
+      () => { clientClosed = true },
+    )
+    await expect(launched.spawned).rejects.toMatchObject({ code: 'ENOENT' })
+    await expect(launched.close()).rejects.toMatchObject({ code: 'ENOENT' })
+    expect(stdioClosed).toBe(true)
+    expect(clientClosed).toBe(true)
+  })
+
+  it('centralizes ACP boot, captures, updates, fail-closed permissions, and shutdown', { timeout: 20_000 }, async () => {
+    const { dir, fixtureFile } = await scenario({ permissionProbe: true, echoEnv: true, stderrNote: 'launcher stderr' })
+    const sessionsRoot = await mkdtemp(join(tmpdir(), 'acp-launcher-sessions-'))
+    tempDirs.push(sessionsRoot)
+    const launched = launchAcpTestAgent({
+      agent: AGENT,
+      cwd: dir,
+      configPath: AGENT.configPath,
+      env: {
+        DSH_SNAPSHOT: 'replay',
+        DSH_SNAPSHOT_FILE: fixtureFile,
+        DSH_SNAPSHOT_SESSIONS_ROOT: sessionsRoot,
+      },
+    })
+    await launched.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
+    const { sessionId } = await launched.client.newSession({ cwd: dir, mcpServers: [] })
+    const nextChunk = launched.waitForUpdate(update => update.sessionUpdate === 'agent_message_chunk')
+    const predicateFailure = new Error('predicate failed')
+    const failedPredicate = launched.waitForUpdate(() => { throw predicateFailure })
+      .catch((error: unknown): unknown => error)
+    await launched.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
+    expect(await failedPredicate).toBe(predicateFailure)
+    expect((await nextChunk).sessionUpdate).toBe('agent_message_chunk')
+    expect(launched.updates.some(update => update.sessionUpdate === 'agent_message_chunk')).toBe(true)
+    expect(launched.rawStdout()).toContain('permission:{\\"outcome\\":\\"cancelled\\"}')
+    expect(launched.stderr()).toContain('launcher stderr')
+    const unmatched = expect(launched.waitForUpdate(() => false)).rejects.toThrow(/update stream closed/)
+    await launched.close()
+    await unmatched
+    await expect(launched.waitForUpdate(() => true)).rejects.toThrow(/update stream closed/)
+    await launched.close('SIGKILL')
+
+    // The minimal shape needs no environment or config override.
+    const minimal = launchAcpTestAgent({ agent: AGENT, cwd: dir })
+    await minimal.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
+    const childFailure = new Error('child process failed')
+    let exited = false
+    minimal.child.once('exit', () => { exited = true })
+    minimal.child.emit('error', childFailure)
+    await expect(minimal.close('SIGTERM')).rejects.toBe(childFailure)
+    // close rejects only after the fallback SIGKILL has produced an exit edge.
+    expect(exited).toBe(true)
+  })
+
+  it('waits for inherited stdio and buffered ACP parsing after the parent exits', { timeout: 20_000 }, async () => {
+    const { dir, fixtureFile } = await scenario({ lateInheritedOutput: true })
+    const launched = launchAcpTestAgent({
+      agent: AGENT,
+      cwd: dir,
+      env: { DSH_SNAPSHOT_FILE: fixtureFile },
+    })
+    await launched.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
+    await launched.client.newSession({ cwd: dir, mcpServers: [] })
+    const lateUpdate = launched.waitForUpdate(update =>
+      update.sessionUpdate === 'agent_message_chunk'
+      && update.content.type === 'text'
+      && update.content.text === 'late inherited stdout')
+
+    await launched.close()
+
+    await expect(lateUpdate).resolves.toMatchObject({ sessionUpdate: 'agent_message_chunk' })
+    expect(launched.rawStdout()).toContain('late inherited stdout')
+    expect(launched.stderr()).toContain('late inherited stderr')
+  })
+
+  it('rejects promptly when fallback termination is refused', async () => {
+    const { dir } = await scenario({})
+    const launched = launchAcpTestAgent({ agent: AGENT, cwd: dir })
+    await launched.spawned
+
+    const childFailure = Object.assign(new Error('signal refused'), { code: 'EPERM' })
+    const originalKill = launched.child.kill.bind(launched.child)
+    const kill = vi.spyOn(launched.child, 'kill').mockReturnValue(false)
+    const closed = new Promise<void>(resolve => launched.child.once('close', () => { resolve() }))
+    try {
+      launched.child.emit('error', childFailure)
+      const rejection = await launched.close('SIGTERM').catch((error: unknown): unknown => error)
+      expect(rejection).toBeInstanceOf(AggregateError)
+      expect(rejection).toMatchObject({
+        message: 'ACP test agent failed and fallback termination was refused',
+        errors: [
+          childFailure,
+          expect.objectContaining({ message: 'Fallback SIGKILL was not accepted by the child process' }),
+        ],
+      })
+      expect(kill).toHaveBeenNthCalledWith(1, 'SIGTERM')
+      expect(kill).toHaveBeenNthCalledWith(2, 'SIGKILL')
+    } finally {
+      kill.mockRestore()
+      originalKill('SIGKILL')
+      await closed
+    }
+  })
+
+  it('rejects promptly when fallback termination emits an error', async () => {
+    const { dir } = await scenario({})
+    const launched = launchAcpTestAgent({ agent: AGENT, cwd: dir })
+    await launched.spawned
+
+    const childFailure = Object.assign(new Error('signal refused'), { code: 'EPERM' })
+    const fallbackFailure = Object.assign(new Error('fallback signal refused'), { code: 'EPERM' })
+    const originalKill = launched.child.kill.bind(launched.child)
+    const kill = vi.spyOn(launched.child, 'kill').mockImplementation((signal) => {
+      if (signal === 'SIGKILL') queueMicrotask(() => launched.child.emit('error', fallbackFailure))
+      return signal === 'SIGKILL'
+    })
+    const closed = new Promise<void>(resolve => launched.child.once('close', () => { resolve() }))
+    try {
+      launched.child.emit('error', childFailure)
+      const rejection = await launched.close('SIGTERM').catch((error: unknown): unknown => error)
+      expect(rejection).toBeInstanceOf(AggregateError)
+      expect(rejection).toMatchObject({
+        message: 'ACP test agent failed and fallback termination was refused',
+        errors: [childFailure, fallbackFailure],
+      })
+      expect(kill).toHaveBeenNthCalledWith(1, 'SIGTERM')
+      expect(kill).toHaveBeenNthCalledWith(2, 'SIGKILL')
+    } finally {
+      kill.mockRestore()
+      originalKill('SIGKILL')
+      await closed
+    }
+  })
+
+  it('waits for in-flight client callbacks after the ACP stream closes', { timeout: 20_000 }, async () => {
+    const { dir, fixtureFile } = await scenario({ permissionProbe: true })
+    let releasePermission: (() => void) | undefined
+    const permissionReleased = new Promise<void>((resolve) => { releasePermission = resolve })
+    let markPermissionStarted: (() => void) | undefined
+    const permissionStarted = new Promise<void>((resolve) => { markPermissionStarted = resolve })
+    let permissionFinished = false
+    const launched = launchAcpTestAgent({
+      agent: AGENT,
+      cwd: dir,
+      env: { DSH_SNAPSHOT_FILE: fixtureFile },
+      async requestPermission() {
+        markPermissionStarted?.()
+        await permissionReleased
+        permissionFinished = true
+        return { outcome: { outcome: 'cancelled' } }
+      },
+    })
+    await launched.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
+    const { sessionId } = await launched.client.newSession({ cwd: dir, mcpServers: [] })
+    void launched.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }).catch(() => undefined)
+    await permissionStarted
+
+    const childClosed = once(launched.child, 'close')
+    let closeSettled = false
+    const closing = launched.close('SIGKILL').then(() => { closeSettled = true })
+    await childClosed
+    await launched.client.closed
+    expect(closeSettled).toBe(false)
+
+    releasePermission?.()
+    await closing
+    expect(permissionFinished).toBe(true)
+  })
+
  it('includes agent stderr when the ACP connection closes during startup', { timeout: 20_000 }, async () => {
    const { fixtureFile } = await scenario({ failOnBoot: true, stderrNote: 'fake agent requested startup failure' })
    await expect(runScenario(
@@ -48,6 +244,23 @@ describe('runScenario', () => {
    )).rejects.toThrow(/agent stderr:\nfake agent requested startup failure/)
  })

+  it('preserves launch-resolution errors when no child process exists', async () => {
+    const { dir, fixtureFile } = await scenario({})
+    vi.stubEnv('DSH_EXAMPLE_MODE', 'lib')
+    try {
+      await expect(runScenario(
+        { steps: [] },
+        {
+          agent: { ...AGENT, binScript: join(dir, 'outside-src.ts'), libBinScript: undefined },
+          mode: 'replay',
+          fixtureFile,
+        },
+      )).rejects.toThrow(/expected a "\/src\/" segment/)
+    } finally {
+      vi.unstubAllEnvs()
+    }
+  })
+
  it('drives a full turn: initialize (terminal caps), session, prompt, permission stub, harvest', { timeout: 20_000 }, async () => {
    const { fixtureFile } = await scenario({
      permissionProbe: true,
@@ -138,6 +351,39 @@ describe('runScenario', () => {
    )).rejects.toThrow(/expected the prompt to fail/)
  })

+  it('reports scenario and cleanup failures together', { timeout: 20_000 }, async () => {
+    const { fixtureFile } = await scenario({ prompt: 'respond' })
+    const cleanupFailure = new Error('cleanup failed')
+    fsControl.cleanupFailure = cleanupFailure
+
+    const failure = await runScenario(
+      { steps: [...boot, { op: 'promptExpectError', text: 'fine' }] },
+      { agent: AGENT, mode: 'replay', fixtureFile },
+    ).catch((error: unknown): unknown => error)
+
+    expect(failure).toBeInstanceOf(AggregateError)
+    const failures = (failure as AggregateError).errors as unknown[]
+    expect(failures).toHaveLength(2)
+    expect(failures[0]).toBeInstanceOf(Error)
+    expect((failures[0] as Error).message).toMatch(/expected the prompt to fail/)
+    expect(failures[1]).toBe(cleanupFailure)
+  })
+
+  it('reports cleanup failure after an otherwise successful scenario', { timeout: 20_000 }, async () => {
+    const { fixtureFile } = await scenario({})
+    const cleanupFailure = new Error('cleanup failed')
+    fsControl.cleanupFailure = cleanupFailure
+
+    const failure = await runScenario(
+      { steps: boot },
+      { agent: AGENT, mode: 'replay', fixtureFile },
+    ).catch((error: unknown): unknown => error)
+
+    expect(failure).toBeInstanceOf(AggregateError)
+    expect((failure as AggregateError).message).toBe('snapshot cleanup failed')
+    expect((failure as AggregateError).errors as unknown[]).toEqual([cleanupFailure])
+  })
+
  it('newSessionExpectError swallows the rejection, with and without extra dirs', { timeout: 20_000 }, async () => {
    const { fixtureFile } = await scenario({ rejectExtraDirs: true })
    const result = await runScenario(
--- a/packages/support/acp-snapshot/tests/suite.spec.ts
+++ b/packages/support/acp-snapshot/tests/suite.spec.ts
@@ -1,4 +1,4 @@
-import { cpSync, mkdtempSync, readFileSync, writeFileSync } from 'node:fs'
+import { cpSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
 import { rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
@@ -6,7 +6,6 @@ import { fileURLToPath } from 'node:url'
 import { afterAll, describe, expect, it } from 'vitest'
 import { defineAcpSnapshotSuite, type HarvestedLog, type Scenario } from '../src/index.ts'
 import {
-  childFixturePaths,
  fixtureContext,
  formatSystemPromptSnapshot,
  headerChangeCount,
@@ -16,6 +15,7 @@ import {
  normalizedToolSchemas,
  parseToolSchemasSnapshot,
  refreshFixtureReplacements,
+  sessionFixtureNames,
  restorePinnedToolSchemas,
  stabilizeRefreshLog,
  unknownToolCallIds,
@@ -46,7 +46,7 @@ const RECORD_SRC = fileURLToPath(new URL('./fixtures/record-suite', import.meta.
 // Replay pins explicit header classes; recording covers the default fallback.
 const REPLAY_SCENARIOS: Scenario[] = [
  { name: 'pin-turn', hasModelTurn: true, recorded: true, pinsHeader: true, expectedHeaderChanges: 1, headerClass: 'main' },
-  { name: 'plain-turn', hasModelTurn: true, recorded: true, childSessions: 1, headerClass: 'main', configPath: AGENT.configPath },
+  { name: 'plain-turn', hasModelTurn: true, recorded: true, headerClass: 'main', configPath: AGENT.configPath },
  { name: 'no-model', hasModelTurn: false, recorded: false, headerClass: 'main' },
  { name: 'blocked-log', hasModelTurn: false, comparesLog: true, recorded: false, headerClass: 'main' },
  { name: 'authored-error', hasModelTurn: true, recorded: false, overridden: true, headerClass: 'main' },
@@ -54,7 +54,7 @@ const REPLAY_SCENARIOS: Scenario[] = [

 const RECORD_SCENARIOS: Scenario[] = [
  { name: 'rec-pin', hasModelTurn: true, recorded: true, pinsHeader: true },
-  { name: 'rec-child', hasModelTurn: true, recorded: true, childSessions: 1 },
+  { name: 'rec-child', hasModelTurn: true, recorded: true },
  // recorded:false in record mode → registered but skipped (never re-recorded).
  { name: 'rec-skip', hasModelTurn: true, recorded: false, overridden: true },
 ]
@@ -64,7 +64,13 @@ const RECORD_SCENARIOS: Scenario[] = [
 // committed record fixtures/goldens in place.
 const BOOTSTRAP = process.env.ACP_SNAPSHOT_SPEC_BOOTSTRAP === '1'
 const recordDir = BOOTSTRAP ? RECORD_SRC : mkdtempSync(join(tmpdir(), 'acp-snap-record-suite-'))
-if (!BOOTSTRAP) cpSync(RECORD_SRC, recordDir, { recursive: true })
+if (!BOOTSTRAP) {
+  cpSync(RECORD_SRC, recordDir, { recursive: true })
+  // Record mode owns its output inventory: a new scenario has no primary yet,
+  // while a changed child count can leave old numbered fixtures behind.
+  rmSync(join(recordDir, 'rec-pin', 'session.jsonl'))
+  writeFileSync(join(recordDir, 'rec-child', 'session.2.jsonl'), 'stale child\n')
+}
 const refreshDir = mkdtempSync(join(tmpdir(), 'acp-snap-refresh-suite-'))
 cpSync(REPLAY_DIR, refreshDir, { recursive: true })
 staleRefreshFixtures(refreshDir)
@@ -140,6 +146,13 @@ describe('defineAcpSnapshotSuite: refresh write-back', () => {
  })
 })

+describe('defineAcpSnapshotSuite: record inventory write-back', () => {
+  it('creates a missing primary fixture and prunes stale child fixtures', () => {
+    expect(readFileSync(join(recordDir, 'rec-pin', 'session.jsonl'), 'utf8')).toContain('"type":"session"')
+    expect(() => readFileSync(join(recordDir, 'rec-child', 'session.2.jsonl'), 'utf8')).toThrow()
+  })
+})
+
 describe('defineAcpSnapshotSuite: registration contract', () => {
  it("throws when a scenario's header class has no pinning scenario", () => {
    expect(() => {
@@ -179,13 +192,41 @@ describe('defineAcpSnapshotSuite: registration contract', () => {
  })
 })

-describe('childFixturePaths', () => {
-  it('yields one sibling path per child, 1-based', () => {
-    expect(childFixturePaths('/snap/s', 2)).toEqual(['/snap/s/session.1.jsonl', '/snap/s/session.2.jsonl'])
+describe('sessionFixtureNames', () => {
+  it('orders the primary and contiguous child fixtures while ignoring other files', () => {
+    expect(sessionFixtureNames([
+      'stdout.golden.jsonl',
+      'session.2.jsonl',
+      'session.jsonl',
+      'session.1.jsonl',
+      'input.json',
+    ])).toEqual(['session.jsonl', 'session.1.jsonl', 'session.2.jsonl'])
  })

-  it('yields nothing for a single-session scenario', () => {
-    expect(childFixturePaths('/snap/s', 0)).toEqual([])
+  it('accepts a primary-only scenario', () => {
+    expect(sessionFixtureNames(['session.jsonl'])).toEqual(['session.jsonl'])
+  })
+
+  it('rejects a directory without the primary fixture', () => {
+    expect(() => sessionFixtureNames(['session.1.jsonl'])).toThrow('missing session.jsonl')
+  })
+
+  it('rejects gapped child fixtures', () => {
+    expect(() => sessionFixtureNames(['session.jsonl', 'session.2.jsonl']))
+      .toThrow('expected session.1.jsonl, found session.2.jsonl')
+  })
+
+  it.each(['session.0.jsonl', 'session.child.jsonl', 'session.01.jsonl'])(
+    'rejects invalid child fixture name %s',
+    (name) => {
+      expect(() => sessionFixtureNames(['session.jsonl', name]))
+        .toThrow(`invalid child session fixture name: ${name}`)
+    },
+  )
+
+  it('rejects duplicate child indexes', () => {
+    expect(() => sessionFixtureNames(['session.jsonl', 'session.1.jsonl', 'session.1.jsonl']))
+      .toThrow('expected session.2.jsonl, found session.1.jsonl')
  })
 })