fix: preserve the command journal during warm attachment waits

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-09-21 09:33:24 -05:00
1 parent d56db90645
commit 7e1a841ba5
4 files changed
+97 -23

No files matched your search

+6
View File
@@ -1221,6 +1221,12 @@ session. A staged provider package is reused only after the complete expected
manifest and artifact hashes verify. A missing, changed, or incompatible package
must be replaced and verified before launch.
Warm attachment requires two consecutive authenticated readiness snapshots.
Blocked readiness probes back off within the reconnect deadline so they do not
fill the durable command journal while waiting. The fast ready path keeps its
short second barrier. If readiness never arrives, attachment fails closed with
the last observed blocker; a full journal is not a substitute for that diagnosis.
Safe native replacement may clear a Blocked status only with a durable receipt
that the same failed run projected that exact status version. Explicitly
reasserting Blocked or changing its blockers advances the status version, even
@@ -1,4 +1,5 @@
import { RunnerdTraceFrameIndex } from "./runnerd-trace-frame-index.js";
import { waitForWarmAttachmentReadiness } from "./warm-attachment-readiness.js";
import { codexExecutableReadOnlyRoots } from "../drivers/codex/codex-security-config.js";
import { isCanonicalProviderEventType } from "../provider-events.js";
import { execFileSync } from "node:child_process";
@@ -3670,34 +3671,18 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
// same budget here so a transient tunnel reconnect cannot trip the shorter
// generic command timeout and replace an otherwise healthy warm runner.
const reconnectGraceMs = this.options.runnerReconnectGraceMs ?? 5_000;
const deadline = Date.now() + reconnectGraceMs;
let consecutiveReadyProbes = 0;
let lastBlockers: unknown = null;
while (Date.now() < deadline) {
await this.#awaitWarmRunnerConnection(deadline);
const snapshot = await this.#commandResult(
await waitForWarmAttachmentReadiness({
graceMs: reconnectGraceMs,
waitForConnection: (deadline) => this.#awaitWarmRunnerConnection(deadline),
snapshot: (deadline) => this.#commandResult(
"session.snapshot",
{
quiesceForWarmAttach: true,
},
deadline,
);
lastBlockers = snapshot.warmAttachBlockers;
if (snapshot.warmAttachReady === true) {
consecutiveReadyProbes += 1;
// A second barrier prevents a provider frame emitted immediately after
// its terminal notification from racing the authority rotation. Each
// snapshot wakes runnerd, polls the provider, and drains the preceding
// durable event prefix before the next probe.
if (consecutiveReadyProbes >= 2) return;
} else {
consecutiveReadyProbes = 0;
}
await new Promise<void>((resolveWait) => setTimeout(resolveWait, 25));
}
throw new Error(
`native_runner_warm_attachment_not_quiescent: ${JSON.stringify(lastBlockers)}`,
);
),
onBlocked: (blockers) => this.#diagnostic(`warm attachment awaiting quiescence: ${JSON.stringify(blockers)}`),
});
}
async #awaitWarmRunnerConnection(deadline: number): Promise<void> {
@@ -0,0 +1,50 @@
import { afterEach, expect, it, vi } from "vitest";
import { waitForWarmAttachmentReadiness } from "./warm-attachment-readiness.js";
afterEach(() => vi.useRealTimers());
it("preserves the command journal while a remote provider takes a minute to settle", async () => {
vi.useFakeTimers();
vi.setSystemTime(0);
let commands = 12;
const snapshot = vi.fn(async () => {
if (++commands >= 500) throw new Error("Durable PRP command journal bound exceeded.");
return { warmAttachReady: Date.now() >= 60_000, warmAttachBlockers: Date.now() >= 60_000 ? [] : ["pending_messages"] };
});
const result = waitForWarmAttachmentReadiness({ graceMs: 120_000, waitForConnection: async () => {}, snapshot })
.then(() => "ready", error => String(error));
await vi.advanceTimersByTimeAsync(120_000);
expect(await result).toBe("ready");
expect(commands).toBeLessThan(200);
});
it("keeps the fast path and requires two consecutive ready snapshots", async () => {
vi.useFakeTimers();
const sequence = [true, false, true, true];
const snapshot = vi.fn(async () => ({ warmAttachReady: sequence.shift(), warmAttachBlockers: [] }));
const waiting = waitForWarmAttachmentReadiness({ graceMs: 5_000, waitForConnection: async () => {}, snapshot });
await vi.advanceTimersByTimeAsync(200);
await waiting;
expect(snapshot).toHaveBeenCalledTimes(4);
});
it("fails closed with the actual blocker when quiescence never arrives", async () => {
vi.useFakeTimers();
const snapshot = vi.fn(async () => ({ warmAttachReady: false, warmAttachBlockers: ["pending_runtime_requests"] }));
const result = waitForWarmAttachmentReadiness({ graceMs: 120_000, waitForConnection: async () => {}, snapshot })
.then(() => "unexpected-ready", error => String(error));
await vi.advanceTimersByTimeAsync(122_000);
expect(await result).toContain('native_runner_warm_attachment_not_quiescent: ["pending_runtime_requests"]');
expect(snapshot.mock.calls.length).toBeLessThan(200);
});
it("does not reset the probe budget when readiness alternates", async () => {
vi.useFakeTimers();
let calls = 0;
const snapshot = vi.fn(async () => ({ warmAttachReady: ++calls % 2 === 0, warmAttachBlockers: ["pending_events"] }));
const result = waitForWarmAttachmentReadiness({ graceMs: 120_000, waitForConnection: async () => {}, snapshot })
.then(() => "unexpected-ready", error => String(error));
await vi.advanceTimersByTimeAsync(122_000);
expect(await result).toContain("native_runner_warm_attachment_not_quiescent");
expect(calls).toBeLessThan(250);
});
@@ -0,0 +1,33 @@
/** Require two fresh authenticated barriers before rotating a warm run authority. */
export async function waitForWarmAttachmentReadiness(input: {
graceMs: number;
waitForConnection: (deadline: number) => Promise<void>;
snapshot: (deadline: number) => Promise<Record<string, unknown>>;
onBlocked?: (blockers: unknown) => void;
}): Promise<void> {
const deadline = Date.now() + input.graceMs;
let consecutiveReadyProbes = 0;
let lastBlockers: unknown = null;
let blockedDelayMs = 25;
// Each probe is a durable command. A 25 ms loop over the remote 120 second
// reconnect budget can exhaust its 500-command journal before that budget.
const maxBlockedDelayMs = Math.max(1_000, Math.ceil(input.graceMs / 100));
while (Date.now() < deadline) {
await input.waitForConnection(deadline);
const snapshot = await input.snapshot(deadline);
if (snapshot.warmAttachReady !== true && JSON.stringify(snapshot.warmAttachBlockers) !== JSON.stringify(lastBlockers)) {
input.onBlocked?.(snapshot.warmAttachBlockers);
}
lastBlockers = snapshot.warmAttachBlockers;
if (snapshot.warmAttachReady === true) {
consecutiveReadyProbes += 1;
if (consecutiveReadyProbes >= 2) return;
} else {
consecutiveReadyProbes = 0;
}
const delayMs = snapshot.warmAttachReady === true ? 25 : blockedDelayMs;
if (snapshot.warmAttachReady !== true) blockedDelayMs = Math.min(blockedDelayMs * 2, maxBlockedDelayMs);
await new Promise<void>(resolve => setTimeout(resolve, Math.min(delayMs, Math.max(0, deadline - Date.now()))));
}
throw new Error(`native_runner_warm_attachment_not_quiescent: ${JSON.stringify(lastBlockers)}`);
}