Skip to content

refactor(session): drop the run.json payload skip #1482

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheGreatAxios merged 1 commit into from
Oct 17, 2026
Merged

refactor(session): drop the run.json payload skip #1482

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 0 additions & 94 deletions src/exec/runner.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -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<never>((_, 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", () => {
Expand Down
18 changes: 1 addition & 17 deletions src/exec/runner.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -541,14 +541,9 @@ export async function runExec(config: Config): Promise<ExecResult> {
| ReturnType<typeof createRunSnapshotDebouncer>
| 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<void> => {
if (finalized && status === "running") return;
if (status !== "running") {
Expand Down Expand Up @@ -583,12 +578,6 @@ export async function runExec(config: Config): Promise<ExecResult> {
? { 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(() => {
Expand All @@ -599,9 +588,6 @@ export async function runExec(config: Config): Promise<ExecResult> {
: 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}",
Expand All @@ -620,9 +606,7 @@ export async function runExec(config: Config): Promise<ExecResult> {
});
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 {
Expand Down
134 changes: 0 additions & 134 deletions src/tui/runner/exit.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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<void>;
heartbeatRunSnapshot: () => Promise<void>;
persistOnce: () => Promise<void>;
} {
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<void> {
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();
}
});
});
30 changes: 5 additions & 25 deletions src/tui/runner/exit.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<void> => {
const task =
state.runTaskTitle.trim().length > 0
Expand Down Expand Up @@ -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
Expand All @@ -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<void> => {
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"),
Expand Down Expand Up @@ -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;
Expand Down
Loading