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.

Author: TheGreatAxios

src/tui/runner.ts

@@ -209,7 +209,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";
 }
 

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<RunState>, 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<void>[] = [];
+      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 });
+    }
+  });
+});

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");