From fbd385cfb1c782c45fe796db2f57f52365855aac Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 16 Oct 2026 20:02:06 -0700 Subject: [PATCH 1/2] refactor(session): drop the run.json snapshot debounce --- src/exec/runner.ts | 12 +--- src/session/run-snapshot-debounce.test.ts | 82 ----------------------- src/session/state.ts | 56 ---------------- src/tui/runner/exit.ts | 37 ++-------- src/tui/runner/session.ts | 3 +- src/tui/runner/state.ts | 4 -- 6 files changed, 6 insertions(+), 188 deletions(-) delete mode 100644 src/session/run-snapshot-debounce.test.ts diff --git a/src/exec/runner.ts b/src/exec/runner.ts index d7b80f24f..d02256a23 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -102,7 +102,6 @@ import { clearActiveDisposeHost, } from "../session/active-host.js"; import { - createRunSnapshotDebouncer, finalizeRunState, loadState, saveState, @@ -536,10 +535,6 @@ export async function runExec(config: Config): Promise { : {}), }; let stopHeartbeat: (() => void) | undefined; - // CL-10667: debounced per-turn boundary snapshots; terminal exposes flush. - let runSnapshotDebouncer: - | ReturnType - | undefined; const persist = async ( status: "running" | "done" | "failed" | "cancelled", @@ -547,7 +542,6 @@ export async function runExec(config: Config): Promise { ): Promise => { if (finalized && status === "running") return; if (status !== "running") { - await runSnapshotDebouncer?.flush(); finalized = true; stopHeartbeat?.(); stopHeartbeat = undefined; @@ -601,9 +595,6 @@ export async function runExec(config: Config): Promise { }; await persist("running"); - runSnapshotDebouncer = createRunSnapshotDebouncer({ - write: () => persist("running"), - }); stopHeartbeat = startRunHeartbeat({ shouldTick: () => !finalized, tick: () => persist("running"), @@ -1057,8 +1048,7 @@ export async function runExec(config: Config): Promise { activeRunHandle.cacheWriteModel = `${event.data.source.provider}:${event.data.source.model}`; } } - // CL-10667: debounced — the write recomputes state when it fires. - runSnapshotDebouncer?.schedule(); + void persist("running"); }, resolveContextDir: () => workdir, }); diff --git a/src/session/run-snapshot-debounce.test.ts b/src/session/run-snapshot-debounce.test.ts deleted file mode 100644 index e54c68d13..000000000 --- a/src/session/run-snapshot-debounce.test.ts +++ /dev/null @@ -1,82 +0,0 @@ -import { describe, expect, test } from "bun:test"; - -import { createTempDirs } from "../../testkit/temporary-dirs.js"; -import { generateSessionId } from "./index.js"; -import { - createRunSnapshotDebouncer, - loadState, - saveState, - type RunState, -} from "./state.js"; - -function snap(turnsUsed: number): RunState { - return { - status: "running", - turnsUsed, - task: "debounce test", - startedAt: 1, - model: "test-provider:test-model", - mcpServers: [], - }; -} - -describe("createRunSnapshotDebouncer (CL-10667)", () => { - test("a burst of schedules behind one window lands as one write", async () => { - const { cwd, home, cleanup } = createTempDirs( - "corbits-debounce-cwd-", - "corbits-debounce-home-", - ); - const sessionId = generateSessionId(); - try { - let turnsUsed = 0; - const debouncer = createRunSnapshotDebouncer({ - write: async () => { - turnsUsed += 1; - await saveState(cwd, sessionId, snap(turnsUsed), home); - }, - intervalMs: 20, - }); - - for (let i = 0; i < 10; i++) debouncer.schedule(); - // The timer fires the coalesced batch once the window elapses. - await new Promise((resolve) => setTimeout(resolve, 120)); - - const onDisk = await loadState(cwd, sessionId, home); - expect(onDisk).toMatchObject({ kind: "ok", state: { turnsUsed: 1 } }); - } finally { - cleanup(); - } - }); - - test("flush runs a pending write immediately; a clean flush writes nothing", async () => { - let writes = 0; - const debouncer = createRunSnapshotDebouncer({ - write: async () => { - writes += 1; - }, - intervalMs: 60_000, // armed but never able to fire on its own - }); - - debouncer.schedule(); - await debouncer.flush(); - expect(writes).toBe(1); - // Nothing pending: flush is a no-op, not a second write. - await debouncer.flush(); - expect(writes).toBe(1); - }); - - test("flush after the timer already fired is a no-op", async () => { - let writes = 0; - const debouncer = createRunSnapshotDebouncer({ - write: async () => { - writes += 1; - }, - intervalMs: 15, - }); - - debouncer.schedule(); - await new Promise((resolve) => setTimeout(resolve, 60)); - await debouncer.flush(); - expect(writes).toBe(1); - }); -}); diff --git a/src/session/state.ts b/src/session/state.ts index c1060d5fe..678a7b7ed 100644 --- a/src/session/state.ts +++ b/src/session/state.ts @@ -131,62 +131,6 @@ export async function saveState( return write; } -/** - * Debounce window for mid-run `running` snapshots (CL-10667): coalesces the - * per-turn boundary write to ~0.2/turn at a ~1s/turn cadence, while the - * run-liveness heartbeat keeps mtime freshness over longer gaps. - */ -export const RUN_SNAPSHOT_DEBOUNCE_MS = 5_000; - -export interface RunSnapshotDebouncer { - /** Mark a pending write and arm the one-shot timer once. */ - schedule: () => void; - /** Cancel the timer and run any pending write immediately. */ - flush: () => Promise; -} - -/** - * Lazy payoff: the deferred write recomputes its snapshot when it fires, not - * at schedule time, so a front of rapid turns persists the state current at - * fire time. One timer; `schedule` while armed only re-marks dirty. The - * writer maps write failures to undefined so a stalled flush cannot wedge - * the terminal path that awaits it. - */ -export function createRunSnapshotDebouncer(args: { - write: () => Promise; - intervalMs?: number; -}): RunSnapshotDebouncer { - const intervalMs = args.intervalMs ?? RUN_SNAPSHOT_DEBOUNCE_MS; - let timer: ReturnType | null = null; - let dirty = false; - let inFlight: Promise = Promise.resolve(); - - const run = (): void => { - dirty = false; - inFlight = args.write().catch(() => undefined); - }; - - return { - schedule: () => { - dirty = true; - if (timer !== null) return; - timer = setTimeout(() => { - timer = null; - if (dirty) run(); - }, intervalMs); - timer.unref?.(); - }, - flush: async () => { - if (timer !== null) { - clearTimeout(timer); - timer = null; - } - if (dirty) run(); - await inFlight; - }, - }; -} - // Terminal-write path: clears the in-memory active-run handle (active-run.ts) // alongside the on-disk status so the two cannot drift. Non-terminal // ("running") snapshots use saveState directly. diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index 1f39040a4..af3754e0c 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -24,7 +24,6 @@ import { setAgentSourceUnlessClosed } from "../agent-source-sync.js"; import { billingIdentityFromSource } from "../../cost/session-cost.js"; import { createRunSummary, type RunSummary } from "../../session/hooks.js"; import { - createRunSnapshotDebouncer, finalizeRunState, saveState, type RunState, @@ -262,20 +261,10 @@ export function createRunPersistence( await writeRunSnapshot("running"); }; - // CL-10667: per-turn boundary snapshots coalesce behind the debounce - // window (lazy payoff — the deferred write recomputes state at fire time, - // so nothing scheduled goes stale). - // Observable exposes flush before they observe or supersede run.json. - const runSnapshotDebouncer = createRunSnapshotDebouncer({ - write: () => persistRunSnapshot("running"), - }); - return { writeRunSnapshot, persistRunSnapshot, heartbeatRunSnapshot, - scheduleRunSnapshot: runSnapshotDebouncer.schedule, - flushRunSnapshot: runSnapshotDebouncer.flush, }; } @@ -322,18 +311,11 @@ export async function createRunLifecycle( newSession: () => void; agentProxy: Agent; }> { - const { - persistRunSnapshot, - heartbeatRunSnapshot, - scheduleRunSnapshot, - flushRunSnapshot, - } = createRunPersistence(state, services); + const { persistRunSnapshot, heartbeatRunSnapshot } = createRunPersistence( + state, + services, + ); state.persistRunSnapshot = persistRunSnapshot; - // CL-10667: per-turn boundary snapshots schedule behind the debounce - // window; observable exposes (interrupt rebuild, rotation, finalize) flush - // the pending write before they observe or supersede run.json. - state.scheduleRunSnapshot = scheduleRunSnapshot; - state.flushRunSnapshot = flushRunSnapshot; const activateAndCommitWire = (names: string[]): void => { if (!services.activatedToolNames.activate(names)) return; if (services.flushPromotions()) { @@ -618,10 +600,6 @@ export async function createRunLifecycle( enqueue: enqueueOp, rebuild: async () => { try { - // CL-10667: run.json is observable the moment the rebuild starts - // reading store state, so the pending debounced snapshot flushes - // first. - await state.flushRunSnapshot?.(); // close() tears down stream consumers before the aborted cycle's // inference.error is delivered, so the recorder never sees a // terminal event. Dispose before that teardown salvages the buffer @@ -702,10 +680,6 @@ export async function createRunLifecycle( "stream drain during session-rotation teardown failed", ), ); - // CL-10667: the rotation terminal write supersets the pending - // snapshot for the outgoing session; flush so the debounced write - // cannot straddle the id repoint below. - await state.flushRunSnapshot?.(); await persistRunSnapshot( "done", { finishedAt: Date.now() }, @@ -789,9 +763,6 @@ export async function finalizeTUIRun( await hostOf(state).waitUntilExit(); state.stopRunHeartbeat?.(); delete state.stopRunHeartbeat; - // CL-10667: pending debounced snapshots flush ahead of the terminal write - // so a failed or dropped terminal write cannot leave run.json stale. - await state.flushRunSnapshot?.(); // Stop workers before awaiting the session-op tail so a hung enqueue cannot // delay abort/reap; persistence, hooks, and telemetry stay after stop. // Toolset dispose lives inside shutdownRuntime so quit, crash, and signals diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 0fb0fc368..86b5c46c8 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -176,8 +176,7 @@ export async function assembleTUISession( start.activeRunHandle.cacheWriteModel = `${event.data.source.provider}:${event.data.source.model}`; } } - // CL-10667: debounced — the write recomputes state when it fires. - state.scheduleRunSnapshot?.(); + void state.persistRunSnapshot?.("running"); }, hookEnabled: initialHookEnabled, onHookEvent: (event) => emitter.emit("hook", event), diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index c3393a911..941769d05 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -332,10 +332,6 @@ export interface RunnerState { extra?: SnapshotExtra, kind?: Exclude, ) => Promise; - /** CL-10667: debounced per-turn boundary snapshot writer. */ - scheduleRunSnapshot?: () => void; - /** CL-10667: flush any pending debounced snapshot before observing run.json. */ - flushRunSnapshot?: () => Promise; shutdownRuntime?: () => Promise; stopFleetReporting?: () => void; stopRunHeartbeat?: () => void; From 340bbe88d698192695f8cdeb6f1ec0b048638c95 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 16 Oct 2026 20:09:52 -0700 Subject: [PATCH 2/2] fix(tui): keep a snapshot write failure off the crash handler --- src/tui/runner/exit.test.ts | 60 +++++++++++++++++++++++++++++++++++++ src/tui/runner/exit.ts | 20 ++++++++++++- 2 files changed, 79 insertions(+), 1 deletion(-) diff --git a/src/tui/runner/exit.test.ts b/src/tui/runner/exit.test.ts index d7ac5b2e3..7427f3f0e 100644 --- a/src/tui/runner/exit.test.ts +++ b/src/tui/runner/exit.test.ts @@ -1,6 +1,7 @@ import { describe, expect, spyOn, test } from "bun:test"; import { EventEmitter } from "node:events"; import { AgentContextLockError, type Agent } from "@intx/agent"; +import { getLogger } from "@intx/log"; import type { InferenceSource } from "@intx/types/runtime"; import * as codexSession from "../../auth/codex/session.js"; @@ -17,6 +18,7 @@ import { } from "../../subagent/session-store.js"; import type { ReactorInboundEvent } from "@intx/types/runtime"; import { defined } from "../../../testkit/defined.js"; +import { LOG_NAMESPACE_ROOT } from "../../branding.js"; import { stubReactorCapabilities, stubReactorState, @@ -31,10 +33,12 @@ import { createDeliveryGeneration, createSessionOperationQueue, } from "../delivery-queue.js"; +import * as sessionState from "../../session/state.js"; import { agentRebuildFailure, closeAgentForRebuild, createRunLifecycle, + createRunPersistence, finalizeTUIRun, observeRecoveryAttempts, resetSessionForRotation, @@ -976,6 +980,62 @@ describe("rebuild close helpers", () => { ).toMatch(/restart/i); }); + test("a rejecting saveState on the turn-boundary snapshot does not emit unhandledRejection", async () => { + const save = spyOn(sessionState, "saveState").mockRejectedValue( + new Error("disk full"), + ); + const warn = spyOn(getLogger([LOG_NAMESPACE_ROOT, "tui"]), "warn"); + const state = { + runTaskTitle: "t", + liveSource: { id: "p", model: "m", credentialId: "cred" }, + startedAt: 1, + connectedMcpServers: [], + config: { cwd: "/tmp" }, + sessionId: "sess-boundary", + } as unknown as RunnerState; + const services = { + runSink: { getTurnCount: () => 2 }, + activatedToolNames: { list: () => [] }, + activeRunHandle: { + sessionId: "sess-boundary", + cwd: "/tmp", + task: "t", + startedAt: 1, + turnsUsed: 0, + }, + crashGuard: { isFinalized: () => false }, + } as unknown as RunnerServices; + state.persistRunSnapshot = createRunPersistence( + state, + services, + ).persistRunSnapshot; + + let unhandled: unknown = null; + const onUnhandledRejection = (reason: unknown): void => { + unhandled = reason; + }; + process.on("unhandledRejection", onUnhandledRejection); + try { + // session.ts onTurnBoundarySnapshot: fire-and-forget running snapshot. + void state.persistRunSnapshot?.("running"); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(unhandled).toBeNull(); + expect(save).toHaveBeenCalled(); + expect(warn).toHaveBeenCalledWith( + "saveState failed for session {sessionId} status={status}: {error}", + { + sessionId: "sess-boundary", + status: "running", + error: "disk full", + }, + ); + } finally { + process.off("unhandledRejection", onUnhandledRejection); + save.mockRestore(); + warn.mockRestore(); + } + }); + // Overlay accept/decline tests stub bump() inside resolveSuspended, so // deleting the interrupt-site bump would not fail them. Drive the interrupt // helper itself. diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index af3754e0c..4b8e4f123 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -56,6 +56,7 @@ import { liveAgent, recordRunError, runWhileAgentBusy, + sanitizeRunnerDiagnostic, type RunnerServices, type RunnerState, type SnapshotExtra, @@ -251,7 +252,24 @@ export function createRunPersistence( kind: Exclude = "progress", ): Promise => { if (services.crashGuard.isFinalized()) return; - await writeRunSnapshot(status, extra, kind); + try { + await writeRunSnapshot(status, extra, kind); + } catch (err: unknown) { + // Turn boundaries and other progress sites fire this with `void`. A + // rejecting saveState must not surface as unhandledRejection — the + // process crash handler treats that as fatal. + tuiLogger.warn( + "saveState failed for session {sessionId} status={status}: {error}", + { + sessionId: state.sessionId, + status, + error: sanitizeRunnerDiagnostic( + state, + err instanceof Error ? err.message : String(err), + ), + }, + ); + } }; // Heartbeat refreshes run.json mtime so a quiet live turn is not aged to