Sync run.json turn count and idle the shell at every turn boundary

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 };

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) => {

src/tui/runner.ts

@@ -1387,6 +1387,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.
+    onTurnBoundarySnapshot: () => {
+      void persistRunSnapshot("running");
+    },
   });
 
   // MCP servers connected so far, keyed by name so a reconnect after a failure
@@ -1440,9 +1445,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 (event.type === "reactor.done") {
-      void persistRunSnapshot("running");
-    }
   };
 
   // Tool count before any MCP server connects; a reload is only worthwhile if

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

@@ -0,0 +1,164 @@
+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. 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: () => [] };
+
+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 writes: Promise<void>[] = [];
+      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 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,
+        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 writes: Promise<void>[] = [];
+      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());
+      }
+      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 {
+      let runningWrite: Promise<void> | 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.
+      runSink.sink(inferenceDone());
+      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 });
+    }
+  });
+});