Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
12 changes: 1 addition & 11 deletions src/exec/runner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,6 @@ import {
clearActiveDisposeHost,
} from "../session/active-host.js";
import {
createRunSnapshotDebouncer,
finalizeRunState,
loadState,
saveState,
Expand Down Expand Up @@ -536,18 +535,13 @@ export async function runExec(config: Config): Promise<ExecResult> {
: {}),
};
let stopHeartbeat: (() => void) | undefined;
// CL-10667: debounced per-turn boundary snapshots; terminal exposes flush.
let runSnapshotDebouncer:
| ReturnType<typeof createRunSnapshotDebouncer>
| undefined;

const persist = async (
status: "running" | "done" | "failed" | "cancelled",
extra?: { error?: string },
): Promise<void> => {
if (finalized && status === "running") return;
if (status !== "running") {
await runSnapshotDebouncer?.flush();
finalized = true;
stopHeartbeat?.();
stopHeartbeat = undefined;
Expand Down Expand Up @@ -601,9 +595,6 @@ export async function runExec(config: Config): Promise<ExecResult> {
};

await persist("running");
runSnapshotDebouncer = createRunSnapshotDebouncer({
write: () => persist("running"),
});
stopHeartbeat = startRunHeartbeat({
shouldTick: () => !finalized,
tick: () => persist("running"),
Expand Down Expand Up @@ -1057,8 +1048,7 @@ export async function runExec(config: Config): Promise<ExecResult> {
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,
});
Expand Down
82 changes: 0 additions & 82 deletions src/session/run-snapshot-debounce.test.ts

This file was deleted.

56 changes: 0 additions & 56 deletions src/session/state.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<void>;
}

/**
* 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<void>;
intervalMs?: number;
}): RunSnapshotDebouncer {
const intervalMs = args.intervalMs ?? RUN_SNAPSHOT_DEBOUNCE_MS;
let timer: ReturnType<typeof setTimeout> | null = null;
let dirty = false;
let inFlight: Promise<void> = 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.
Expand Down
60 changes: 60 additions & 0 deletions src/tui/runner/exit.test.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand All @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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.
Expand Down
Loading
Loading