Merge pull request #5868 from deepseek-harness/fix/snapshot-title-event-wait-diagnostics

test(ci): retain snapshot wait and coverage worker-exit diagnostics

packages/test-support/session-snapshot/README.i18n.yaml

@@ -30,8 +30,8 @@
   en: 64666ea2894acc4a
   zh: 0e78df32bcfd3633
 /deepseek-ai-dsh-session-snapshot/use-this-package/what-can-go-wrong:
-  en: f94db51dc8c30ccd
-  zh: fc040ad34f99e7ec
+  en: 0b03592947ca43d6
+  zh: b14a64238c4b5f80
 /deepseek-ai-dsh-session-snapshot/understand-the-implementation:
   en: 18ff76b440fe40da
   zh: 4dbf2eb5f9ee98a2

packages/test-support/session-snapshot/README.md

@@ -92,7 +92,7 @@ A scenario requiring a non-Windows host declares `posixOnly`, which skips its ru
 
 ### What can go wrong
 
-- **A durable wait times out** — `waitForTurnStart`, `waitForTurnEnd`, `waitForSubagentTurnEnd`, `waitForGoalPhase`, and `waitForInboxMessage` identify the session or child and deadline even when the first log harvest exceeds that deadline. Child waits also name the requested turn. All retain the underlying failure as the error cause. `waitForTurnStart` rejects malformed turn records immediately.
+- **A durable wait times out** — `waitForTurnStart`, `waitForTurnEnd`, `waitForSubagentTurnEnd`, `waitForGoalPhase`, `waitForInboxMessage`, `waitForTitleAfterTurnEnd`, and `waitForEventAfterTurnEnd` identify the session or child and deadline even when the first log harvest exceeds that deadline. Child waits also name the requested turn. All retain the underlying failure as the error cause. `waitForTurnStart` rejects malformed turn records immediately.
 - **A fixture guard rejects the committed files** — orphan scenario dirs, missing files, multiple pins for one header class, duplicate sidecar content, unscrubbed prompt text or tool schemas, a `request/header` with no preceding `system/message`, and malformed pinning headers all fail the suite before comparisons run.
 - **The session harvest needs raw JSONL mode** — snapshot configs set the JSONL backend's `compression: 'none'`; compressed JSONL has no snapshot-harvest path.
 - **Built mode needs current artifacts** — run `pnpm run build` before selecting `DSH_EXAMPLE_MODE=lib`; source mode remains the zero-build path.

packages/test-support/session-snapshot/README.zh.md

@@ -92,7 +92,7 @@ Headless/ACP 与 SDK 适配器在规范化之前，将原始目录中的子创
 
 ### 可能出什么问题
 
-- **持久记录等待超时**——即使首次日志收集就超过期限，`waitForTurnStart`、`waitForTurnEnd`、`waitForSubagentTurnEnd`、`waitForGoalPhase` 和 `waitForInboxMessage` 也会指出会话或子会话及等待期限。子会话等待还会指出目标轮次。这些等待都通过错误的 cause 保留底层失败。`waitForTurnStart` 会立即拒绝格式错误的轮次记录。
+- **持久记录等待超时**——即使首次日志收集就超过期限，`waitForTurnStart`、`waitForTurnEnd`、`waitForSubagentTurnEnd`、`waitForGoalPhase`、`waitForInboxMessage`、`waitForTitleAfterTurnEnd` 和 `waitForEventAfterTurnEnd` 也会指出会话或子会话及等待期限。子会话等待还会指出目标轮次。这些等待都通过错误的 cause 保留底层失败。`waitForTurnStart` 会立即拒绝格式错误的轮次记录。
 - **fixture 保护拒绝已提交文件**——遗留场景目录、缺失文件、一个 header 类别包含多个 pin、重复的伴随文件内容、未擦除的提示文本或工具 schema、没有前置 `system/message` 的 `request/header`，以及格式错误的 pin header 都会在比较运行前使套件失败。
 - **会话收集需要原始 JSONL mode**——快照配置使用 JSONL 后端的 `compression: 'none'`；压缩 JSONL 没有快照收集路径。
 - **构建 mode 需要当前产物**——选择 `DSH_EXAMPLE_MODE=lib` 前先运行 `pnpm run build`；源 mode 仍是零构建路径。

packages/test-support/session-snapshot/src/harness.ts

@@ -68,14 +68,13 @@ const WAIT_POLL_INTERVAL_MS = 10
  * `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.
- * Timeouts in `waitForTurnStart`, `waitForTurnEnd`, `waitForSubagentTurnEnd`,
- * `waitForGoalPhase`, and `waitForInboxMessage` identify the session or child
- * and deadline even if the first log read is still pending; they retain the
- * underlying failure as cause. Child waits also name the requested turn.
  * `waitForTitleAfterTurnEnd` additionally waits for a later durable title.
  * `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.
+ * Session-log wait timeouts identify the session or child and deadline even
+ * if the first log read is still pending; they retain the underlying failure
+ * as cause. Child waits also name the requested turn.
  * A standalone `cancel` may also wait for a cwd-relative readiness marker.
  * All wait timeouts default to 10s.
  */
@@ -712,12 +711,15 @@ async function waitForPersistedTitleAfterTurnEnd(
   sessionId: 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 || !latestTitleFollowsTurnEnd(log.content)) {
-      throw new Error(`snapshot-harness: session "${sessionId}" did not persist session/title after turn/end within ${timeoutMs}ms`)
-    }
-  }, { interval: WAIT_POLL_INTERVAL_MS, timeout: timeoutMs })
+  const message = `snapshot-harness: session "${sessionId}" did not persist session/title after turn/end within ${timeoutMs}ms`
+  try {
+    await vi.waitFor(async () => {
+      const log = (await harvestSessionLogs(root)).find(candidate => candidate.id === sessionId)
+      if (log === undefined || !latestTitleFollowsTurnEnd(log.content)) throw new Error(message)
+    }, { interval: WAIT_POLL_INTERVAL_MS, timeout: timeoutMs })
+  } catch (cause) {
+    throw new Error(message, { cause })
+  }
 }
 
 /** Wait until a complete record of `type` follows the latest closed turn. */
@@ -727,12 +729,15 @@ async function waitForPersistedEventAfterTurnEnd(
   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 })
+  const message = `snapshot-harness: session "${sessionId}" did not persist ${type} after turn/end within ${timeoutMs}ms`
+  try {
+    await vi.waitFor(async () => {
+      const log = (await harvestSessionLogs(root)).find(candidate => candidate.id === sessionId)
+      if (log === undefined || !latestEventFollowsTurnEnd(log.content, type)) throw new Error(message)
+    }, { interval: WAIT_POLL_INTERVAL_MS, timeout: timeoutMs })
+  } catch (cause) {
+    throw new Error(message, { cause })
+  }
 }
 
 /** Wait for a cwd-relative marker proving an external action reached readiness. */

packages/test-support/session-snapshot/tests/harness.spec.ts

@@ -92,8 +92,6 @@ function isolateDiagnosticTimeout(onTestFinished: TestContext['onTestFinished'])
 }
 
 const boot: InputStep[] = [{ op: 'initialize' }, { op: 'newSession' }]
-// A Windows coverage shard can spend more than 20ms harvesting logs before vi.waitFor records the diagnostic error.
-const titleDiagnosticTimeoutMs = process.platform === 'win32' ? 5_000 : 20
 
 it('keeps scenario-owned snapshot spill root length stable across platforms', () => {
   const fixtureFile = '/fixtures/scenario/session.jsonl'
@@ -1009,6 +1007,10 @@ describe('runScenario', () => {
       expected: 'did not persist expected inbox message within 20ms' },
     { label: 'child', step: { op: 'waitForSubagentTurnEnd', child: 2, timeoutMs: 20 },
       expected: 'subagent child #2 did not persist closed turn 1 within 20ms' },
+    { label: 'title', step: { op: 'waitForTitleAfterTurnEnd', timeoutMs: 20 },
+      expected: 'did not persist session/title after turn/end within 20ms' },
+    { label: 'event', step: { op: 'waitForEventAfterTurnEnd', type: 'user/message', timeoutMs: 20 },
+      expected: 'did not persist user/message after turn/end within 20ms' },
   ] satisfies { label: string; step: InputStep; expected: string }[])
   ('identifies the $label wait when its first log harvest outlasts the deadline', async ({ step, expected }) => {
     const { fixtureFile } = await scenario({})
@@ -1130,7 +1132,8 @@ describe('runScenario', () => {
     )).rejects.toThrow(/subagent child #2 did not persist closed turn 1 within 20ms/)
   })
 
-  it('waitForTitleAfterTurnEnd times out when the title precedes the boundary', { timeout: 20_000 }, async () => {
+  it('waitForTitleAfterTurnEnd times out when the title precedes the boundary', { timeout: 20_000 }, async ({ onTestFinished }) => {
+    isolateDiagnosticTimeout(onTestFinished)
     const { fixtureFile } = await scenario({
       prompt: 'hang-until-cancel',
       persistLogsOnCancel: true,
@@ -1148,11 +1151,11 @@ describe('runScenario', () => {
         steps: [
           ...boot,
           { op: 'promptAndCancel', text: 'hang' },
-          { op: 'waitForTitleAfterTurnEnd', timeoutMs: titleDiagnosticTimeoutMs },
+          { op: 'waitForTitleAfterTurnEnd', timeoutMs: 20 },
         ],
       },
       { agent: AGENT, mode: 'replay', fixtureFile },
-    )).rejects.toThrow(new RegExp(`did not persist session/title after turn/end within ${titleDiagnosticTimeoutMs}ms`))
+    )).rejects.toThrow(/did not persist session\/title after turn\/end within 20ms/)
   })
 
   it('waitForEventAfterTurnEnd holds the app for a typed post-boundary record and times out otherwise', { timeout: 20_000 }, async ({ onTestFinished }) => {

scripts/coverage-fork-diagnostics.spec.ts

@@ -3,7 +3,7 @@ import { tmpdir } from 'node:os'
 import { join, resolve } from 'node:path'
 import { runGate } from './run-gates.ts'
 import { PassThrough } from 'node:stream'
-import { afterEach, describe, expect, it, vi } from 'vitest'
+import { afterEach, describe, expect, it, type MockInstance, vi } from 'vitest'
 import { ForksPoolWorker } from 'vitest/node'
 import type { PoolOptions, PoolWorker, TestProject, WorkerRequest, WorkerResponse } from 'vitest/node'
 import { coverageForkPool } from './coverage-fork-diagnostics.ts'
@@ -55,6 +55,43 @@ function poolOptions(root: string) {
   return { options, error, outputStream, errorStream }
 }
 
+/** Observers a pool registers on its worker, once `stubPool` has started it. */
+interface WorkerObservers {
+  off: MockInstance
+  exit?: (code: number | null, signal?: NodeJS.Signals | null) => void
+  error?: () => void
+}
+
+/**
+ * Start a pool whose worker prototypes are stubs, and capture the observers it registers.
+ * The test's cleanups own the pool and its streams.
+ */
+async function stubPool(): Promise<{
+  pool: PoolWorker
+  error: MockInstance<(message: string) => void>
+  observers: WorkerObservers
+}> {
+  const observers: WorkerObservers = {
+    off: vi.spyOn(ForksPoolWorker.prototype, 'off').mockImplementation(() => undefined),
+  }
+  vi.spyOn(ForksPoolWorker.prototype, 'start').mockResolvedValue(undefined)
+  vi.spyOn(ForksPoolWorker.prototype, 'stop').mockResolvedValue(undefined)
+  vi.spyOn(ForksPoolWorker.prototype, 'send').mockImplementation(() => undefined)
+  vi.spyOn(ForksPoolWorker.prototype, 'on').mockImplementation((event, callback) => {
+    if (event === 'exit') observers.exit = callback
+    if (event === 'error') observers.error = callback as () => void
+  })
+  const { options, error, outputStream, errorStream } = poolOptions('/unused')
+  const pool = coverageForkPool.createPoolWorker(options)
+  cleanups.push(async () => {
+    await pool.stop()
+    outputStream.destroy()
+    errorStream.destroy()
+  })
+  await pool.start()
+  return { pool, error, observers }
+}
+
 async function start(root: string, diagnostic = true) {
   const { options, error, outputStream, errorStream } = poolOptions(root)
   const pool: PoolWorker = diagnostic ? coverageForkPool.createPoolWorker(options) : new ForksPoolWorker(options)
@@ -113,21 +150,33 @@ async function configuredPool(pool: 'stock' | 'diagnostic', outcome: 'exit' | 'p
   }
 }
 
+/**
+ * Vitest reports one unexpected worker exit either as the clean exit path or, when
+ * the worker's channel closes under a send, as a runner-level worker error.
+ */
+function workerExitReport(output: string, pool: string): string | undefined {
+  return ['Worker exited unexpectedly', `Worker ${pool} emitted error`].find(message => output.includes(message))
+}
+
 describe('coverage fork configuration entry', () => {
   it('preserves the real Vitest failure and adds metadata before its error report', async ({ signal }) => {
     const [stock, diagnostic] = await Promise.all([
       configuredPool('stock', 'exit', signal),
       configuredPool('diagnostic', 'exit', signal),
     ])
-    for (const result of [stock, diagnostic]) {
+    const stockOutput = stock.output.map(chunk => chunk.text).join('')
+    const output = diagnostic.output.map(chunk => chunk.text).join('')
+    for (const [pool, result, text] of [
+      ['forks', stock, stockOutput],
+      ['coverage-forks', diagnostic, output],
+    ] as const) {
       expect(result.aborted).not.toBe(true)
       expect(result.error).toBeUndefined()
       expect(result.signalCode).toBeNull()
       expect(result.exitCode).toBe(1)
-      expect(result.output.map(chunk => chunk.text).join('')).toContain('Worker exited unexpectedly')
+      expect(workerExitReport(text, pool)).toBeDefined()
     }
-    const output = diagnostic.output.map(chunk => chunk.text).join('')
-    expect(stock.output.map(chunk => chunk.text).join('')).not.toContain('coverage-worker-exit:')
+    expect(stockOutput).not.toContain('coverage-worker-exit:')
     const encoded = /coverage-worker-exit: (\{[^\r\n]*\})/u.exec(output)?.[1]
     expect(encoded).toBeDefined()
     const record = JSON.parse(encoded!) as {
@@ -147,7 +196,8 @@ describe('coverage fork configuration entry', () => {
     })
     expect(record.files.map(file => file.replaceAll('\\', '/')))
       .toEqual([resolve('scripts/fixtures/coverage-fork.fixture.ts').replaceAll('\\', '/')])
-    expect(output.indexOf('coverage-worker-exit:')).toBeLessThan(output.indexOf('Worker exited unexpectedly'))
+    expect(output.indexOf('coverage-worker-exit:'))
+      .toBeLessThan(output.indexOf(workerExitReport(output, 'coverage-forks')!))
   })
 
   it('keeps a successful configured run silent and successful', async ({ signal }) => {
@@ -165,33 +215,36 @@ describe('coverage fork exit callback', () => {
     ['native receiver', { pid: 41 }, 41],
     ['unavailable receiver', undefined, null],
   ] as const)('retains a signal with %s and silences expected shutdown', async (_name, receiver, pid) => {
-    let exitListener: ((value: unknown) => void) | undefined
-    vi.spyOn(ForksPoolWorker.prototype, 'start').mockResolvedValue(undefined)
-    vi.spyOn(ForksPoolWorker.prototype, 'stop').mockResolvedValue(undefined)
-    vi.spyOn(ForksPoolWorker.prototype, 'send').mockImplementation(() => undefined)
-    vi.spyOn(ForksPoolWorker.prototype, 'off').mockImplementation(() => undefined)
-    vi.spyOn(ForksPoolWorker.prototype, 'on').mockImplementation((event, callback) => {
-      if (event === 'exit') exitListener = callback
-    })
-    const { options, error, outputStream, errorStream } = poolOptions('/unused')
-    const pool = coverageForkPool.createPoolWorker(options)
-    cleanups.push(async () => {
-      await pool.stop()
-      outputStream.destroy()
-      errorStream.destroy()
-    })
-    await pool.start()
+    const { pool, error, observers } = await stubPool()
     pool.send(request('/signal.spec.ts', 9))
-    if (exitListener === undefined) throw new Error('exit observer was not registered')
-    Reflect.apply(exitListener, receiver, [null, 'SIGTERM'])
+    const exit = observers.exit
+    if (exit === undefined) throw new Error('exit observer was not registered')
+    Reflect.apply(exit, receiver, [null, 'SIGTERM'])
     expect(error).toHaveBeenCalledWith(`coverage-worker-exit: ${JSON.stringify({
       pid, project: 'fixture', workerId: 9, files: ['/signal.spec.ts'], exitCode: null, signalCode: 'SIGTERM',
     })}`)
     error.mockClear()
     pool.send(stop)
-    expect(() => { Reflect.apply(exitListener!, undefined, [null, 'SIGTERM']) }).not.toThrow()
+    expect(() => { Reflect.apply(exit, undefined, [null, 'SIGTERM']) }).not.toThrow()
     expect(error).not.toHaveBeenCalled()
   })
+
+  it('reports an exit whose worker errored even when a teardown raced it', async () => {
+    const { pool, error, observers } = await stubPool()
+    pool.send(request('/raced.spec.ts', 11))
+    const failed = observers.error
+    if (failed === undefined) throw new Error('error observer was not registered')
+    // Node reports the failed reply before it reports the exit of the same child.
+    failed()
+    await pool.stop()
+    expect(observers.off).not.toHaveBeenCalled()
+    const exit = observers.exit
+    if (exit === undefined) throw new Error('exit observer was not registered')
+    Reflect.apply(exit, { pid: 42 }, [23, null])
+    expect(error).toHaveBeenCalledWith(`coverage-worker-exit: ${JSON.stringify({
+      pid: 42, project: 'fixture', workerId: 11, files: ['/raced.spec.ts'], exitCode: 23, signalCode: null,
+    })}`)
+  })
 })
 
 describe('coverage fork exit diagnostics', () => {

scripts/coverage-fork-diagnostics.ts

@@ -5,12 +5,17 @@ import type { PoolOptions, PoolRunnerInitializer, WorkerRequest, WorkerResponse
 
 const FINISHED_RESPONSE_TYPE: WorkerResponse['type'] = 'testfileFinished'
 
-/** Attribute only outstanding requests; stop silences exits, while cancel does not. */
+/**
+ * Attribute only outstanding requests; stop silences exits, while cancel and a
+ * worker error observed before that stop do not.
+ */
 class DiagnosticForkWorker extends ForksPoolWorker {
   private expectedExit = false
   private observing = false
+  private workerErrored = false
   private request: Extract<WorkerRequest, { type: 'run' | 'collect' }> | undefined
   private readonly reportExit: (this: ChildProcess | undefined, code: number | null, signal?: NodeJS.Signals | null) => void
+  private readonly observeError: () => void
 
   constructor(options: PoolOptions) {
     super(options)
@@ -27,17 +32,24 @@ class DiagnosticForkWorker extends ForksPoolWorker {
     }
     // ForksPoolWorker.on forwards Node ChildProcess events and their receiver.
     this.reportExit = function (code, signal) { report(this, code, signal) }
+    this.observeError = () => {
+      // A channel that closed under a send means the worker is not shutting down
+      // orderly, so the exit that follows keeps its code and signal. A stop
+      // requested before this error still silences that exit.
+      if (!this.expectedExit) this.workerErrored = true
+    }
   }
 
   override async start(): Promise<void> {
     await super.start()
     this.on('exit', this.reportExit)
+    this.on('error', this.observeError)
     this.observing = true
   }
 
   override send(message: WorkerRequest): void {
     if (message.type === 'run' || message.type === 'collect') this.request = message
-    if (message.type === 'stop') this.expectedExit = true
+    if (message.type === 'stop' && !this.workerErrored) this.expectedExit = true
     super.send(message)
   }
 
@@ -53,10 +65,14 @@ class DiagnosticForkWorker extends ForksPoolWorker {
   }
 
   override async stop(): Promise<void> {
-    this.expectedExit = true
-    if (this.observing) {
-      this.off('exit', this.reportExit)
-      this.observing = false
+    // A worker that already errored is not shutting down orderly; keep the exit
+    // observer so its code and signal still reach the coverage log.
+    if (!this.workerErrored) {
+      this.expectedExit = true
+      if (this.observing) {
+        this.off('exit', this.reportExit)
+        this.observing = false
+      }
     }
     await super.stop()
   }