diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs index fc2ff4f575f..6ed53c42917 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs @@ -21,6 +21,46 @@ pub const ACPX_SIDECAR_MAX_FRAME_BYTES: usize = 1024 * 1024; const MAX_BUFFERED_EVENTS: usize = 512; const MAX_EVENT_POLL_TIMEOUT: Duration = Duration::from_secs(120); const MAX_JSON_SAFE_INTEGER: u64 = 9_007_199_254_740_991; +/// A response wait blocks in slices of at most this long. +const RESPONSE_WAIT_SLICE: Duration = Duration::from_secs(1); +/// A slice that returns this much later than it asked was not waiting: this +/// process was stopped (SIGSTOP, a frozen container, a whole-process stall). +const STOPPED_TIME_THRESHOLD: Duration = Duration::from_secs(2); +/// After a stop, the thawed sidecar keeps at least this long to answer. +const POST_STOP_GRACE: Duration = Duration::from_secs(5); +/// One request never credits more stopped time than this in total. +const MAX_STOPPED_CREDIT: Duration = Duration::from_secs(15 * 60); + +/// Moves `deadline` later after a wait slice that this process spent stopped. +/// +/// The sidecar is a child of this process and is normally stopped with it, so +/// it could not answer either. Without this, a thaw found the deadline already +/// past and failed the request before either process ran (chaos t05: a 61 s +/// freeze failed a turn start against its 30 s bound). The slice overshoot is +/// credited, and the deadline is at least `POST_STOP_GRACE` after the thaw, so +/// a stop in the last slice still leaves the sidecar time to answer. Total +/// credit is capped, so a process that never runs normally still times out. +/// `credited` carries the request's total credit between slices. +pub(crate) fn credit_stopped_time( + deadline: Instant, + slice: Duration, + slice_ended: Instant, + elapsed: Duration, + credited: &mut Duration, +) -> Instant { + let Some(stopped) = elapsed.checked_sub(slice) else { + return deadline; + }; + if stopped <= STOPPED_TIME_THRESHOLD || *credited >= MAX_STOPPED_CREDIT { + return deadline; + } + let extended = (deadline + stopped).max(slice_ended + POST_STOP_GRACE); + let credit = extended + .saturating_duration_since(deadline) + .min(MAX_STOPPED_CREDIT - *credited); + *credited += credit; + deadline + credit +} #[derive(Clone, Debug)] pub struct AcpxSidecarTransportConfig { @@ -710,13 +750,26 @@ impl AcpxSidecarTransport { request_id: u64, deadline: Instant, ) -> Result, LocalRunnerError> { + let mut deadline = deadline; + let mut credited = Duration::ZERO; loop { - let remaining = deadline.saturating_duration_since(Instant::now()); + let slice_started = Instant::now(); + let remaining = deadline.saturating_duration_since(slice_started); if remaining.is_zero() { return Ok(None); } - let Some(line) = self.receive_stdout_line(remaining, command.as_str())? else { - return Ok(None); + let slice = remaining.min(RESPONSE_WAIT_SLICE); + let received = self.receive_stdout_line(slice, command.as_str())?; + let slice_ended = Instant::now(); + deadline = credit_stopped_time( + deadline, + slice, + slice_ended, + slice_ended.saturating_duration_since(slice_started), + &mut credited, + ); + let Some(line) = received else { + continue; }; match parse_frame(&line)? { ParsedFrame::Event(event) => self.buffer_event(event)?, @@ -1287,6 +1340,61 @@ fn response_error_classification(error: &ResponseError) -> &'static str { #[cfg(test)] mod tests { + #[test] + fn credits_a_stop_and_leaves_the_sidecar_time_to_answer() { + let start = Instant::now(); + let deadline = start + Duration::from_secs(30); + let slice = Duration::from_secs(1); + let mut credited = Duration::ZERO; + // A slice that returned on time, or a little late on a loaded host. + for elapsed in [Duration::from_millis(400), Duration::from_millis(2_900)] { + assert_eq!( + credit_stopped_time(deadline, slice, start + elapsed, elapsed, &mut credited), + deadline + ); + } + assert_eq!(credited, Duration::ZERO); + // A 61 s freeze inside a 1 s slice early in the wait: 60 s are credited. + let ended = start + Duration::from_secs(61); + assert_eq!( + credit_stopped_time( + deadline, + slice, + ended, + Duration::from_secs(61), + &mut credited + ), + deadline + Duration::from_secs(60) + ); + // A freeze in the last 0.5 s slice still leaves the grace after thaw. + let mut credited = Duration::ZERO; + let last = start + Duration::from_millis(29_500); + let half = Duration::from_millis(500); + let ended = last + Duration::from_secs(60); + assert_eq!( + credit_stopped_time( + deadline, + half, + ended, + Duration::from_secs(60), + &mut credited + ), + ended + POST_STOP_GRACE + ); + // Total credit is capped. + let mut credited = MAX_STOPPED_CREDIT; + assert_eq!( + credit_stopped_time( + deadline, + slice, + ended, + Duration::from_secs(61), + &mut credited + ), + deadline + ); + } + #[test] #[cfg(unix)] fn a_timed_out_turn_start_never_establishes_a_resumable_rejection() { diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_stopped_time.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_stopped_time.rs new file mode 100644 index 00000000000..039cf654d64 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_stopped_time.rs @@ -0,0 +1,55 @@ +//! Runs alone in its own test binary: it stops this whole test process. +#![cfg(unix)] + +use std::path::PathBuf; +use std::time::{Duration, Instant}; + +use paperclip_runner_core::acpx_sidecar_transport::{ + AcpxSidecarTransport, AcpxSidecarTransportConfig, +}; +use paperclip_runner_core::generated_acpx_sidecar_contract::{ + GeneratedAcpxSidecarCommand, GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, +}; +use serde_json::json; + +/// Chaos w-20261111T011704-s9107/t05: the Runner and its ACPX sidecar were +/// stopped together (SIGSTOP, a host-sleep stand-in) past the sidecar request +/// bound. On thaw the bound had already passed, so the Runner failed the turn +/// start before the sidecar could answer. Here the sidecar stops this process +/// for 6 s against a 3 s bound and answers 0.5 s after the thaw: once early in +/// the wait, and once in its last half second. Both cases share one test so +/// no other test thread runs while this process is stopped. +#[test] +fn a_request_answered_after_this_process_was_stopped_past_its_bound_succeeds() { + for stop_after in ["0", "2.5"] { + answered_after_stop(stop_after); + } +} + +fn answered_after_stop(stop_after: &str) { + let response = json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "id": 1, "ok": true, "result": { "answered": true }, + }); + let runner = std::process::id(); + let script = format!( + "IFS= read -r request; sleep {stop_after}; \ + (sleep 6; kill -CONT {runner}; sleep 0.5; printf '%s\\n' '{response}') & \ + kill -STOP {runner}; wait; IFS= read -r next" + ); + let mut transport = AcpxSidecarTransport::start(&AcpxSidecarTransportConfig { + command: PathBuf::from("/bin/sh"), + args: vec!["-c".to_owned(), script], + verified_launch: None, + request_timeout: Duration::from_secs(3), + shutdown_grace: Duration::from_millis(100), + }) + .expect("sidecar should start"); + let started = Instant::now(); + let result = transport + .request(GeneratedAcpxSidecarCommand::Initialize, json!({})) + .expect("stopped time must not count against the request bound"); + assert_eq!(result["answered"], true); + assert!(started.elapsed() >= Duration::from_secs(6)); + transport.shutdown().expect("sidecar should stop"); +} diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-stall.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-stall.test.ts new file mode 100644 index 00000000000..7489002b66a --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-stall.test.ts @@ -0,0 +1,55 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, it, vi } from "vitest"; +import { createCapabilityRunnerdCodexTransport } from "./runnerd-codex-transport.js"; +import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js"; + +const signalGroup = (pid: number, signal: NodeJS.Signals) => { + try { process.kill(-pid, signal); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } +}; + +// Chaos w-20261111T011704-s9107/t05: the control plane and its Runner tree were +// stopped (SIGSTOP, a host-sleep stand-in) for 61 s while turn.start waited on its +// 30 s deadline. On thaw the wall clock had passed the deadline before either +// process could act, and the run reported "PRP command turn.start timed out". +// This drives the real transport and Runner: it fails if turn start does not run +// on the stopped-time clock, or if its waits do not read that clock. +it("starts a turn whose Runner and control plane were stopped past the turn-start deadline", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-turn-start-stall-")); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(root), + stateDirectory: root, + turnStartTimeoutMs: 30_000, + prpIdentity: { + runnerInstanceId: "stall-runner", environmentLeaseId: "stall-lease", runId: "stall-run", + normalizedSessionId: "stall-session", turnId: "stall-turn", itemId: "stall-item", + }, + }); + const realNow = Date.now.bind(Date); + let stoppedMs = 0; + let runnerPid: number | null = null; + try { + await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] }); + runnerPid = bundle.evidence().runnerPid; + expect(runnerPid).toEqual(expect.any(Number)); + // Stop the Runner tree, then let this process "sleep" 40 s: the wall clock + // jumps while no timer of this process runs. + signalGroup(runnerPid!, "SIGSTOP"); + vi.spyOn(Date, "now").mockImplementation(() => realNow() + stoppedMs); + const started = bundle.transport.request("turn/start", { input: [{ type: "text", text: "Continue after the freeze." }] }); + await new Promise(resolve => setTimeout(resolve, 300)); + stoppedMs = 40_000; + await new Promise(resolve => setTimeout(resolve, 300)); + signalGroup(runnerPid!, "SIGCONT"); + await expect(started).resolves.toMatchObject({ turn: { status: "inProgress" } }); + } finally { + vi.restoreAllMocks(); + if (runnerPid !== null) signalGroup(runnerPid, "SIGCONT"); + await bundle.transport.close().catch(() => undefined); + if (runnerPid !== null) signalGroup(runnerPid, "SIGKILL"); + await rm(root, { recursive: true, force: true }); + } +}, 30_000); 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 ab762e4d317..49fa481f4eb 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, RUNNER_STALL_THRESHOLD_MS, RunnerStallLedger, waitForRunnerCommand } from "./runnerd-codex-transport.js"; +import { coldAdmissionTimeoutMs, RUNNER_MAX_STOPPED_CREDIT_MS, RUNNER_POST_STOP_GRACE_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"; @@ -6753,13 +6753,29 @@ describe("admission waits do not count time this process was stopped", () => { 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); + // Running time reaches the deadline: 1 s before, then the rest of the + // 30 s budget plus the 5 s post-stop grace. + await vi.advanceTimersByTimeAsync(40_000); await rejection; } finally { stop(); vi.useRealTimers(); } }); + it("leaves the Runner time to answer after a freeze in a wait's last half second", 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(29_500); + vi.setSystemTime(Date.now() + 60_000); + await vi.advanceTimersByTimeAsync(2_000); + status = "completed"; + await vi.advanceTimersByTimeAsync(10); + await expect(drain).resolves.toBeUndefined(); + } 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(); @@ -6793,11 +6809,15 @@ describe("admission waits do not count time this process was stopped", () => { 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. + // A 6 s gap is a stop: credited in full plus the post-stop grace, and only once. wall += 6_000; clock.now(); clock.now(); - expect(clock.creditedMs).toBe(6_000 - RUNNER_STALL_THRESHOLD_MS); + expect(clock.creditedMs).toBe(6_000 + RUNNER_POST_STOP_GRACE_MS); + // Total credit is capped, so a process that never runs normally still times out. + wall += 2 * RUNNER_MAX_STOPPED_CREDIT_MS; + clock.now(); + expect(clock.creditedMs).toBe(RUNNER_MAX_STOPPED_CREDIT_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; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index b5ce4030a0c..e3e972a6a05 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -4071,7 +4071,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #nextTraceDebugSequence = 1; #traceRehydrationSpoolOverflow = false; readonly #traceFrameIndex = new RunnerdTraceFrameIndex(); - /** Clock of the startup or recovery admission in progress, if any. */ + /** Clock of the startup, recovery, or turn-start admission in progress, if any. */ #admissionClock: RunnerStallLedger | null = null; #startupSteps: { begin(name: RunnerdStartupStep["name"]): void; @@ -6234,10 +6234,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } /** - * 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. + * One stopped-time clock per admission (startup, recovery, or a turn start); + * a nested admission keeps the outer one. It covers the Runner command + * barriers, the provider-identity wait, and turn.start with its start-event + * and snapshot waits. 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(); @@ -6252,7 +6253,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } } - /** Wall time, or the admission clock while a startup or recovery runs. */ + /** Wall time, or the admission clock while a startup, recovery, or turn start runs. */ #waitNow(): number { return this.#admissionClock?.now() ?? Date.now(); } @@ -7348,7 +7349,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #startTurn( params: Record, ): Promise> { - return this.options.measureOperation ? this.options.measureOperation("runner.transport.turn_start", () => this.#startTurnWork(params), {inputBytes:Buffer.byteLength(JSON.stringify(params)), tokenDataAvailable:false}) : this.#startTurnWork(params); + // turn.start and its start-event and snapshot waits share one deadline, so + // they share one stopped-time clock (chaos t05: a 61 s freeze failed a + // 30 s turn.start before the Runner or its sidecar could answer). + const work = () => this.#withAdmissionClock(() => this.#startTurnWork(params)); + return this.options.measureOperation ? this.options.measureOperation("runner.transport.turn_start", work, {inputBytes:Buffer.byteLength(JSON.stringify(params)), tokenDataAvailable:false}) : work(); } /** @@ -7390,14 +7395,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (!Number.isSafeInteger(turnStartTimeoutMs) || turnStartTimeoutMs <= 0) { throw new Error("turnStartTimeoutMs must be a positive safe integer"); } - let commandDeadline = Date.now() + turnStartTimeoutMs; + let commandDeadline = this.#waitNow() + turnStartTimeoutMs; const heldCalls = this.#core; if (typeof heldCalls?.semanticToolResultsSettled === "function" && !heldCalls.semanticToolResultsSettled()) { let recovering = true; void Promise.resolve(heldCalls.recoverPendingSemanticInputs()) .catch(error => this.#failTransport(error instanceof Error ? error : new Error(String(error)))) .finally(() => { recovering = false; }); - while (recovering && Date.now() < commandDeadline) { + while (recovering && this.#waitNow() < commandDeadline) { this.#assertStartupOpen(); this.#throwIfFailed(); await new Promise((resolveWait) => setTimeout(resolveWait, 10)); @@ -7405,7 +7410,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#assertStartupOpen(); this.#throwIfFailed(); if (recovering) throw new Error("native_turn_start_held_tool_calls_unresolved: recovering an earlier turn's held Paperclip tool calls did not finish before the turn-start deadline"); - commandDeadline = Date.now() + turnStartTimeoutMs; + commandDeadline = this.#waitNow() + turnStartTimeoutMs; } const compactionTurn = params[COMPACTION_TURN] === true; const compactionPayload = compactionTurn ? runnerdCompactionTurnPayload(this.options.provider ?? "codex") : null; @@ -7507,7 +7512,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // Codex and OpenCode return their provider-assigned identity. const deadline = this.options.turnStartTimeoutMs === undefined - ? Date.now() + 30_000 + ? this.#waitNow() + 30_000 : commandDeadline; const providerTurnStarted = () => turnStartResponseReady({ @@ -7520,7 +7525,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // the start event early enough to read the runner's snapshot before it. const adoptionCheckAt = deadline - Math.min( TURN_START_ADOPTION_CHECK_RESERVE_MS, - Math.floor(Math.max(0, deadline - Date.now()) / 4), + Math.floor(Math.max(0, deadline - this.#waitNow()) / 4), ); // The runner can end the accepted turn before it starts. Its terminal is // held for the pending start, so report that real cause at once. @@ -7535,7 +7540,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { throw new Error(`runnerd failed the provider turn before it started: ${failure}`); }; const waitForStarted = async () => { - while (!providerTurnStarted() && Date.now() < adoptionCheckAt) { + while (!providerTurnStarted() && this.#waitNow() < adoptionCheckAt) { this.#throwIfFailed(); this.#pumpEvents(); if (providerTurnStarted()) break; @@ -7576,7 +7581,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } // Not provably live: keep the original contract and wait for the // start event until the configured bound, then fail as before. - while (!providerTurnStarted() && Date.now() < deadline) { + while (!providerTurnStarted() && this.#waitNow() < deadline) { this.#throwIfFailed(); this.#pumpEvents(); if (providerTurnStarted()) break; @@ -9295,6 +9300,10 @@ function closedAcpxRejection(message: unknown): string | null { */ export const RUNNER_STALL_THRESHOLD_MS = 5_000; const RUNNER_STALL_TICK_MS = 1_000; +/** After a stop, the thawed Runner keeps at least this long to answer. */ +export const RUNNER_POST_STOP_GRACE_MS = 5_000; +/** One admission never credits more stopped time than this in total. */ +export const RUNNER_MAX_STOPPED_CREDIT_MS = 15 * 60_000; /** * The clock for one startup or recovery admission. Time this process spends @@ -9307,9 +9316,11 @@ const RUNNER_STALL_TICK_MS = 1_000; * `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()`. + * loop, a checkpoint flush, a spawn). A stop is credited in full plus a + * grace, so a stop near a deadline still leaves the Runner time to answer + * after the thaw. Gaps below the threshold are running time, and total credit + * is capped, so a slow process still reaches its deadlines. Deadlines + * compared with `now()` must be taken from `now()`. */ export class RunnerStallLedger { #creditedMs = 0; @@ -9324,7 +9335,10 @@ export class RunnerStallLedger { now(): number { const at = this.wallNow(); - if (at - this.#last > RUNNER_STALL_THRESHOLD_MS) this.#creditedMs += at - this.#last - RUNNER_STALL_THRESHOLD_MS; + if (at - this.#last > RUNNER_STALL_THRESHOLD_MS) { + this.#creditedMs = Math.min(RUNNER_MAX_STOPPED_CREDIT_MS, + this.#creditedMs + at - this.#last + RUNNER_POST_STOP_GRACE_MS); + } this.#last = Math.max(this.#last, at); return at - this.#creditedMs; }