Move the run.json turn-boundary snapshot trigger into run-sink

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 onTurnBoundary callback, so the renderer only owns how to persist
a snapshot, not when one is due.

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.

Author: TheGreatAxios

src/session/run-sink.ts

@@ -22,6 +22,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.
+  onTurnBoundary?: () => void;
 };
 
 export type RunSink = {
@@ -69,7 +80,7 @@ export function resolveExecRunStatus(args: {
 }
 
 export function createRunSink(args: RunSinkArgs): RunSink {
-  const { emitter, hookManager, onTurnComplete, initialTurnCount } = args;
+  const { emitter, hookManager, onTurnComplete, initialTurnCount, onTurnBoundary } = args;
 
   function hasConfiguredHooks(): boolean {
     return hookManager.getStatuses().length > 0;
@@ -114,6 +125,7 @@ export function createRunSink(args: RunSinkArgs): RunSink {
     // would mark a recovered successful send as failed.
     if (event.type === "inference.done") {
       runError = undefined;
+      onTurnBoundary?.();
     }
     if (event.type === "reactor.error") {
       const data = event.data as { error: string };

src/tui/runner.ts

@@ -201,18 +201,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";
@@ -1358,6 +1346,11 @@ export async function runTUI(initialConfig: Config): Promise<number> {
         duration_ms: ctx.durationMs,
       });
     },
+    // persistRunSnapshot is defined below but not invoked until the stream
+    // starts consuming events, well after this closure captures it.
+    onTurnBoundary: () => {
+      void persistRunSnapshot("running");
+    },
   });
 
   // MCP servers connected so far, keyed by name so a reconnect after a failure
@@ -1411,9 +1404,6 @@ export async function runTUI(initialConfig: Config): Promise<number> {
   const streamSink = (event: Parameters<typeof runSink.sink>[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

tests/unit/session/run-state-e2e.test.ts

@@ -12,10 +12,11 @@ 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 onTurnBoundary 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 onTurnBoundary.
 
 const noopHookManager = { dispatchPostTurn: () => undefined, getStatuses: () => [] };
 
@@ -48,18 +49,22 @@ 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<void>[] = [];
+      const runSink = createRunSink({
+        emitter: new EventEmitter(),
+        hookManager: noopHookManager,
+        onTurnBoundary: () => {
+          writes.push(saveState(cwd, sessionId, baseState({ status: "running" }, runSink.getTurnCount()), home));
+        },
+      });
 
-      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);
       }
+      await Promise.all(writes);
 
-      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(
@@ -82,16 +87,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<void>[] = [];
+      const runSink = createRunSink({
+        emitter: new EventEmitter(),
+        hookManager: noopHookManager,
+        onTurnBoundary: () => {
+          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 +119,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<void> | undefined;
+      const runSink = createRunSink({
+        emitter: new EventEmitter(),
+        hookManager: noopHookManager,
+        onTurnBoundary: () => {
+          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,