From e15087dfbfa9d2d22529372e56076c077cb420d6 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 7 Aug 2026 01:45:17 -0700 Subject: [PATCH 1/4] Cover the run-idle turn boundary with a two-turn regression test runtime-bridge.ts already keys the shell's run-idle transition off inference.done (the turn boundary), not reactor.done (shutdown, fires once) -- fixed alongside the queued-message drain in an earlier commit. That left the multi-turn case unverified: a test only proved a single turn settled run back to idle, not that a second turn starts from idle again after the first one closes it out. Add a regression test that drives two consecutive turns end to end and asserts run returns to idle, and that a message submitted between them sends immediately rather than routing through the queue. --- src/tui-opentui/runtime-bridge.test.ts | 43 ++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/src/tui-opentui/runtime-bridge.test.ts b/src/tui-opentui/runtime-bridge.test.ts index cc521e590..9e2354d31 100644 --- a/src/tui-opentui/runtime-bridge.test.ts +++ b/src/tui-opentui/runtime-bridge.test.ts @@ -298,6 +298,49 @@ describe("attachSessionBridge", () => { ) }) + test("run returns to idle between two consecutive turns, not only at reactor shutdown", async () => { + // CL-5570: `run` must flip back to idle at every turn boundary + // (`inference.done`), so a second Enter after the first reply sends + // immediately instead of routing through the queue. reactor.done is + // shutdown, not a turn boundary, and never fires between turns. + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }) + const port = createRecordingPort() + const bridge = attachSessionBridge(shell, port) + try { + bridge.submit("first turn", "immediate") + expect(shell.session.run).toBe("busy") + bridge.handle({ type: "inference.start" }) + bridge.handle({ + type: "inference.text.delta", + data: { token: "hi" }, + }) + bridge.handle({ type: "inference.done" }) + expect(shell.session.run).toBe("idle") + + bridge.submit("second turn", "immediate") + expect(shell.session.run).toBe("busy") + bridge.handle({ type: "inference.start" }) + bridge.handle({ + type: "inference.text.delta", + data: { token: "hi again" }, + }) + bridge.handle({ type: "inference.done" }) + expect(shell.session.run).toBe("idle") + } finally { + bridge.dispose() + shell.dispose() + } + }, + { width: 80, height: 24 }, + ) + }) + test("run stays busy after inference.done while a tool call is still outstanding", async () => { await withTestRenderer( async (h) => { From 08145fc2923e704deeb17e4e70c687a6100d4fc8 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 7 Aug 2026 01:45:27 -0700 Subject: [PATCH 2/4] Persist run.json's turn count at every turn boundary The mid-run progress snapshot in the TUI runner only re-fired on reactor.done, which fires exactly once, at agent shutdown, and never between turns of a long-lived interactive session. A live monorepo session showed run.json stuck at turnsUsed: 0 and status: running for its entire multi-turn lifetime, with dozens of completed turns already in context/turns.jsonl -- resume pickers and anything else trusting run.json had no truthful signal until the process closed. inference.done is the turn boundary every reactor cycle guarantees (the same one the shell's run-idle transition keys off), so key the snapshot write off that instead. The terminal write on clean exit and the crash path both already write through directly with the final status, so this only changes progress snapshots taken while the run is still live. --- src/tui/runner.ts | 14 +++++++++++++- tests/unit/tui/runner.test.ts | 14 ++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/src/tui/runner.ts b/src/tui/runner.ts index b451f8c57..0853a4e7a 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -207,6 +207,18 @@ export function resolveExitCode(args: ResolveExitCodeArgs): number { return 0; } +// `inference.done` is the turn boundary every reactor cycle guarantees; +// `reactor.done` fires once, at shutdown, and never between turns of a +// long-lived interactive session. Keying the mid-run run.json snapshot off +// `reactor.done` left turnsUsed frozen at its resume-time value for the +// entire session — a live monorepo session showed turnsUsed: 0 with dozens +// of turns already in the turns log. The terminal write on close still goes +// through writeRunSnapshot directly with the real final status, so this +// only needs to cover progress snapshots taken while the run is live. +export function isRunSnapshotTurnBoundary(eventType: string): boolean { + return eventType === "inference.done"; +} + /** One-line transcript block when resume history fails to load. */ export function resumeTranscriptLoadErrorBlock(err: unknown): { type: "error"; @@ -1440,7 +1452,7 @@ export async function runTUI(initialConfig: Config): Promise { const streamSink = (event: Parameters[0]): void => { runSink.sink(event); cycleRecorder.handleEvent(event); - if (event.type === "reactor.done") { + if (isRunSnapshotTurnBoundary(event.type)) { void persistRunSnapshot("running"); } }; diff --git a/tests/unit/tui/runner.test.ts b/tests/unit/tui/runner.test.ts index 8ae66c3e6..a91aa472c 100644 --- a/tests/unit/tui/runner.test.ts +++ b/tests/unit/tui/runner.test.ts @@ -3,6 +3,7 @@ import { EventEmitter } from "node:events"; import { createTUIEventEmitter, getTUIRunSummaryStatus, + isRunSnapshotTurnBoundary, loadLocalSettingsWriteBase, resumeTranscriptLoadErrorBlock, } from "../../../src/tui/runner.js"; @@ -21,6 +22,19 @@ test("createTUIEventEmitter can emit and receive events", () => { expect(received.length).toBe(1); }); +// Regression: run.json's turnsUsed must update every turn, not only once +// at reactor shutdown. reactor.done fires exactly once, at agent shutdown, +// so a live multi-turn interactive session never had its progress snapshot +// re-fire until close — turnsUsed sat frozen at its resume-time value the +// whole session (CL-5534). inference.done is the turn boundary every +// reactor cycle guarantees, so that's what a mid-run snapshot must key off. +test("isRunSnapshotTurnBoundary fires on inference.done, not reactor.done", () => { + expect(isRunSnapshotTurnBoundary("inference.done")).toBe(true); + expect(isRunSnapshotTurnBoundary("reactor.done")).toBe(false); + expect(isRunSnapshotTurnBoundary("reactor.error")).toBe(false); + expect(isRunSnapshotTurnBoundary("connector.reply")).toBe(false); +}); + test("getTUIRunSummaryStatus distinguishes done, failed, and cancelled runs", () => { expect(getTUIRunSummaryStatus(true, undefined)).toBe("done"); expect(getTUIRunSummaryStatus(true, "network failed")).toBe("failed"); From 801ee1f23d8a16296057a881d767a3ff4394dae5 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 7 Aug 2026 07:36:10 -0700 Subject: [PATCH 3/4] Un-export the turn-boundary predicate, cover it end to end instead isRunSnapshotTurnBoundary had exactly one production caller in the same file, so its export existed only for a unit test. Make it module-private and replace that test with an end-to-end one that drives createRunSink, saveState, and loadState against a real temp session directory: turnsUsed incrementing per turn as read back off disk, 20 rapid back-to-back turns with no settling delay, and a late running write racing a done write to confirm status never resurrects. --- src/tui/runner.ts | 2 +- tests/unit/session/run-state-e2e.test.ts | 141 +++++++++++++++++++++++ tests/unit/tui/runner.test.ts | 14 --- 3 files changed, 142 insertions(+), 15 deletions(-) create mode 100644 tests/unit/session/run-state-e2e.test.ts diff --git a/src/tui/runner.ts b/src/tui/runner.ts index 0853a4e7a..63ef8544a 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -215,7 +215,7 @@ export function resolveExitCode(args: ResolveExitCodeArgs): number { // of turns already in the turns log. The terminal write on close still goes // through writeRunSnapshot directly with the real final status, so this // only needs to cover progress snapshots taken while the run is live. -export function isRunSnapshotTurnBoundary(eventType: string): boolean { +function isRunSnapshotTurnBoundary(eventType: string): boolean { return eventType === "inference.done"; } diff --git a/tests/unit/session/run-state-e2e.test.ts b/tests/unit/session/run-state-e2e.test.ts new file mode 100644 index 000000000..51985317a --- /dev/null +++ b/tests/unit/session/run-state-e2e.test.ts @@ -0,0 +1,141 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { EventEmitter } from "node:events"; + +import { describe, expect, test } from "bun:test"; +import type { ReactorEmittedEvent } from "@intx/inference"; + +import { generateSessionId } from "../../../src/session/index.js"; +import { createRunSink } from "../../../src/session/run-sink.js"; +import { saveState, loadState, type RunState } from "../../../src/session/state.js"; + +// End-to-end coverage for the run.json turn-boundary snapshot fix (CL-5534): +// createRunSink, saveState, and loadState run for real against a temp +// session directory — nothing mocked. This is what now covers +// isRunSnapshotTurnBoundary's observable effect, since that predicate was +// un-exported from src/tui/runner.ts as a testability-only surface with a +// single production caller. + +const noopHookManager = { dispatchPostTurn: () => undefined, getStatuses: () => [] }; + +function inferenceDone(): ReactorEmittedEvent { + return { + type: "inference.done", + data: { + turn: { content: [] }, + usage: {}, + source: "primary", + }, + } as unknown as ReactorEmittedEvent; +} + +function baseState(overrides: Partial, turnsUsed: number): RunState { + return { + status: "running", + turnsUsed, + task: "e2e run-state test", + startedAt: Date.now(), + model: "test-provider:test-model", + mcpServers: [], + ...overrides, + }; +} + +describe("run.json turn-boundary snapshots — end to end", () => { + test("turnsUsed increments and is readable off disk after every turn, and status settles to done", async () => { + const cwd = mkdtempSync(join(tmpdir(), "corbits-run-state-cwd-")); + const home = mkdtempSync(join(tmpdir(), "corbits-run-state-home-")); + const sessionId = generateSessionId(); + try { + const runSink = createRunSink({ emitter: new EventEmitter(), hookManager: noopHookManager }); + + const observed: number[] = []; + for (let turn = 1; turn <= 4; turn++) { + runSink.sink(inferenceDone()); + await saveState(cwd, sessionId, baseState({ status: "running" }, runSink.getTurnCount()), home); + const onDisk = await loadState(cwd, sessionId, home); + expect(onDisk).not.toBeNull(); + observed.push(onDisk!.turnsUsed); + } + + expect(observed).toEqual([1, 2, 3, 4]); + + runSink.sink({ type: "reactor.done", data: {} } as unknown as ReactorEmittedEvent); + await saveState( + cwd, + sessionId, + baseState({ status: "done", finishedAt: Date.now() }, runSink.getTurnCount()), + home, + ); + const finalState = await loadState(cwd, sessionId, home); + expect(finalState?.status).toBe("done"); + expect(finalState?.turnsUsed).toBe(4); + } finally { + rmSync(cwd, { recursive: true, force: true }); + rmSync(home, { recursive: true, force: true }); + } + }); + + test("20 rapid back-to-back turns with no settling delay serialize without dropping a write", async () => { + const cwd = mkdtempSync(join(tmpdir(), "corbits-run-state-cwd-")); + const home = mkdtempSync(join(tmpdir(), "corbits-run-state-home-")); + const sessionId = generateSessionId(); + try { + const runSink = createRunSink({ emitter: new EventEmitter(), hookManager: noopHookManager }); + + // Fire all 20 turns and their snapshot writes back to back, with no + // await between them — the per-session writeChains promise chain in + // src/session/state.ts is what keeps these ordered rather than the + // caller awaiting each one before starting the next. + const writes: Promise[] = []; + for (let turn = 1; turn <= 20; turn++) { + runSink.sink(inferenceDone()); + writes.push(saveState(cwd, sessionId, baseState({ status: "running" }, runSink.getTurnCount()), home)); + } + await Promise.all(writes); + + const finalState = await loadState(cwd, sessionId, home); + expect(finalState?.turnsUsed).toBe(20); + expect(runSink.getTurnCount()).toBe(20); + } finally { + rmSync(cwd, { recursive: true, force: true }); + rmSync(home, { recursive: true, force: true }); + } + }); + + test("a late in-flight running write racing a done write never resurrects status to running", async () => { + const cwd = mkdtempSync(join(tmpdir(), "corbits-run-state-cwd-")); + const home = mkdtempSync(join(tmpdir(), "corbits-run-state-home-")); + const sessionId = generateSessionId(); + try { + const runSink = createRunSink({ emitter: new EventEmitter(), hookManager: noopHookManager }); + runSink.sink(inferenceDone()); + + // Issue a "running" progress snapshot but do not await it before + // issuing the terminal "done" write right behind it — this models a + // straggler turn-boundary snapshot racing the close-out write. + const runningWrite = saveState( + cwd, + sessionId, + baseState({ status: "running" }, runSink.getTurnCount()), + home, + ); + runSink.sink({ type: "reactor.done", data: {} } as unknown as ReactorEmittedEvent); + const doneWrite = saveState( + cwd, + sessionId, + baseState({ status: "done", finishedAt: Date.now() }, runSink.getTurnCount()), + home, + ); + + await Promise.all([runningWrite, doneWrite]); + + const finalState = await loadState(cwd, sessionId, home); + expect(finalState?.status).toBe("done"); + } finally { + rmSync(cwd, { recursive: true, force: true }); + rmSync(home, { recursive: true, force: true }); + } + }); +}); diff --git a/tests/unit/tui/runner.test.ts b/tests/unit/tui/runner.test.ts index a91aa472c..8ae66c3e6 100644 --- a/tests/unit/tui/runner.test.ts +++ b/tests/unit/tui/runner.test.ts @@ -3,7 +3,6 @@ import { EventEmitter } from "node:events"; import { createTUIEventEmitter, getTUIRunSummaryStatus, - isRunSnapshotTurnBoundary, loadLocalSettingsWriteBase, resumeTranscriptLoadErrorBlock, } from "../../../src/tui/runner.js"; @@ -22,19 +21,6 @@ test("createTUIEventEmitter can emit and receive events", () => { expect(received.length).toBe(1); }); -// Regression: run.json's turnsUsed must update every turn, not only once -// at reactor shutdown. reactor.done fires exactly once, at agent shutdown, -// so a live multi-turn interactive session never had its progress snapshot -// re-fire until close — turnsUsed sat frozen at its resume-time value the -// whole session (CL-5534). inference.done is the turn boundary every -// reactor cycle guarantees, so that's what a mid-run snapshot must key off. -test("isRunSnapshotTurnBoundary fires on inference.done, not reactor.done", () => { - expect(isRunSnapshotTurnBoundary("inference.done")).toBe(true); - expect(isRunSnapshotTurnBoundary("reactor.done")).toBe(false); - expect(isRunSnapshotTurnBoundary("reactor.error")).toBe(false); - expect(isRunSnapshotTurnBoundary("connector.reply")).toBe(false); -}); - test("getTUIRunSummaryStatus distinguishes done, failed, and cancelled runs", () => { expect(getTUIRunSummaryStatus(true, undefined)).toBe("done"); expect(getTUIRunSummaryStatus(true, "network failed")).toBe("failed"); From b0493e6195580b52f5416008668b81b5f8627a9c Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 7 Aug 2026 08:49:25 -0700 Subject: [PATCH 4/4] Move the run.json turn-boundary snapshot trigger into run-sink MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit run.json is session domain state, and src/session/run-sink.ts already owns the turn count and already special-cases inference.done there (clearing a stale runError). The mid-run snapshot trigger lived in src/tui/runner.ts instead, keyed off a second subscription to the same event stream — the exact shape that has already cost this constraint three renderer swaps. Move the cadence into run-sink via an onTurnBoundarySnapshot callback, so the renderer only owns how to persist a snapshot, not when one is due. Named apart from the onTurnBoundary predicate in src/agent/reactor-events.ts, which run-sink now uses for the same inference.done check it already made inline. The end-to-end test now drives the callback the way production wiring does, instead of calling saveState directly after each sink call, so it actually exercises the trigger rather than simulating it. --- src/session/run-sink.ts | 14 ++++- src/tui/runner.ts | 20 ++---- tests/unit/session/run-state-e2e.test.ts | 79 +++++++++++++++--------- 3 files changed, 69 insertions(+), 44 deletions(-) diff --git a/src/session/run-sink.ts b/src/session/run-sink.ts index 0e475f5df..d86f344b0 100644 --- a/src/session/run-sink.ts +++ b/src/session/run-sink.ts @@ -23,6 +23,17 @@ export type RunSinkArgs = { // Continues a resumed session's persisted run.json turn count instead of // restarting the collector at zero. initialTurnCount?: number; + // Fired at every turn boundary so a caller can persist a mid-run run.json + // snapshot. `inference.done` is the turn boundary every reactor cycle + // guarantees; `reactor.done` fires once, at shutdown, and never between + // turns of a long-lived interactive session. Keying the mid-run snapshot + // off `reactor.done` left turnsUsed frozen at its resume-time value for + // the entire session — a live monorepo session showed turnsUsed: 0 with + // dozens of turns already in the turns log. This cadence lives here, + // alongside the turn count it reports, rather than in a second + // subscription to the same event stream in a renderer: the renderer has + // already been swapped out from under this constraint three times. + onTurnBoundarySnapshot?: () => void; }; export type RunSink = { @@ -70,7 +81,7 @@ export function resolveExecRunStatus(args: { } export function createRunSink(args: RunSinkArgs): RunSink { - const { emitter, hookManager, onTurnComplete, initialTurnCount } = args; + const { emitter, hookManager, onTurnComplete, initialTurnCount, onTurnBoundarySnapshot } = args; function hasConfiguredHooks(): boolean { return hookManager.getStatuses().length > 0; @@ -115,6 +126,7 @@ export function createRunSink(args: RunSinkArgs): RunSink { // would mark a recovered successful send as failed. if (onTurnBoundary(event)) { runError = undefined; + onTurnBoundarySnapshot?.(); } if (event.type === "reactor.error") { const data = event.data as { error: string }; diff --git a/src/tui/runner.ts b/src/tui/runner.ts index 63ef8544a..49a5d0ed8 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -207,18 +207,6 @@ export function resolveExitCode(args: ResolveExitCodeArgs): number { return 0; } -// `inference.done` is the turn boundary every reactor cycle guarantees; -// `reactor.done` fires once, at shutdown, and never between turns of a -// long-lived interactive session. Keying the mid-run run.json snapshot off -// `reactor.done` left turnsUsed frozen at its resume-time value for the -// entire session — a live monorepo session showed turnsUsed: 0 with dozens -// of turns already in the turns log. The terminal write on close still goes -// through writeRunSnapshot directly with the real final status, so this -// only needs to cover progress snapshots taken while the run is live. -function isRunSnapshotTurnBoundary(eventType: string): boolean { - return eventType === "inference.done"; -} - /** One-line transcript block when resume history fails to load. */ export function resumeTranscriptLoadErrorBlock(err: unknown): { type: "error"; @@ -1399,6 +1387,11 @@ export async function runTUI(initialConfig: Config): Promise { duration_ms: ctx.durationMs, }); }, + // persistRunSnapshot is defined below but not invoked until the stream + // starts consuming events, well after this closure captures it. + onTurnBoundarySnapshot: () => { + void persistRunSnapshot("running"); + }, }); // MCP servers connected so far, keyed by name so a reconnect after a failure @@ -1452,9 +1445,6 @@ export async function runTUI(initialConfig: Config): Promise { const streamSink = (event: Parameters[0]): void => { runSink.sink(event); cycleRecorder.handleEvent(event); - if (isRunSnapshotTurnBoundary(event.type)) { - void persistRunSnapshot("running"); - } }; // Tool count before any MCP server connects; a reload is only worthwhile if diff --git a/tests/unit/session/run-state-e2e.test.ts b/tests/unit/session/run-state-e2e.test.ts index 51985317a..fde2f9adf 100644 --- a/tests/unit/session/run-state-e2e.test.ts +++ b/tests/unit/session/run-state-e2e.test.ts @@ -12,10 +12,12 @@ import { saveState, loadState, type RunState } from "../../../src/session/state. // End-to-end coverage for the run.json turn-boundary snapshot fix (CL-5534): // createRunSink, saveState, and loadState run for real against a temp -// session directory — nothing mocked. This is what now covers -// isRunSnapshotTurnBoundary's observable effect, since that predicate was -// un-exported from src/tui/runner.ts as a testability-only surface with a -// single production caller. +// session directory — nothing mocked. The snapshot cadence itself now lives +// in createRunSink's onTurnBoundarySnapshot callback (moved out of +// src/tui/runner.ts so a future renderer swap cannot silently drop it +// again), so these tests drive that callback exactly as production wiring +// does: nothing here calls saveState directly from the turn loop, only from +// inside onTurnBoundarySnapshot. const noopHookManager = { dispatchPostTurn: () => undefined, getStatuses: () => [] }; @@ -48,19 +50,29 @@ describe("run.json turn-boundary snapshots — end to end", () => { const home = mkdtempSync(join(tmpdir(), "corbits-run-state-home-")); const sessionId = generateSessionId(); try { - const runSink = createRunSink({ emitter: new EventEmitter(), hookManager: noopHookManager }); - + const writes: Promise[] = []; const observed: number[] = []; + const runSink = createRunSink({ + emitter: new EventEmitter(), + hookManager: noopHookManager, + onTurnBoundarySnapshot: () => { + observed.push(runSink.getTurnCount()); + writes.push(saveState(cwd, sessionId, baseState({ status: "running" }, runSink.getTurnCount()), home)); + }, + }); + for (let turn = 1; turn <= 4; turn++) { runSink.sink(inferenceDone()); - await saveState(cwd, sessionId, baseState({ status: "running" }, runSink.getTurnCount()), home); - const onDisk = await loadState(cwd, sessionId, home); - expect(onDisk).not.toBeNull(); - observed.push(onDisk!.turnsUsed); } + await Promise.all(writes); + // The snapshot callback must fire once per turn, not just once at the + // end — that's the regression this test guards against. expect(observed).toEqual([1, 2, 3, 4]); + const onDisk = await loadState(cwd, sessionId, home); + expect(onDisk?.turnsUsed).toBe(4); + runSink.sink({ type: "reactor.done", data: {} } as unknown as ReactorEmittedEvent); await saveState( cwd, @@ -82,16 +94,21 @@ describe("run.json turn-boundary snapshots — end to end", () => { const home = mkdtempSync(join(tmpdir(), "corbits-run-state-home-")); const sessionId = generateSessionId(); try { - const runSink = createRunSink({ emitter: new EventEmitter(), hookManager: noopHookManager }); - - // Fire all 20 turns and their snapshot writes back to back, with no - // await between them — the per-session writeChains promise chain in - // src/session/state.ts is what keeps these ordered rather than the - // caller awaiting each one before starting the next. const writes: Promise[] = []; + const runSink = createRunSink({ + emitter: new EventEmitter(), + hookManager: noopHookManager, + onTurnBoundarySnapshot: () => { + writes.push(saveState(cwd, sessionId, baseState({ status: "running" }, runSink.getTurnCount()), home)); + }, + }); + + // Fire all 20 turns back to back, with no await between them — the + // per-session writeChains promise chain in src/session/state.ts is what + // keeps the resulting writes ordered, not the caller awaiting each one + // before starting the next. for (let turn = 1; turn <= 20; turn++) { runSink.sink(inferenceDone()); - writes.push(saveState(cwd, sessionId, baseState({ status: "running" }, runSink.getTurnCount()), home)); } await Promise.all(writes); @@ -109,18 +126,24 @@ describe("run.json turn-boundary snapshots — end to end", () => { const home = mkdtempSync(join(tmpdir(), "corbits-run-state-home-")); const sessionId = generateSessionId(); try { - const runSink = createRunSink({ emitter: new EventEmitter(), hookManager: noopHookManager }); - runSink.sink(inferenceDone()); - - // Issue a "running" progress snapshot but do not await it before - // issuing the terminal "done" write right behind it — this models a + let runningWrite: Promise | undefined; + const runSink = createRunSink({ + emitter: new EventEmitter(), + hookManager: noopHookManager, + onTurnBoundarySnapshot: () => { + runningWrite = saveState( + cwd, + sessionId, + baseState({ status: "running" }, runSink.getTurnCount()), + home, + ); + }, + }); + + // The turn-boundary snapshot fires and is left un-awaited before the + // terminal "done" write follows right behind it — this models a // straggler turn-boundary snapshot racing the close-out write. - const runningWrite = saveState( - cwd, - sessionId, - baseState({ status: "running" }, runSink.getTurnCount()), - home, - ); + runSink.sink(inferenceDone()); runSink.sink({ type: "reactor.done", data: {} } as unknown as ReactorEmittedEvent); const doneWrite = saveState( cwd,