test(snapshot): await cancelled turn durability

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

@@ -21,6 +21,7 @@ import { existsSync } from 'node:fs'
 import { createHash } from 'node:crypto'
 import { tmpdir } from 'node:os'
 import { basename, dirname, join, delimiter } from 'node:path'
+import { setTimeout as delay } from 'node:timers/promises'
 import {
   ClientSideConnection,
   PROTOCOL_VERSION,
@@ -34,6 +35,9 @@ import { launchAcpTestAgent, type AgentUnderTest, type LaunchedAcpTestAgent } fr
 
 export type { AgentUnderTest } from './launcher.ts'
 
+const DEFAULT_TURN_END_TIMEOUT_MS = 10_000
+const TURN_END_POLL_INTERVAL_MS = 10
+
 /**
  * One step of a scenario's deterministic input script (`input.json`). The
  * harness interprets these in order. `newSession` captures the server-issued
@@ -46,6 +50,8 @@ export type { AgentUnderTest } from './launcher.ts'
  * step open for a terminal tool update that may follow the prompt response.
  * `promptAndWaitForAgentMessage` arms an exact text-chunk waiter before sending
  * the prompt, then keeps the application live until that later update arrives.
+ * `waitForTurnEnd` holds the subprocess open until the selected session's latest
+ * complete raw-JSONL turn boundary is `turn/end`; its timeout defaults to 10s.
  */
 export type InputStep =
   | { op: 'initialize'; terminalOutput?: boolean }
@@ -60,6 +66,7 @@ export type InputStep =
     afterUpdate?: 'agent_message_chunk' | 'tool_call'
     waitForToolCallUpdate?: string
   }
+  | { op: 'waitForTurnEnd'; timeoutMs?: number }
   | { op: 'cancel' }
   | { op: 'setMode'; modeId: string }
   | { op: 'setModeExpectError'; modeId: string }
@@ -295,7 +302,15 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
     const { client } = active
 
     for (const step of input.steps) {
-      await runStep(client, step, cwd, match => active.waitForUpdate(match), () => sessionId, (id) => { sessionId = id })
+      await runStep(
+        client,
+        step,
+        cwd,
+        match => active.waitForUpdate(match),
+        () => sessionId,
+        (id) => { sessionId = id },
+        (id, timeoutMs) => waitForPersistedTurnEnd(sessionsRoot, id, 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 —
       // fail the run HERE, as a harness error, rather than hoping the agent's
@@ -365,6 +380,7 @@ async function runStep(
   waitForUpdate: (match: (u: SessionNotification['update']) => boolean) => Promise<SessionNotification['update']>,
   getSessionId: () => string | undefined,
   setSessionId: (id: string) => void,
+  waitForTurnEnd: (sessionId: string, timeoutMs?: number) => Promise<void>,
 ): Promise<void> {
   switch (step.op) {
     case 'initialize':
@@ -438,6 +454,12 @@ async function runStep(
       if (toolCallUpdateDone !== undefined) await toolCallUpdateDone
       return
     }
+    case 'waitForTurnEnd': {
+      const sessionId = getSessionId()
+      if (sessionId === undefined) throw new Error('snapshot-harness: waitForTurnEnd before newSession')
+      await waitForTurnEnd(sessionId, step.timeoutMs)
+      return
+    }
     case 'cancel': {
       const sessionId = getSessionId()
       if (sessionId === undefined) throw new Error('snapshot-harness: cancel before newSession')
@@ -485,6 +507,35 @@ async function runStep(
   }
 }
 
+/**
+ * Wait until the raw JSONL backend exposes one complete closing turn boundary.
+ * The ACP cancel notification settles its prompt before the agent necessarily
+ * reaches quiescence, so cancellation snapshots use this external boundary to
+ * keep subprocess disposal from changing an `aborted` turn into `disposed`.
+ */
+async function waitForPersistedTurnEnd(
+  root: string,
+  sessionId: string,
+  timeoutMs = DEFAULT_TURN_END_TIMEOUT_MS,
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs
+  while (true) {
+    const log = (await harvestSessionLogs(root)).find(candidate => candidate.id === sessionId)
+    if (log !== undefined && latestTurnIsClosed(log.content)) return
+    if (Date.now() >= deadline) {
+      throw new Error(`snapshot-harness: session "${sessionId}" did not persist turn/end within ${timeoutMs}ms`)
+    }
+    await delay(TURN_END_POLL_INTERVAL_MS)
+  }
+}
+
+/** Return whether the last complete raw-JSONL turn boundary closes its turn. */
+function latestTurnIsClosed(content: string): boolean {
+  const complete = content.slice(0, content.lastIndexOf('\n') + 1)
+  return complete.lastIndexOf('\n{"type":"turn/end",')
+    > complete.lastIndexOf('\n{"type":"turn/start",')
+}
+
 /**
  * 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