Skip to content

Commit fe841a9

Browse files
committed
refactor(session): drop the run.json payload skip
1 parent 510d686 commit fe841a9

4 files changed

Lines changed: 6 additions & 270 deletions

File tree

‎src/exec/runner.test.ts‎

Lines changed: 0 additions & 94 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
import { describe, expect, spyOn, test } from "bun:test";
22
import type { ToolCall } from "@intx/types/runtime";
3-
import { defined } from "../../testkit/defined.js";
43
import type { AgentTool } from "@intx/agent";
54
import { submitOutputDefinition } from "../agent/director.js";
65
import { DIRECTOR_REGISTRY } from "../agent/directors/registry.js";
@@ -565,99 +564,6 @@ describe("runExec", () => {
565564
cleanup();
566565
}
567566
});
568-
569-
test("heartbeat forces identical-payload writes; a failed save does not take the baseline", async () => {
570-
// CL-10667 review fix: the exec heartbeat tick must bypass the
571-
// identical-payload skip (mtime freshness is a liveness contract read
572-
// cross-process by ageStaleRunningState), and the baseline must record
573-
// only successes so a transient save failure cannot freeze run.json.
574-
const previous = getActiveRun();
575-
clearActiveRun();
576-
const sessionId = "exec-heartbeat-fail";
577-
const { cwd, home, cleanup } = createTempDirs(
578-
"corbits-exec-heartbeat-cwd-",
579-
"corbits-exec-heartbeat-home-",
580-
);
581-
try {
582-
const saveCalls: { ok: boolean; status?: string }[] = [];
583-
await withMockedHomedir(home, () =>
584-
withMockedModuleDuring(
585-
import.meta.resolve("../session/state.js"),
586-
(real: typeof import("../session/state.js")) =>
587-
({
588-
...real,
589-
saveState(
590-
cwd: string,
591-
sessionId: string,
592-
state: import("../session/state.js").RunState,
593-
dir?: string,
594-
) {
595-
if (saveCalls.length === 0) {
596-
saveCalls.push({ ok: false, status: state.status });
597-
// Transient failure on the first running write.
598-
return Promise.reject(new Error("ENOSPC-like"));
599-
}
600-
saveCalls.push({ ok: true, status: state.status });
601-
return real.saveState(cwd, sessionId, state, dir);
602-
},
603-
}) as typeof real,
604-
() =>
605-
withMockedModuleDuring(
606-
import.meta.resolve("../session/run-liveness.js"),
607-
(real: typeof import("../session/run-liveness.js")) =>
608-
({
609-
...real,
610-
RUN_HEARTBEAT_INTERVAL_MS: 10,
611-
}) as typeof real,
612-
() =>
613-
withMockedModuleDuring(
614-
import.meta.resolve("../session/assemble-runtime.js"),
615-
(inner: typeof import("../session/assemble-runtime.js")) =>
616-
({
617-
...inner,
618-
assembleInferenceBase: () =>
619-
// Delay the bootstrap failure so heartbeat ticks run
620-
// while the run is unsettled.
621-
new Promise<never>((_, reject) => {
622-
setTimeout(
623-
() => reject(new Error("bootstrap failed")),
624-
80,
625-
);
626-
}),
627-
}) as typeof inner,
628-
async () => {
629-
const { runExec: runExecUnderMock } =
630-
await import("./runner.js");
631-
const result = await runExecUnderMock({
632-
...bareConfig("do the thing"),
633-
cwd,
634-
sessionId,
635-
});
636-
expect(result.status).toBe("failed");
637-
},
638-
),
639-
),
640-
),
641-
);
642-
643-
// Write 1 (running) failed; the heartbeat retried the identical payload
644-
// and forced a disk write past any baseline; the terminal wrote last.
645-
expect(saveCalls[0]).toEqual({ ok: false, status: "running" });
646-
const runningOk = saveCalls.findIndex((call) => call.ok);
647-
expect(runningOk).toBeGreaterThan(0);
648-
expect(defined(saveCalls[runningOk]).status).toBe("running");
649-
expect(defined(saveCalls[saveCalls.length - 1]).status).toBe("failed");
650-
const finalPersisted = await loadState(cwd, sessionId, home);
651-
expect(finalPersisted).toMatchObject({
652-
kind: "ok",
653-
state: { status: "failed" },
654-
});
655-
} finally {
656-
if (previous !== null) setActiveRun(previous);
657-
else clearActiveRun();
658-
cleanup();
659-
}
660-
});
661567
});
662568

663569
describe("resolveExecDirectorOverlay", () => {

‎src/exec/runner.ts‎

Lines changed: 1 addition & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -541,14 +541,9 @@ export async function runExec(config: Config): Promise<ExecResult> {
541541
| ReturnType<typeof createRunSnapshotDebouncer>
542542
| undefined;
543543

544-
// CL-10667: last written running snapshot is the compare baseline for the
545-
// identical-payload skip (terminal writes always land).
546-
let lastProgressJson: string | null = null;
547-
548544
const persist = async (
549545
status: "running" | "done" | "failed" | "cancelled",
550546
extra?: { error?: string },
551-
force = false,
552547
): Promise<void> => {
553548
if (finalized && status === "running") return;
554549
if (status !== "running") {
@@ -583,12 +578,6 @@ export async function runExec(config: Config): Promise<ExecResult> {
583578
? { error: sanitizeExecDiagnostic(extra.error) }
584579
: {}),
585580
};
586-
const payloadJson = JSON.stringify(snapshot, null, 2);
587-
// force = heartbeat tick: mtime freshness is the write's job even when
588-
// the payload is byte-identical to the baseline.
589-
if (status === "running" && !force && payloadJson === lastProgressJson) {
590-
return;
591-
}
592581
const write =
593582
status === "running"
594583
? saveState(config.cwd, sessionId, snapshot).then(() => {
@@ -599,9 +588,6 @@ export async function runExec(config: Config): Promise<ExecResult> {
599588
: finalizeRunState(config.cwd, sessionId, snapshot);
600589
try {
601590
await write;
602-
// Baseline only what actually landed: a transient save failure (next
603-
// tick or boundary snapshot then retries the same payload).
604-
lastProgressJson = payloadJson;
605591
} catch (err: unknown) {
606592
logger.warn(
607593
"saveState failed for session {sessionId} status={status}: {error}",
@@ -620,9 +606,7 @@ export async function runExec(config: Config): Promise<ExecResult> {
620606
});
621607
stopHeartbeat = startRunHeartbeat({
622608
shouldTick: () => !finalized,
623-
// The heartbeat's job is mtime freshness, not payload change: force the
624-
// write past the identical-payload skip.
625-
tick: () => persist("running", undefined, true),
609+
tick: () => persist("running"),
626610
});
627611

628612
try {

‎src/tui/runner/exit.test.ts‎

Lines changed: 0 additions & 134 deletions
Original file line numberDiff line numberDiff line change
@@ -27,11 +27,6 @@ import {
2727
withMockedModuleDuring,
2828
} from "../../../testkit/mock-module.js";
2929
import { createTempDirs } from "../../../testkit/temporary-dirs.js";
30-
import fsSync from "node:fs";
31-
import { readFile, stat, utimes } from "node:fs/promises";
32-
import { sessionDir } from "../../session/index.js";
33-
import type { RunState } from "../../session/state.js";
34-
import { createRunPersistence } from "./exit.js";
3530
import {
3631
createDeliveryGeneration,
3732
createSessionOperationQueue,
@@ -1183,132 +1178,3 @@ describe("observeRecoveryAttempts fan-out", () => {
11831178
expect(seat.reconnectRecovery.settle(reconnectAttempt)).not.toBeNull();
11841179
});
11851180
});
1186-
1187-
describe("createRunPersistence — identical-payload skip (CL-10667)", () => {
1188-
function persistenceStub(args: {
1189-
runSinkTurnCount: () => number;
1190-
cwd: string;
1191-
}): {
1192-
writeRunSnapshot: () => Promise<void>;
1193-
heartbeatRunSnapshot: () => Promise<void>;
1194-
persistOnce: () => Promise<void>;
1195-
} {
1196-
const state = {
1197-
host: { waitUntilExit: async () => undefined },
1198-
config: { cwd: args.cwd, task: "t" },
1199-
sessionId: "s",
1200-
startedAt: 1,
1201-
runTaskTitle: "t",
1202-
connectedMcpServers: [],
1203-
liveSource: { id: "p", model: "m" },
1204-
} as unknown as RunnerState;
1205-
const services = {
1206-
runSink: { getTurnCount: args.runSinkTurnCount },
1207-
crashGuard: { markFinalized: () => undefined, isFinalized: () => false },
1208-
activeRunHandle: { task: "", startedAt: 0, turnsUsed: 0, model: "" },
1209-
activatedToolNames: { list: () => [] },
1210-
} as unknown as RunnerServices;
1211-
const { writeRunSnapshot, persistRunSnapshot, heartbeatRunSnapshot } =
1212-
createRunPersistence(state, services);
1213-
return {
1214-
writeRunSnapshot: () => writeRunSnapshot("running"),
1215-
heartbeatRunSnapshot,
1216-
persistOnce: () => persistRunSnapshot("running"),
1217-
};
1218-
}
1219-
1220-
test("identical progress snapshots skip; heartbeat ticks bypass the skip", async () => {
1221-
const directories = createTempDirs(
1222-
"corbits-runpersist-cwd-",
1223-
"corbits-runpersist-home-",
1224-
);
1225-
try {
1226-
let turnsUsed = 0;
1227-
await withMockedHomedir(directories.home, async () => {
1228-
const { heartbeatRunSnapshot, persistOnce } = persistenceStub({
1229-
runSinkTurnCount: () => turnsUsed,
1230-
cwd: directories.cwd,
1231-
});
1232-
const runPath = `${sessionDir(directories.cwd, "s")}/run.json`;
1233-
1234-
await heartbeatRunSnapshot();
1235-
const first = JSON.parse(await readFile(runPath, "utf8"));
1236-
expect(first.turnsUsed).toBe(0);
1237-
// Stamp a long-gone mtime: any rewrite would move it.
1238-
await utimes(runPath, new Date(9e11), new Date(9e11));
1239-
1240-
// Payload-identical non-heartbeat progress write: skipped, mtime
1241-
// untouched.
1242-
await persistOnce();
1243-
expect((await stat(runPath)).mtimeMs).toBe(9e11);
1244-
1245-
// Heartbeat tick with the SAME payload: forces the write — the mtime
1246-
// is what ageStaleRunningState reads cross-process, and a skipped
1247-
// heartbeat would age a live run to "interrupted".
1248-
await heartbeatRunSnapshot();
1249-
expect((await stat(runPath)).mtimeMs).not.toBe(9e11);
1250-
1251-
// A changed payload lands like before.
1252-
turnsUsed = 1;
1253-
await persistOnce();
1254-
const changed = JSON.parse(await readFile(runPath, "utf8"));
1255-
expect(changed.turnsUsed).toBe(1);
1256-
});
1257-
} finally {
1258-
await directories.cleanup();
1259-
}
1260-
});
1261-
1262-
test("a failed save does not take the baseline; the identical retry rewrites", async () => {
1263-
const directories = createTempDirs(
1264-
"corbits-runpersist-fail-cwd-",
1265-
"corbits-runpersist-fail-home-",
1266-
);
1267-
try {
1268-
let turnsUsed = 0;
1269-
await withMockedHomedir(directories.home, () =>
1270-
withMockedModuleDuring(
1271-
import.meta.resolve("../../session/state.js"),
1272-
(real: typeof import("../../session/state.js")) => {
1273-
let failingFirst = true;
1274-
return {
1275-
...real,
1276-
saveState(
1277-
cwd: string,
1278-
sessionId: string,
1279-
persisted: RunState,
1280-
home?: string,
1281-
): Promise<void> {
1282-
if (!failingFirst)
1283-
return real.saveState(cwd, sessionId, persisted, home);
1284-
failingFirst = false;
1285-
return Promise.reject(new Error("transient save failure"));
1286-
},
1287-
};
1288-
},
1289-
async () => {
1290-
const { persistOnce } = persistenceStub({
1291-
runSinkTurnCount: () => turnsUsed,
1292-
cwd: directories.cwd,
1293-
});
1294-
// First attempt fails; the caller awaits the rejection.
1295-
await persistOnce().catch((err: unknown) => {
1296-
expect(String(err)).toMatch(/transient save failure/);
1297-
});
1298-
const runPath = `${sessionDir(directories.cwd, "s")}/run.json`;
1299-
expect(fsSync.existsSync(runPath)).toBe(false);
1300-
1301-
// Identical payload is still rewritten: the failed attempt
1302-
// never took the baseline.
1303-
await persistOnce();
1304-
expect(JSON.parse(await readFile(runPath, "utf8")).turnsUsed).toBe(
1305-
0,
1306-
);
1307-
},
1308-
),
1309-
);
1310-
} finally {
1311-
await directories.cleanup();
1312-
}
1313-
});
1314-
});

‎src/tui/runner/exit.ts‎

Lines changed: 5 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -192,22 +192,15 @@ export function clearsActiveRun(kind: SnapshotKind): boolean {
192192
/**
193193
* The run.json snapshot writers. Pure over (state, services), so the
194194
* lifecycle and the finalize tail can each build their own instance.
195-
* Exported for tests building the minimal (state, services) stub.
196195
*/
197196
export function createRunPersistence(
198197
state: RunnerState,
199198
services: RunnerServices,
200199
) {
201-
// Identical-payload skip (CL-10667): the last written snapshot is the
202-
// compare baseline; an unchanged progress snapshot skips the atomicWrite
203-
// (terminal kinds always write — the skipped path never clears the active
204-
// run handle, which finalizeRunState does).
205-
let lastProgressJson: string | null = null;
206200
const writeRunSnapshot = async (
207201
status: SnapshotStatus,
208202
extra?: SnapshotExtra,
209203
kind: SnapshotKind = "progress",
210-
force = false,
211204
): Promise<void> => {
212205
const task =
213206
state.runTaskTitle.trim().length > 0
@@ -238,20 +231,11 @@ export function createRunPersistence(
238231
: {}),
239232
...extra,
240233
};
241-
const payloadJson = JSON.stringify(persisted, null, 2);
242-
// force = heartbeat tick: mtime freshness is the write's job even when
243-
// the payload is byte-identical to the baseline.
244-
if (!force && kind === "progress" && payloadJson === lastProgressJson) {
245-
return;
246-
}
247234
if (clearsActiveRun(kind)) {
248235
await finalizeRunState(state.config.cwd, state.sessionId, persisted);
249236
} else {
250237
await saveState(state.config.cwd, state.sessionId, persisted);
251238
}
252-
// Baseline only what landed: a failed save must not suppress the next
253-
// attempt at the same payload.
254-
lastProgressJson = payloadJson;
255239
};
256240

257241
// Progress snapshots fire unsequenced (model switch, MCP connect, turn
@@ -271,18 +255,16 @@ export function createRunPersistence(
271255
await writeRunSnapshot(status, extra, kind);
272256
};
273257

274-
// The run-liveness heartbeat must bypass the identical-payload skip: its
275-
// payload is often byte-identical (a long quiet turn), but run.json's
276-
// mtime is what `ageStaleRunningState` reads cross-process — a skipped
277-
// heartbeat would age a live run to "interrupted" after two missed beats.
258+
// Heartbeat refreshes run.json mtime so a quiet live turn is not aged to
259+
// "interrupted" by ageStaleRunningState.
278260
const heartbeatRunSnapshot = async (): Promise<void> => {
279261
if (services.crashGuard.isFinalized()) return;
280-
await writeRunSnapshot("running", undefined, "progress", true);
262+
await writeRunSnapshot("running");
281263
};
282264

283265
// CL-10667: per-turn boundary snapshots coalesce behind the debounce
284-
// window plus the identical-payload skip above (lazy payoff — the deferred
285-
// write recomputes state at fire time, so nothing scheduled goes stale).
266+
// window (lazy payoff — the deferred write recomputes state at fire time,
267+
// so nothing scheduled goes stale).
286268
// Observable exposes flush before they observe or supersede run.json.
287269
const runSnapshotDebouncer = createRunSnapshotDebouncer({
288270
write: () => persistRunSnapshot("running"),
@@ -365,8 +347,6 @@ export async function createRunLifecycle(
365347
};
366348
const stopHeartbeat = startRunHeartbeat({
367349
shouldTick: () => !services.crashGuard.isFinalized(),
368-
// The heartbeat's job is mtime freshness, not payload change: force the
369-
// write past the identical-payload skip (see heartbeatRunSnapshot).
370350
tick: () => heartbeatRunSnapshot(),
371351
});
372352
state.stopRunHeartbeat = stopHeartbeat;

0 commit comments

Comments
 (0)