test(acp-snapshot): wait for the durable goal pause before disposal

The goal scenario asserted a paused revision-2 goal, but the pause is
appended only after cancellation reaches idle — after turn/end. Under
parallel snapshot files the subprocess could dispose before the pause
record persisted, folding the log to an active revision-1 goal.

Add a waitForEventAfterTurnEnd input step (the turn-end/title waiters'
shape, parameterized by event type) and use it in the goal scenario to
hold the subprocess open until the goal-state record lands.

examples/acp-agent/tests/goal-snapshots/goal-session/input.json

@@ -2,13 +2,10 @@
   "steps": [
     { "op": "initialize" },
     { "op": "newSession" },
-    {
-      "op": "promptAndWaitForAgentMessage",
-      "text": "Create a durable two-round goal for the ACP snapshot, inspect it, then report readiness.",
-      "waitForText": "GOAL ROUND ONE"
-    },
+    { "op": "promptAndWaitForAgentMessage", "text": "Create a durable two-round goal for the ACP snapshot, inspect it, then report readiness.", "waitForText": "GOAL ROUND ONE" },
     { "op": "waitForTurnStart", "minimumTurn": 3 },
     { "op": "cancel", "waitForFile": { "path": ".dsh-snapshot-goal-cancel-ready" } },
-    { "op": "waitForTurnEnd" }
+    { "op": "waitForTurnEnd" },
+    { "op": "waitForEventAfterTurnEnd", "type": "user/message" }
   ]
 }

packages/support/acp-snapshot/src/harness.ts

@@ -43,16 +43,22 @@ const WAIT_POLL_INTERVAL_MS = 10
  * reference, since a committed file cannot know the id in advance.
  *
  * `promptAndCancel` starts a prompt without awaiting completion, waits for a
- * readiness condition, then cancels and awaits completion. `waitForFile`
- * observes a cwd-relative marker; the default observes the durable turn start.
+ * readiness condition, then cancels and awaits completion. Its optional
+ * `waitForFile` observes a cwd-relative marker; otherwise it waits for the
+ * durable turn start. The standalone `waitForFile` holds the next script step
+ * behind the same marker.
  * `promptAndWaitForAgentMessage` arms an exact text-chunk waiter before sending
  * the prompt, then keeps the application live until that later update arrives.
  * `waitForTurnStart` waits for an open durable turn, optionally at or beyond a
  * specified turn number. `waitForTurnEnd` holds the subprocess open until the
  * selected session's latest complete raw-JSONL turn boundary is `turn/end`.
+ * `waitForSubagentTurnEnd` waits until one background child has persisted a
+ * closed model-work turn after its own descriptor; child progress has no ACP
+ * update to wait on.
  * `waitForTitleAfterTurnEnd` additionally waits for a later durable title.
- * `waitForSubagentTurnEnd` applies the same work-turn boundary to one
- * background child, whose progress has no ACP update to wait on.
+ * `waitForEventAfterTurnEnd` waits until a complete record of the given event
+ * type follows the latest closed turn — for scenarios whose asserted state
+ * (e.g. a goal pause) is appended only after cancellation reaches idle.
  * A standalone `cancel` may also wait for a cwd-relative readiness marker.
  * All wait timeouts default to 10s.
  */
@@ -73,6 +79,7 @@ export type InputStep =
   | { op: 'waitForTurnEnd'; timeoutMs?: number }
   | { op: 'waitForSubagentTurnEnd'; child?: number; timeoutMs?: number }
   | { op: 'waitForTitleAfterTurnEnd'; timeoutMs?: number }
+  | { op: 'waitForEventAfterTurnEnd'; type: string; timeoutMs?: number }
   | { op: 'cancel'; waitForFile?: { path: string; timeoutMs?: number } }
 
 /** A scenario's `input.json`: an ordered list of input steps. */
@@ -296,6 +303,7 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
         (id, timeoutMs) => waitForPersistedTurnEnd(sessionsRoot, id, timeoutMs),
         (child, timeoutMs) => waitForPersistedChildTurnEnd(sessionsRoot, child, timeoutMs),
         (id, timeoutMs) => waitForPersistedTitleAfterTurnEnd(sessionsRoot, id, timeoutMs),
+        (id, type, timeoutMs) => waitForPersistedEventAfterTurnEnd(sessionsRoot, id, type, timeoutMs),
       )
       // 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 —
@@ -371,6 +379,7 @@ async function runStep(
   waitForTurnEnd: (sessionId: string, timeoutMs?: number) => Promise<void>,
   waitForChildTurnEnd: (child: number, timeoutMs?: number) => Promise<void>,
   waitForTitleAfterTurnEnd: (sessionId: string, timeoutMs?: number) => Promise<void>,
+  waitForEventAfterTurnEnd: (sessionId: string, type: string, timeoutMs?: number) => Promise<void>,
 ): Promise<void> {
   switch (step.op) {
     case 'initialize':
@@ -460,6 +469,12 @@ async function runStep(
       await waitForTitleAfterTurnEnd(sessionId, step.timeoutMs)
       return
     }
+    case 'waitForEventAfterTurnEnd': {
+      const sessionId = getSessionId()
+      if (sessionId === undefined) throw new Error('snapshot-harness: waitForEventAfterTurnEnd before newSession')
+      await waitForEventAfterTurnEnd(sessionId, step.type, step.timeoutMs)
+      return
+    }
     case 'waitForTurnStart': {
       const sessionId = getSessionId()
       if (sessionId === undefined) throw new Error('snapshot-harness: waitForTurnStart before newSession')
@@ -576,6 +591,21 @@ async function waitForPersistedTitleAfterTurnEnd(
   }, { interval: WAIT_POLL_INTERVAL_MS, timeout: timeoutMs })
 }
 
+/** Wait until a complete record of `type` follows the latest closed turn. */
+async function waitForPersistedEventAfterTurnEnd(
+  root: string,
+  sessionId: string,
+  type: string,
+  timeoutMs = DEFAULT_WAIT_TIMEOUT_MS,
+): Promise<void> {
+  await vi.waitFor(async () => {
+    const log = (await harvestSessionLogs(root)).find(candidate => candidate.id === sessionId)
+    if (log === undefined || !latestEventFollowsTurnEnd(log.content, type)) {
+      throw new Error(`snapshot-harness: session "${sessionId}" did not persist ${type} after turn/end within ${timeoutMs}ms`)
+    }
+  }, { interval: WAIT_POLL_INTERVAL_MS, timeout: timeoutMs })
+}
+
 /** Wait for a cwd-relative marker proving an external action reached readiness. */
 async function waitForWorkspaceFile(
   cwd: string,
@@ -604,6 +634,13 @@ function latestTitleFollowsTurnEnd(content: string): boolean {
   return turnEnd >= 0 && complete.lastIndexOf('\n{"type":"session/title",') > turnEnd
 }
 
+/** Return whether a complete record of `type` occurs after the last complete turn end. */
+function latestEventFollowsTurnEnd(content: string, type: string): boolean {
+  const complete = content.slice(0, content.lastIndexOf('\n') + 1)
+  const turnEnd = complete.lastIndexOf('\n{"type":"turn/end",')
+  return turnEnd >= 0 && complete.lastIndexOf(`\n{"type":"${type}",`) > turnEnd
+}
+
 /** Return the latest open turn number, validating the persisted boundary record. */
 function latestOpenTurn(content: string): number | undefined {
   const complete = content.slice(0, content.lastIndexOf('\n') + 1)

vitest.snapshot.config.ts

@@ -55,16 +55,17 @@ export default defineConfig({
       'packages/sdk/*/tests/**/*.snapshot.ts',
       'packages/ui/tui/tests/**/*.snapshot.ts',
     ],
-    // Each test boots a subprocess; give it room. Replay scenarios are
-    // read-only (unique temp dir and fixture set per subprocess), so replay
-    // runs the snapshot files in parallel and bounds in-file concurrency with
-    // the environment knob (value 1 restores serial replay on constrained
-    // machines). Record and refresh stay fully serial: record spends real API
-    // quota per scenario, and refresh write-back harvests volatile values from
-    // fixtures already on disk, so concurrent writers would corrupt goldens.
+    // Replay never writes committed outputs and every scenario owns its
+    // mutable runtime state (the subprocess suites use a unique temp dir and
+    // fixture set per scenario), so replay runs the snapshot files in
+    // parallel and bounds in-file concurrency with the environment knob
+    // (value 1 restores fully serial replay on constrained machines). Record
+    // and refresh stay serial: record spends real API quota per scenario, and
+    // refresh write-back harvests volatile values from fixtures already on
+    // disk, so concurrent writers would corrupt goldens.
     testTimeout: 120_000,
     hookTimeout: 30_000,
-    fileParallelism: (process.env.DSH_SNAPSHOT || 'replay') === 'replay',
+    fileParallelism: (process.env.DSH_SNAPSHOT || 'replay') === 'replay' && snapshotMaxConcurrency > 1,
     maxConcurrency: snapshotMaxConcurrency,
   },
 })