fix inbox lifecycle downstream contracts

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

@@ -50,6 +50,7 @@ const WAIT_POLL_INTERVAL_MS = 10
  * `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`.
+ * `waitForInboxMessage` waits for inserted inbox text containing a scenario marker.
  * `waitForTitleAfterTurnEnd` additionally waits for a later durable title.
  * A standalone `cancel` may also wait for a cwd-relative readiness marker.
  * All wait timeouts default to 10s.
@@ -68,6 +69,7 @@ export type InputStep =
   }
   | { op: 'waitForTurnStart'; minimumTurn?: number; timeoutMs?: number }
   | { op: 'waitForTurnEnd'; timeoutMs?: number }
+  | { op: 'waitForInboxMessage'; text: string; timeoutMs?: number }
   | { op: 'waitForTitleAfterTurnEnd'; timeoutMs?: number }
   | { op: 'cancel'; waitForFile?: { path: string; timeoutMs?: number } }
 
@@ -290,6 +292,7 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
         (id) => { sessionId = id },
         (id, timeoutMs, minimumTurn) => waitForPersistedTurnStart(sessionsRoot, id, timeoutMs, minimumTurn),
         (id, timeoutMs) => waitForPersistedTurnEnd(sessionsRoot, id, timeoutMs),
+        (id, text, timeoutMs) => waitForPersistedInboxMessage(sessionsRoot, id, text, timeoutMs),
         (id, timeoutMs) => waitForPersistedTitleAfterTurnEnd(sessionsRoot, id, timeoutMs),
       )
       // A permission exchange happens while a step's request is in flight, so
@@ -364,6 +367,7 @@ async function runStep(
   setSessionId: (id: string) => void,
   waitForTurnStart: (sessionId: string, timeoutMs?: number, minimumTurn?: number) => Promise<void>,
   waitForTurnEnd: (sessionId: string, timeoutMs?: number) => Promise<void>,
+  waitForInboxMessage: (sessionId: string, text: string, timeoutMs?: number) => Promise<void>,
   waitForTitleAfterTurnEnd: (sessionId: string, timeoutMs?: number) => Promise<void>,
 ): Promise<void> {
   switch (step.op) {
@@ -442,6 +446,12 @@ async function runStep(
       await waitForTurnEnd(sessionId, step.timeoutMs)
       return
     }
+    case 'waitForInboxMessage': {
+      const sessionId = getSessionId()
+      if (sessionId === undefined) throw new Error('snapshot-harness: waitForInboxMessage before newSession')
+      await waitForInboxMessage(sessionId, step.text, step.timeoutMs)
+      return
+    }
     case 'waitForTitleAfterTurnEnd': {
       const sessionId = getSessionId()
       if (sessionId === undefined) throw new Error('snapshot-harness: waitForTitleAfterTurnEnd before newSession')
@@ -515,6 +525,29 @@ async function waitForPersistedTurnEnd(
   }, { interval: WAIT_POLL_INTERVAL_MS, timeout: timeoutMs })
 }
 
+/** Wait until an inserted inbox message contains scenario-owned text. */
+async function waitForPersistedInboxMessage(
+  root: string,
+  sessionId: string,
+  text: string,
+  timeoutMs = DEFAULT_WAIT_TIMEOUT_MS,
+): Promise<void> {
+  await vi.waitFor(async () => {
+    const log = (await harvestSessionLogs(root)).find(candidate => candidate.id === sessionId)
+    const matched = log?.content.split('\n').some((line) => {
+      if (line.length === 0) return false
+      const record = JSON.parse(line) as {
+        type?: unknown
+        data?: { inserted?: Array<{ content?: Array<{ type?: unknown; text?: unknown }> }> }
+      }
+      return record.type === 'agent/inbox/spliced' && record.data?.inserted?.some(message =>
+        message.content?.some(block => block.type === 'text'
+          && typeof block.text === 'string' && block.text.includes(text))) === true
+    }) ?? false
+    if (!matched) throw new Error(`snapshot-harness: session "${sessionId}" did not persist expected inbox message within ${timeoutMs}ms`)
+  }, { interval: WAIT_POLL_INTERVAL_MS, timeout: timeoutMs })
+}
+
 /** Wait until a complete provider or fallback title record follows the latest closed turn. */
 async function waitForPersistedTitleAfterTurnEnd(
   root: string,

packages/support/acp-snapshot/tests/harness.spec.ts

@@ -584,6 +584,55 @@ describe('runScenario', () => {
     expect(result.sessionLogs[0]?.content).toContain('"type":"turn/end"')
   })
 
+  it('waitForInboxMessage holds the app through a matching durable insertion', { timeout: 20_000 }, async () => {
+    const { fixtureFile } = await scenario({
+      prompt: 'hang-until-cancel',
+      persistLogsOnCancel: true,
+      logs: [{
+        file: 'project/main/session.jsonl',
+        lines: [
+          { type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
+          {
+            type: 'agent/inbox/spliced',
+            seq: 0,
+            time: 2,
+            data: {
+              target: 'next-turn',
+              start: 0,
+              inserted: [{ role: 'user', content: [{ type: 'text', text: 'durable marker' }] }],
+            },
+          },
+        ],
+      }],
+    })
+    const result = await runScenario(
+      { steps: [...boot, { op: 'promptAndCancel', text: 'hang' }, { op: 'waitForInboxMessage', text: 'marker' }] },
+      { agent: AGENT, mode: 'replay', fixtureFile },
+    )
+    expect(result.sessionLogs[0]?.content).toContain('durable marker')
+  })
+
+  it('waitForInboxMessage times out when the session log or matching insertion is absent', { timeout: 20_000 }, async () => {
+    const absent = await scenario({ prompt: 'hang-until-cancel', persistLogsOnCancel: true })
+    await expect(runScenario(
+      { steps: [...boot, { op: 'promptAndCancel', text: 'hang' }, { op: 'waitForInboxMessage', text: 'missing', timeoutMs: 20 }] },
+      { agent: AGENT, mode: 'replay', fixtureFile: absent.fixtureFile },
+    )).rejects.toThrow(/did not persist expected inbox message within 20ms/)
+
+    const unmatched = await scenario({
+      prompt: 'hang-until-cancel',
+      persistLogsOnCancel: true,
+      logs: [{
+        file: 'project/main/session.jsonl',
+        lines: [{ type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 }],
+      }],
+    })
+    await expect(runScenario(
+      { steps: [...boot, { op: 'promptAndCancel', text: 'hang' }, { op: 'waitForInboxMessage', text: 'missing', timeoutMs: 20 }] },
+      { agent: AGENT, mode: 'replay', fixtureFile: unmatched.fixtureFile },
+    )).rejects.toThrow(/did not persist expected inbox message within 20ms/)
+  })
+
   it('waitForTitleAfterTurnEnd holds the app through a standalone durable title', { timeout: 20_000 }, async () => {
     const { fixtureFile } = await scenario({
       prompt: 'hang-until-cancel',
@@ -872,6 +921,7 @@ describe('runScenario', () => {
     [{ op: 'promptAndCancel', text: 'x' }, /promptAndCancel before newSession/],
     [{ op: 'waitForTurnStart' }, /waitForTurnStart before newSession/],
     [{ op: 'waitForTurnEnd' }, /waitForTurnEnd before newSession/],
+    [{ op: 'waitForInboxMessage', text: 'marker' }, /waitForInboxMessage before newSession/],
     [{ op: 'waitForTitleAfterTurnEnd' }, /waitForTitleAfterTurnEnd before newSession/],
     [{ op: 'cancel' }, /cancel before newSession/],
   ] as [InputStep, RegExp][])('rejects %j before newSession', { timeout: 20_000 }, async (step, message) => {

packages/support/llm-replay/src/index.ts

@@ -173,9 +173,10 @@ export function parseSessionHeader(text: string): { id: string; createdAt: numbe
 /**
  * Reconstruct the per-`stream()` replay script from a recorded session log.
  *
- * Groups `assistant/chunk` events by turn and step. Every group must end in a
- * `finish`; a missing terminator means the live stream threw, so derivation
- * rejects and the scenario must provide an explicit override.
+ * Splits `assistant/chunk` events at every `finish`, using turn and step changes
+ * to detect an unterminated prior call. A missing terminator means the live
+ * stream threw, so derivation rejects and the scenario must provide an explicit
+ * override. Multiple calls may share one turn and step when the loop retries.
  * @param events - the recorded session's events; only `assistant/chunk` is consulted.
  * @returns one `chunks` entry per recorded model call, in call order.
  */
@@ -197,14 +198,16 @@ export function deriveReplayScript(events: SessionEvent[]): ReplayEntry[] {
     if (event.type !== 'assistant/chunk') continue
     const { turn, step, chunk } = event.data
     const key = `${turn}/${step}`
-    if (key !== currentKey) {
-      // A new (turn, step) — i.e. a new stream() call. Close the previous one
-      // (skip the initial empty buffer before any chunk has been seen).
+    if (current.length > 0 && key !== currentKey) {
       close(currentKey, current)
-      currentKey = key
+    }
+    if (current.length === 0) currentKey = key
+    current.push(chunk)
+    if (chunk.type === 'finish') {
+      close(currentKey, current)
+      currentKey = undefined
       current = []
     }
-    current.push(chunk)
   }
   close(currentKey, current)
   return script

packages/support/llm-replay/tests/llm-replay.spec.ts

@@ -106,11 +106,27 @@ describe('parseSessionLog', () => {
 })
 
 describe('deriveReplayScript', () => {
-  it('groups assistant/chunk by (turn, step) into one entry per stream() call', () => {
+  it('groups one finished assistant/chunk stream into one replay entry', () => {
     const events: SessionEvent[] = TEXT_CHUNKS.map((c, i) => chunkEvent(i + 1, 1, 1, c))
     expect(deriveReplayScript(events)).toEqual([{ kind: 'chunks', chunks: TEXT_CHUNKS }])
   })
 
+  it('separates retry calls that share one turn and step at their finish chunks', () => {
+    const failed: StreamChunk[] = [
+      { type: 'usage', usage: { inputTokens: 0, outputTokens: 0 } },
+      { type: 'finish', reason: { kind: 'error', failure: { message: 'empty', code: 'EMPTY_RESPONSE' } } },
+    ]
+    let seq = 1
+    const events: SessionEvent[] = [
+      ...failed.map(chunk => chunkEvent(seq++, 1, 1, chunk)),
+      ...TEXT_CHUNKS.map(chunk => chunkEvent(seq++, 1, 1, chunk)),
+    ]
+    expect(deriveReplayScript(events)).toEqual([
+      { kind: 'chunks', chunks: failed },
+      { kind: 'chunks', chunks: TEXT_CHUNKS },
+    ])
+  })
+
   it('produces one entry per distinct (turn, step), in log order', () => {
     const callA = TEXT_CHUNKS
     const callB: StreamChunk[] = [
@@ -177,6 +193,14 @@ describe('deriveReplayScript', () => {
     ]
     expect(() => deriveReplayScript(events)).toThrow(/2\/3/)
   })
+
+  it('rejects an unfinished call before consuming chunks from a new step', () => {
+    const events: SessionEvent[] = [
+      chunkEvent(1, 1, 1, { type: 'block-start', index: 0, blockType: 'text' }),
+      chunkEvent(2, 1, 2, { type: 'finish', reason: { kind: 'stop' } }),
+    ]
+    expect(() => deriveReplayScript(events)).toThrow(/model call 1\/1 ended without a finish chunk/)
+  })
 })
 
 describe('loadReplayScript', () => {