Merge remote-tracking branch 'origin/master' into codex/pr335-merge-master-20260719

# Conflicts:
#	packages/support/acp-snapshot/README.md
#	packages/support/acp-snapshot/src/harness.ts
#	packages/support/acp-snapshot/src/suite.ts
#	packages/support/acp-snapshot/tests/harness.spec.ts
#	packages/support/acp-snapshot/tests/suite.spec.ts

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[]
 }
@@ -174,93 +169,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 = snapshotSpillRoot()
-  // 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' } })
@@ -278,10 +227,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
@@ -290,35 +241,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. */
@@ -326,7 +299,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> {
@@ -343,8 +316,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: [],
@@ -364,8 +339,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 */ })
@@ -374,8 +351,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 })
@@ -411,16 +393,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
@@ -461,8 +433,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

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,

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
+}

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'
@@ -75,14 +75,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.
@@ -162,14 +154,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))
 }
 
 /**
@@ -487,7 +505,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,
@@ -496,7 +519,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.
@@ -524,7 +547,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')))
           : []
@@ -533,18 +555,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)
@@ -575,7 +616,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'))
@@ -654,9 +695,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, pinsNativeWindowsStdout } 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, pinsNativeWindowsStdout } 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)
@@ -671,11 +712,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()
       }
     })
 
@@ -727,10 +764,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`)