From 7e1a841ba5d4403bba32d8deb6c15116c3e021ad Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 21 Sep 2026 09:33:24 -0500 Subject: [PATCH] fix: preserve the command journal during warm attachment waits Co-Authored-By: Paperclip --- doc/execution-semantics.md | 6 +++ .../src/live/runnerd-codex-transport.ts | 31 +++--------- .../live/warm-attachment-readiness.test.ts | 50 +++++++++++++++++++ .../src/live/warm-attachment-readiness.ts | 33 ++++++++++++ 4 files changed, 97 insertions(+), 23 deletions(-) create mode 100644 packages/paperclip-runner/src/live/warm-attachment-readiness.test.ts create mode 100644 packages/paperclip-runner/src/live/warm-attachment-readiness.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index a454fdefc76..d14ce0d70a3 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -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 diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index c244952fdfc..4b624890030 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -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((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 { diff --git a/packages/paperclip-runner/src/live/warm-attachment-readiness.test.ts b/packages/paperclip-runner/src/live/warm-attachment-readiness.test.ts new file mode 100644 index 00000000000..99707149a0a --- /dev/null +++ b/packages/paperclip-runner/src/live/warm-attachment-readiness.test.ts @@ -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); +}); diff --git a/packages/paperclip-runner/src/live/warm-attachment-readiness.ts b/packages/paperclip-runner/src/live/warm-attachment-readiness.ts new file mode 100644 index 00000000000..d45c7be8ba4 --- /dev/null +++ b/packages/paperclip-runner/src/live/warm-attachment-readiness.ts @@ -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; + snapshot: (deadline: number) => Promise>; + onBlocked?: (blockers: unknown) => void; +}): Promise { + 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(resolve => setTimeout(resolve, Math.min(delayMs, Math.max(0, deadline - Date.now())))); + } + throw new Error(`native_runner_warm_attachment_not_quiescent: ${JSON.stringify(lastBlockers)}`); +}