diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 88ebd100055..ab762e4d317 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -4,7 +4,7 @@ import { runnerLaunchRequirementsDigest, type RunnerProviderLaunchRequirements, } from "./runner-provider-launch.js"; -import { coldAdmissionTimeoutMs, waitForRunnerCommand } from "./runnerd-codex-transport.js"; +import { coldAdmissionTimeoutMs, RUNNER_STALL_THRESHOLD_MS, RunnerStallLedger, waitForRunnerCommand } from "./runnerd-codex-transport.js"; import { configuredEnvironmentProjection } from "../configured-environment.js"; import { chmod, cp, mkdir, mkdtemp, readFile, readdir, realpath, rename, rm, stat, truncate, writeFile } from "node:fs/promises"; import { createHash } from "node:crypto"; @@ -6703,6 +6703,109 @@ it("shares Pi's absolute cold deadline across recovery barriers and rejects a la } finally { vi.useRealTimers(); } }); +// Chaos w-20261111T011704-s9107/t01: the control plane was stopped (SIGSTOP, as in +// a host sleep) 1 s into the resume barrier, together with the Runner it waits on. +// On thaw, the wall clock had passed the 30 s deadline before either process could +// act, so recovery reported "runner.drain timed out" and spent a recovery attempt. +describe("admission waits do not count time this process was stopped", () => { + const pending = (clock: RunnerStallLedger | undefined, deadline: number, status: () => string, type = "runner.drain") => + waitForRunnerCommand({ + type, deadline, clock, abortOnClose: true, isClosed: () => false, + throwIfFailed: () => undefined, command: () => ({ status: status() }), + runnerHasExited: async () => false, failureCode: () => "runner_local_connect_failed", + }); + + it("completes a resume barrier that the Runner answers after a 44 s freeze", async () => { + vi.useFakeTimers(); + const clock = new RunnerStallLedger(); + const stop = clock.watch(); + try { + let status = "pending"; + const drain = pending(clock, clock.now() + 30_000, () => status); + await vi.advanceTimersByTimeAsync(1_000); + // Stopped for 44 s: the wall clock moves, but no timer of this process runs. + vi.setSystemTime(Date.now() + 44_000); + await vi.advanceTimersByTimeAsync(2_000); + status = "completed"; + await vi.advanceTimersByTimeAsync(10); + await expect(drain).resolves.toBeUndefined(); + } finally { stop(); vi.useRealTimers(); } + }); + + it("keeps the wall-clock timeout for a wait outside an admission", async () => { + vi.useFakeTimers(); + try { + const drain = pending(undefined, Date.now() + 30_000, () => "pending"); + const rejection = expect(drain).rejects.toThrow("runner_local_connect_failed: PRP command runner.drain timed out"); + await vi.advanceTimersByTimeAsync(1_000); + vi.setSystemTime(Date.now() + 44_000); + await vi.advanceTimersByTimeAsync(10); + await rejection; + } finally { vi.useRealTimers(); } + }); + + it("still times out a barrier that the Runner never answers after a freeze", async () => { + vi.useFakeTimers(); + const clock = new RunnerStallLedger(); + const stop = clock.watch(); + try { + const drain = pending(clock, clock.now() + 30_000, () => "pending"); + const rejection = expect(drain).rejects.toThrow("runner_local_connect_failed: PRP command runner.drain timed out"); + await vi.advanceTimersByTimeAsync(1_000); + vi.setSystemTime(Date.now() + 44_000); + // Running time reaches the deadline: 1 s before, the uncredited 5 s of the + // gap, then the rest of the 30 s budget. + await vi.advanceTimersByTimeAsync(30_000); + await rejection; + } finally { stop(); vi.useRealTimers(); } + }); + + it("credits a freeze outside any poll loop to the next barrier on the same deadline", async () => { + vi.useFakeTimers(); + const clock = new RunnerStallLedger(); + const stop = clock.watch(); + try { + const deadline = clock.now() + coldAdmissionTimeoutMs("acpx", "pi", true); + let status = "pending"; + const prepared = pending(clock, clock.now() + 30_000, () => status, "run.prepare"); + await vi.advanceTimersByTimeAsync(5_000); + status = "completed"; + await vi.advanceTimersByTimeAsync(10); + await prepared; + // Stopped for 70 s between barriers (a checkpoint flush, a spawn): only the ticker observes it. + vi.setSystemTime(Date.now() + 70_000); + await vi.advanceTimersByTimeAsync(1_000); + status = "pending"; + const opened = pending(clock, deadline, () => status, "session.open"); + await vi.advanceTimersByTimeAsync(20_000); + status = "completed"; + await vi.advanceTimersByTimeAsync(10); + await expect(opened).resolves.toBeUndefined(); + } finally { stop(); vi.useRealTimers(); } + }); + + it("credits only stalls, so a slow but running process still reaches its deadline", () => { + let wall = 0; + const clock = new RunnerStallLedger(() => wall); + const deadline = clock.now() + 30_000; + // Every observation is 4.9 s apart on a loaded host: that is running time. + while (clock.now() < deadline) wall += 4_900; + expect(wall).toBeGreaterThanOrEqual(30_000); + expect(wall).toBeLessThan(35_000); + expect(clock.creditedMs).toBe(0); + // A 6 s gap is credited only beyond the stall threshold, and only once. + wall += 6_000; + clock.now(); + clock.now(); + expect(clock.creditedMs).toBe(6_000 - RUNNER_STALL_THRESHOLD_MS); + // A wall-clock grace taken now moves onto the clock without the earlier + // credit, so it keeps exactly its own running-time budget. + const wallGrace = wall + 5_000; + const onClock = wallGrace - clock.creditedMs; + expect(onClock - clock.now()).toBe(5_000); + }); +}); + it("aborts startup promptly on close while allowing the owned cleanup command to drain", async () => { vi.useFakeTimers(); try { diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 3e38d1e2fb0..15cb01e41f8 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -4064,6 +4064,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #nextTraceDebugSequence = 1; #traceRehydrationSpoolOverflow = false; readonly #traceFrameIndex = new RunnerdTraceFrameIndex(); + /** Clock of the startup or recovery admission in progress, if any. */ + #admissionClock: RunnerStallLedger | null = null; #startupSteps: { begin(name: RunnerdStartupStep["name"]): void; /** Move on only from the named open step (later reconnects are not startup milestones). */ @@ -5967,7 +5969,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #start( params: Record, ): Promise> { - const work = () => this.#withStartupSteps("bootstrap", () => this.#startWork(params)); + const work = () => this.#withAdmissionClock(() => this.#withStartupSteps("bootstrap", () => this.#startWork(params))); return this.options.measureOperation ? this.options.measureOperation("runner.transport.cold_start", work) : work(); } @@ -6219,10 +6221,42 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } async #resume(): Promise { - const work = () => this.#withStartupSteps("resume", () => this.#resumeWork()); + const work = () => this.#withAdmissionClock(() => this.#withStartupSteps("resume", () => this.#resumeWork())); return this.options.measureOperation ? this.options.measureOperation("runner.transport.resume", work) : work(); } + /** + * One stopped-time clock per admission; a nested startup keeps the outer one. + * It covers the Runner command barriers and the provider-identity wait. The + * warm-transition and source-attachment reconnect polls keep their own + * wall-clock grace deadlines. + */ + async #withAdmissionClock(work: () => Promise): Promise { + if (this.#admissionClock !== null) return work(); + const clock = new RunnerStallLedger(); + const stopWatching = clock.watch(); + this.#admissionClock = clock; + try { + return await work(); + } finally { + stopWatching(); + if (this.#admissionClock === clock) this.#admissionClock = null; + } + } + + /** Wall time, or the admission clock while a startup or recovery runs. */ + #waitNow(): number { + return this.#admissionClock?.now() ?? Date.now(); + } + + /** Moves a wall-clock deadline onto the admission clock, without the stops already credited. */ + #onWaitClock(wallDeadline: number): number { + const clock = this.#admissionClock; + if (clock === null) return wallDeadline; + clock.now(); + return wallDeadline - clock.creditedMs; + } + async #resumeWork(): Promise { if (this.options.acpxCandidateAdmission?.agent === "antigravity") throw new Error("ACPX candidate durable runner recovery is not admitted"); this.#startupSteps?.begin("runner.runtime.stage"); @@ -6943,7 +6977,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#assertStartupOpen(); const admissionDeadline = this.#sourceAttachmentDeadline === null ? this.#coldAdmissionDeadline(adoptedRunner === undefined) - : Math.min(Date.now() + 30_000, this.#sourceAttachmentDeadline); + : Math.min(this.#waitNow() + 30_000, this.#onWaitClock(this.#sourceAttachmentDeadline)); // An adopted runner is already live: no launch step (a launch span means a fresh process). if (adoptedRunner) this.#startupSteps?.end(); else this.#startupSteps?.begin("runner.process.launch"); @@ -7690,21 +7724,21 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #coldAdmissionDeadline(coldProcess = true): number | undefined { const timeout = coldAdmissionTimeoutMs(this.options.provider, this.options.acpxAgent, coldProcess); // Preserve the existing independent waits for other providers and live adoption. - return timeout === 60_000 ? Date.now() + timeout : undefined; + return timeout === 60_000 ? this.#waitNow() + timeout : undefined; } async #waitForProviderIdentity( expectedEventType?: "harness.ready" | "session.started" | "session.resumed", - deadline = Date.now() + 30_000, + deadline = this.#waitNow() + 30_000, ): Promise { return this.options.measureOperation ? this.options.measureOperation("runner.transport.provider_identity", () => this.#waitForProviderIdentityWork(expectedEventType, deadline)) : this.#waitForProviderIdentityWork(expectedEventType, deadline); } async #waitForProviderIdentityWork( expectedEventType?: "harness.ready" | "session.started" | "session.resumed", - deadline = Date.now() + 30_000, + deadline = this.#waitNow() + 30_000, ): Promise { - while (Date.now() < deadline) { + while (this.#waitNow() < deadline) { this.#throwIfFailed(); if (this.#closed) throw new Error("runnerd transport closed during provider startup"); this.#pumpEvents(); @@ -7728,7 +7762,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #waitCommand( type: string, commandId?: string, - deadline = Date.now() + 30_000, + deadline = this.#waitNow() + 30_000, abortOnClose = false, ): Promise { return this.options.measureOperation ? this.options.measureOperation("runner.transport.command_ack", () => this.#waitCommandWork(type, commandId, deadline, abortOnClose)) : this.#waitCommandWork(type, commandId, deadline, abortOnClose); @@ -7737,12 +7771,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #waitCommandWork( type: string, commandId?: string, - deadline = Date.now() + 30_000, + deadline = this.#waitNow() + 30_000, abortOnClose = false, ): Promise { await waitForRunnerCommand({ type, deadline, + clock: this.#admissionClock, abortOnClose, isClosed: () => this.#closed, throwIfFailed: () => this.#throwIfFailed(), @@ -9165,10 +9200,61 @@ function closedAcpxRejection(message: unknown): string | null { ? message : null; } +/** + * A gap this long between two observations by a running process (its poll + * loops sleep 10 ms, its ticker 1 s) means the process was not running: + * SIGSTOP, host sleep, or a whole-process stall. + */ +export const RUNNER_STALL_THRESHOLD_MS = 5_000; +const RUNNER_STALL_TICK_MS = 1_000; + +/** + * The clock for one startup or recovery admission. Time this process spends + * stopped is not waiting time: the Runner it waits on is usually stopped with + * it (same host sleep or process tree), and could not answer. The wall clock + * still passes the deadline, so a thaw used to report a timeout before either + * process ran (chaos t01: "runner.drain timed out" 45 s into a 30 s wait that + * held a 44 s freeze). + * + * `now()` is wall time minus the stopped time observed since the ledger + * started. Poll loops and a 1 s ticker share one last-observation time, so a + * stop is credited once, wherever the process was when it stopped (a poll + * loop, a checkpoint flush, a spawn). Only the part of a gap beyond the + * threshold is credited, so a slow but running process still reaches its + * deadlines. Deadlines compared with `now()` must be taken from `now()`. + */ +export class RunnerStallLedger { + #creditedMs = 0; + #last: number; + constructor(private readonly wallNow: () => number = Date.now) { + this.#last = wallNow(); + } + + get creditedMs(): number { + return this.#creditedMs; + } + + now(): number { + const at = this.wallNow(); + if (at - this.#last > RUNNER_STALL_THRESHOLD_MS) this.#creditedMs += at - this.#last - RUNNER_STALL_THRESHOLD_MS; + this.#last = Math.max(this.#last, at); + return at - this.#creditedMs; + } + + /** Observes stops between poll loops; runs only while its admission runs. */ + watch(): () => void { + const ticker = setInterval(() => this.now(), RUNNER_STALL_TICK_MS); + ticker.unref?.(); + return () => clearInterval(ticker); + } +} + /** Shared polling path keeps startup cancellation separate from owned cleanup commands. */ export async function waitForRunnerCommand(input: { type: string; deadline: number; + /** Startup or recovery admission clock; `deadline` was taken from it. */ + clock?: RunnerStallLedger | null; abortOnClose: boolean; isClosed: () => boolean; throwIfFailed: () => void; @@ -9176,7 +9262,8 @@ export async function waitForRunnerCommand(input: { runnerHasExited: () => Promise; failureCode: () => string; }): Promise { - while (Date.now() < input.deadline) { + const now = () => input.clock ? input.clock.now() : Date.now(); + while (now() < input.deadline) { input.throwIfFailed(); if (input.abortOnClose && input.isClosed()) { throw new Error(`runnerd transport closed while waiting for ${input.type}`);