From fe841a96cb2d3b416cfa89aa5779d28aaac6fb71 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 16 Oct 2026 19:58:40 -0700 Subject: [PATCH] refactor(session): drop the run.json payload skip --- src/exec/runner.test.ts | 94 ------------------------- src/exec/runner.ts | 18 +---- src/tui/runner/exit.test.ts | 134 ------------------------------------ src/tui/runner/exit.ts | 30 ++------ 4 files changed, 6 insertions(+), 270 deletions(-) diff --git a/src/exec/runner.test.ts b/src/exec/runner.test.ts index 23701272a..50c821bf8 100644 --- a/src/exec/runner.test.ts +++ b/src/exec/runner.test.ts @@ -1,6 +1,5 @@ import { describe, expect, spyOn, test } from "bun:test"; import type { ToolCall } from "@intx/types/runtime"; -import { defined } from "../../testkit/defined.js"; import type { AgentTool } from "@intx/agent"; import { submitOutputDefinition } from "../agent/director.js"; import { DIRECTOR_REGISTRY } from "../agent/directors/registry.js"; @@ -565,99 +564,6 @@ describe("runExec", () => { cleanup(); } }); - - test("heartbeat forces identical-payload writes; a failed save does not take the baseline", async () => { - // CL-10667 review fix: the exec heartbeat tick must bypass the - // identical-payload skip (mtime freshness is a liveness contract read - // cross-process by ageStaleRunningState), and the baseline must record - // only successes so a transient save failure cannot freeze run.json. - const previous = getActiveRun(); - clearActiveRun(); - const sessionId = "exec-heartbeat-fail"; - const { cwd, home, cleanup } = createTempDirs( - "corbits-exec-heartbeat-cwd-", - "corbits-exec-heartbeat-home-", - ); - try { - const saveCalls: { ok: boolean; status?: string }[] = []; - await withMockedHomedir(home, () => - withMockedModuleDuring( - import.meta.resolve("../session/state.js"), - (real: typeof import("../session/state.js")) => - ({ - ...real, - saveState( - cwd: string, - sessionId: string, - state: import("../session/state.js").RunState, - dir?: string, - ) { - if (saveCalls.length === 0) { - saveCalls.push({ ok: false, status: state.status }); - // Transient failure on the first running write. - return Promise.reject(new Error("ENOSPC-like")); - } - saveCalls.push({ ok: true, status: state.status }); - return real.saveState(cwd, sessionId, state, dir); - }, - }) as typeof real, - () => - withMockedModuleDuring( - import.meta.resolve("../session/run-liveness.js"), - (real: typeof import("../session/run-liveness.js")) => - ({ - ...real, - RUN_HEARTBEAT_INTERVAL_MS: 10, - }) as typeof real, - () => - withMockedModuleDuring( - import.meta.resolve("../session/assemble-runtime.js"), - (inner: typeof import("../session/assemble-runtime.js")) => - ({ - ...inner, - assembleInferenceBase: () => - // Delay the bootstrap failure so heartbeat ticks run - // while the run is unsettled. - new Promise((_, reject) => { - setTimeout( - () => reject(new Error("bootstrap failed")), - 80, - ); - }), - }) as typeof inner, - async () => { - const { runExec: runExecUnderMock } = - await import("./runner.js"); - const result = await runExecUnderMock({ - ...bareConfig("do the thing"), - cwd, - sessionId, - }); - expect(result.status).toBe("failed"); - }, - ), - ), - ), - ); - - // Write 1 (running) failed; the heartbeat retried the identical payload - // and forced a disk write past any baseline; the terminal wrote last. - expect(saveCalls[0]).toEqual({ ok: false, status: "running" }); - const runningOk = saveCalls.findIndex((call) => call.ok); - expect(runningOk).toBeGreaterThan(0); - expect(defined(saveCalls[runningOk]).status).toBe("running"); - expect(defined(saveCalls[saveCalls.length - 1]).status).toBe("failed"); - const finalPersisted = await loadState(cwd, sessionId, home); - expect(finalPersisted).toMatchObject({ - kind: "ok", - state: { status: "failed" }, - }); - } finally { - if (previous !== null) setActiveRun(previous); - else clearActiveRun(); - cleanup(); - } - }); }); describe("resolveExecDirectorOverlay", () => { diff --git a/src/exec/runner.ts b/src/exec/runner.ts index 274f6d847..d7b80f24f 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -541,14 +541,9 @@ export async function runExec(config: Config): Promise { | ReturnType | undefined; - // CL-10667: last written running snapshot is the compare baseline for the - // identical-payload skip (terminal writes always land). - let lastProgressJson: string | null = null; - const persist = async ( status: "running" | "done" | "failed" | "cancelled", extra?: { error?: string }, - force = false, ): Promise => { if (finalized && status === "running") return; if (status !== "running") { @@ -583,12 +578,6 @@ export async function runExec(config: Config): Promise { ? { error: sanitizeExecDiagnostic(extra.error) } : {}), }; - const payloadJson = JSON.stringify(snapshot, null, 2); - // force = heartbeat tick: mtime freshness is the write's job even when - // the payload is byte-identical to the baseline. - if (status === "running" && !force && payloadJson === lastProgressJson) { - return; - } const write = status === "running" ? saveState(config.cwd, sessionId, snapshot).then(() => { @@ -599,9 +588,6 @@ export async function runExec(config: Config): Promise { : finalizeRunState(config.cwd, sessionId, snapshot); try { await write; - // Baseline only what actually landed: a transient save failure (next - // tick or boundary snapshot then retries the same payload). - lastProgressJson = payloadJson; } catch (err: unknown) { logger.warn( "saveState failed for session {sessionId} status={status}: {error}", @@ -620,9 +606,7 @@ export async function runExec(config: Config): Promise { }); stopHeartbeat = startRunHeartbeat({ shouldTick: () => !finalized, - // The heartbeat's job is mtime freshness, not payload change: force the - // write past the identical-payload skip. - tick: () => persist("running", undefined, true), + tick: () => persist("running"), }); try { diff --git a/src/tui/runner/exit.test.ts b/src/tui/runner/exit.test.ts index 261f4d0d1..d7ac5b2e3 100644 --- a/src/tui/runner/exit.test.ts +++ b/src/tui/runner/exit.test.ts @@ -27,11 +27,6 @@ import { withMockedModuleDuring, } from "../../../testkit/mock-module.js"; import { createTempDirs } from "../../../testkit/temporary-dirs.js"; -import fsSync from "node:fs"; -import { readFile, stat, utimes } from "node:fs/promises"; -import { sessionDir } from "../../session/index.js"; -import type { RunState } from "../../session/state.js"; -import { createRunPersistence } from "./exit.js"; import { createDeliveryGeneration, createSessionOperationQueue, @@ -1183,132 +1178,3 @@ describe("observeRecoveryAttempts fan-out", () => { expect(seat.reconnectRecovery.settle(reconnectAttempt)).not.toBeNull(); }); }); - -describe("createRunPersistence — identical-payload skip (CL-10667)", () => { - function persistenceStub(args: { - runSinkTurnCount: () => number; - cwd: string; - }): { - writeRunSnapshot: () => Promise; - heartbeatRunSnapshot: () => Promise; - persistOnce: () => Promise; - } { - const state = { - host: { waitUntilExit: async () => undefined }, - config: { cwd: args.cwd, task: "t" }, - sessionId: "s", - startedAt: 1, - runTaskTitle: "t", - connectedMcpServers: [], - liveSource: { id: "p", model: "m" }, - } as unknown as RunnerState; - const services = { - runSink: { getTurnCount: args.runSinkTurnCount }, - crashGuard: { markFinalized: () => undefined, isFinalized: () => false }, - activeRunHandle: { task: "", startedAt: 0, turnsUsed: 0, model: "" }, - activatedToolNames: { list: () => [] }, - } as unknown as RunnerServices; - const { writeRunSnapshot, persistRunSnapshot, heartbeatRunSnapshot } = - createRunPersistence(state, services); - return { - writeRunSnapshot: () => writeRunSnapshot("running"), - heartbeatRunSnapshot, - persistOnce: () => persistRunSnapshot("running"), - }; - } - - test("identical progress snapshots skip; heartbeat ticks bypass the skip", async () => { - const directories = createTempDirs( - "corbits-runpersist-cwd-", - "corbits-runpersist-home-", - ); - try { - let turnsUsed = 0; - await withMockedHomedir(directories.home, async () => { - const { heartbeatRunSnapshot, persistOnce } = persistenceStub({ - runSinkTurnCount: () => turnsUsed, - cwd: directories.cwd, - }); - const runPath = `${sessionDir(directories.cwd, "s")}/run.json`; - - await heartbeatRunSnapshot(); - const first = JSON.parse(await readFile(runPath, "utf8")); - expect(first.turnsUsed).toBe(0); - // Stamp a long-gone mtime: any rewrite would move it. - await utimes(runPath, new Date(9e11), new Date(9e11)); - - // Payload-identical non-heartbeat progress write: skipped, mtime - // untouched. - await persistOnce(); - expect((await stat(runPath)).mtimeMs).toBe(9e11); - - // Heartbeat tick with the SAME payload: forces the write — the mtime - // is what ageStaleRunningState reads cross-process, and a skipped - // heartbeat would age a live run to "interrupted". - await heartbeatRunSnapshot(); - expect((await stat(runPath)).mtimeMs).not.toBe(9e11); - - // A changed payload lands like before. - turnsUsed = 1; - await persistOnce(); - const changed = JSON.parse(await readFile(runPath, "utf8")); - expect(changed.turnsUsed).toBe(1); - }); - } finally { - await directories.cleanup(); - } - }); - - test("a failed save does not take the baseline; the identical retry rewrites", async () => { - const directories = createTempDirs( - "corbits-runpersist-fail-cwd-", - "corbits-runpersist-fail-home-", - ); - try { - let turnsUsed = 0; - await withMockedHomedir(directories.home, () => - withMockedModuleDuring( - import.meta.resolve("../../session/state.js"), - (real: typeof import("../../session/state.js")) => { - let failingFirst = true; - return { - ...real, - saveState( - cwd: string, - sessionId: string, - persisted: RunState, - home?: string, - ): Promise { - if (!failingFirst) - return real.saveState(cwd, sessionId, persisted, home); - failingFirst = false; - return Promise.reject(new Error("transient save failure")); - }, - }; - }, - async () => { - const { persistOnce } = persistenceStub({ - runSinkTurnCount: () => turnsUsed, - cwd: directories.cwd, - }); - // First attempt fails; the caller awaits the rejection. - await persistOnce().catch((err: unknown) => { - expect(String(err)).toMatch(/transient save failure/); - }); - const runPath = `${sessionDir(directories.cwd, "s")}/run.json`; - expect(fsSync.existsSync(runPath)).toBe(false); - - // Identical payload is still rewritten: the failed attempt - // never took the baseline. - await persistOnce(); - expect(JSON.parse(await readFile(runPath, "utf8")).turnsUsed).toBe( - 0, - ); - }, - ), - ); - } finally { - await directories.cleanup(); - } - }); -}); diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index 16a86b6db..1f39040a4 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -192,22 +192,15 @@ export function clearsActiveRun(kind: SnapshotKind): boolean { /** * The run.json snapshot writers. Pure over (state, services), so the * lifecycle and the finalize tail can each build their own instance. - * Exported for tests building the minimal (state, services) stub. */ export function createRunPersistence( state: RunnerState, services: RunnerServices, ) { - // Identical-payload skip (CL-10667): the last written snapshot is the - // compare baseline; an unchanged progress snapshot skips the atomicWrite - // (terminal kinds always write — the skipped path never clears the active - // run handle, which finalizeRunState does). - let lastProgressJson: string | null = null; const writeRunSnapshot = async ( status: SnapshotStatus, extra?: SnapshotExtra, kind: SnapshotKind = "progress", - force = false, ): Promise => { const task = state.runTaskTitle.trim().length > 0 @@ -238,20 +231,11 @@ export function createRunPersistence( : {}), ...extra, }; - const payloadJson = JSON.stringify(persisted, null, 2); - // force = heartbeat tick: mtime freshness is the write's job even when - // the payload is byte-identical to the baseline. - if (!force && kind === "progress" && payloadJson === lastProgressJson) { - return; - } if (clearsActiveRun(kind)) { await finalizeRunState(state.config.cwd, state.sessionId, persisted); } else { await saveState(state.config.cwd, state.sessionId, persisted); } - // Baseline only what landed: a failed save must not suppress the next - // attempt at the same payload. - lastProgressJson = payloadJson; }; // Progress snapshots fire unsequenced (model switch, MCP connect, turn @@ -271,18 +255,16 @@ export function createRunPersistence( await writeRunSnapshot(status, extra, kind); }; - // The run-liveness heartbeat must bypass the identical-payload skip: its - // payload is often byte-identical (a long quiet turn), but run.json's - // mtime is what `ageStaleRunningState` reads cross-process — a skipped - // heartbeat would age a live run to "interrupted" after two missed beats. + // Heartbeat refreshes run.json mtime so a quiet live turn is not aged to + // "interrupted" by ageStaleRunningState. const heartbeatRunSnapshot = async (): Promise => { if (services.crashGuard.isFinalized()) return; - await writeRunSnapshot("running", undefined, "progress", true); + await writeRunSnapshot("running"); }; // CL-10667: per-turn boundary snapshots coalesce behind the debounce - // window plus the identical-payload skip above (lazy payoff — the deferred - // write recomputes state at fire time, so nothing scheduled goes stale). + // 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"), @@ -365,8 +347,6 @@ export async function createRunLifecycle( }; const stopHeartbeat = startRunHeartbeat({ shouldTick: () => !services.crashGuard.isFinalized(), - // The heartbeat's job is mtime freshness, not payload change: force the - // write past the identical-payload skip (see heartbeatRunSnapshot). tick: () => heartbeatRunSnapshot(), }); state.stopRunHeartbeat = stopHeartbeat;