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-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) => { diff --git a/src/tui/runner.ts b/src/tui/runner.ts index b451f8c57..49a5d0ed8 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -1387,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 @@ -1440,9 +1445,6 @@ export async function runTUI(initialConfig: Config): Promise { const streamSink = (event: Parameters[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 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..fde2f9adf --- /dev/null +++ b/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, 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[] = []; + 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[] = []; + 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 | 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 }); + } + }); +});