From 76e591ab04b8fdb90053c58c7a55ab9fa5148813 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 2 Oct 2026 10:03:43 -0500 Subject: [PATCH] fix(runner): bound Pi cold session admission at sixty seconds --- doc/architecture/runner-pi-capabilities.md | 25 +++++ .../runner-core/src/acpx_sidecar_transport.rs | 36 ++++++- .../tests/acpx_sidecar_transport.rs | 17 ++++ .../src/live/runnerd-codex-transport.test.ts | 69 +++++++++++++- .../src/live/runnerd-codex-transport.ts | 95 +++++++++++++------ .../test/pi-closed-startup.test.mjs | 23 +++-- 6 files changed, 227 insertions(+), 38 deletions(-) diff --git a/doc/architecture/runner-pi-capabilities.md b/doc/architecture/runner-pi-capabilities.md index 55aa795ce0b..260170dbe36 100644 --- a/doc/architecture/runner-pi-capabilities.md +++ b/doc/architecture/runner-pi-capabilities.md @@ -1148,3 +1148,28 @@ separately built and verified Linux provider pack in its runner image; running local setup does not install or qualify a remote image. Published-tar local and Daytona startup evidence must bind the final installation candidate, with no candidate qualification flags, before a production-readiness claim. + + +## Cold process admission budget + +Pi cold `session.open`, including process replacement and durable reopen, has a +60-second admission budget. The controller shares one deadline across its cold +open/recovery command and provider identity barriers (`session.open`, replacement +`run.attach`, and replacement `runner.drain`). A live adopted runner keeps the +ordinary 30-second waits. Rust gives only Pi's sidecar `session.open` request the +60-second budget; initialize, attach, turn start and other sidecar commands stay +at 30 seconds. Inner ACP handshake (30 seconds), post-admission verification +(8 seconds), cancellation (2 seconds), and close (7 seconds) are unchanged. + +Timeout still poisons and retires the Rust sidecar process group. Closing the +controller during its startup wait rejects that wait and uses the existing owned +process cleanup. Callers must await transport close in their cleanup path. +This is trusted controller timing policy, outside the immutable Pi wrapper/profile +closure; it does not change profile 13, runtime 1.0.0, or the explicit model. + +The credential-free closed-startup regression requires the real +`session_ensure_failed` rejection within 60 seconds, with no prompt, no provider +identity and confirmed cleanup. Its child watchdog is 80 seconds (20 seconds for +cleanup after admission), inside a 90-second test timeout. Failed evidence stays +in its reported temporary directory. These margins do not extend production +command deadlines or qualify a target platform without a fresh run. 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 e111ce23553..a15e0ff6038 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 @@ -78,6 +78,7 @@ pub struct AcpxSidecarEvent { pub struct AcpxSidecarTransport { process: SupervisedProcess, request_timeout: Duration, + session_open_timeout: Duration, next_request_id: u64, last_event_sequence: u64, buffered_events: VecDeque, @@ -86,6 +87,16 @@ pub struct AcpxSidecarTransport { poisoned: bool, } +// A fresh Pi process verifies and copies its native closure before ACP admission. +// Reopen/recovery uses the same path. Ordinary sidecar requests retain their bound. +fn session_open_timeout(agent: &str, ordinary: Duration) -> Duration { + if agent == "pi" { + Duration::from_secs(60) + } else { + ordinary + } +} + impl AcpxSidecarTransport { pub fn start(config: &AcpxSidecarTransportConfig) -> Result { Self::start_with_environment_keys(config, &[]) @@ -137,7 +148,9 @@ impl AcpxSidecarTransport { keys.push("PAPERCLIP_ACPX_CREDENTIAL_BINDING"); } keys.extend_from_slice(credential_keys); - Self::start_with_environment_keys(config, &keys) + let mut transport = Self::start_with_environment_keys(config, &keys)?; + transport.session_open_timeout = session_open_timeout(agent, config.request_timeout); + Ok(transport) } fn start_with_environment_keys( @@ -164,6 +177,7 @@ impl AcpxSidecarTransport { Ok(Self { process, request_timeout: config.request_timeout, + session_open_timeout: config.request_timeout, next_request_id: 1, last_event_sequence: 0, buffered_events: VecDeque::new(), @@ -173,6 +187,14 @@ impl AcpxSidecarTransport { }) } + fn command_timeout(&self, command: GeneratedAcpxSidecarCommand) -> Duration { + if command == GeneratedAcpxSidecarCommand::SessionOpen { + self.session_open_timeout + } else { + self.request_timeout + } + } + pub fn process_id(&self) -> u32 { self.process.id() } @@ -268,7 +290,8 @@ impl AcpxSidecarTransport { })?; self.next_request_id = request_id + 1; - let deadline = Instant::now() + self.request_timeout; + let timeout = self.command_timeout(command); + let deadline = Instant::now() + timeout; loop { let remaining = deadline.saturating_duration_since(Instant::now()); if remaining.is_zero() { @@ -767,6 +790,15 @@ fn response_error_classification(error: &ResponseError) -> &'static str { #[cfg(test)] mod tests { + #[test] + fn only_pi_cold_open_gets_the_longer_admission_budget() { + let ordinary = Duration::from_secs(30); + assert_eq!(session_open_timeout("pi", ordinary), Duration::from_secs(60)); + for agent in ["claude", "codex", "grok", "cursor", "copilot"] { + assert_eq!(session_open_timeout(agent, ordinary), ordinary); + } + } + use super::*; #[test] diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_transport.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_transport.rs index 59497b18795..3242b30a311 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_transport.rs @@ -300,3 +300,20 @@ fn assigned_gateway_binding_reaches_qualified_sidecar_without_unrelated_secrets( sidecar.shutdown().unwrap(); } } + +#[test] +fn pi_ordinary_request_timeout_still_retires_the_sidecar() { + let mut sidecar = AcpxSidecarTransport::start_for_agent(&AcpxSidecarTransportConfig { + command: PathBuf::from(env!("CARGO_BIN_EXE_fake-acpx-sidecar")), + args: vec!["--mode".to_owned(), "silent".to_owned()], + verified_launch: None, + request_timeout: Duration::from_millis(30), + shutdown_grace: Duration::from_millis(50), + }, "pi").unwrap(); + let started = Instant::now(); + let error = sidecar.request(GeneratedAcpxSidecarCommand::Initialize, json!({})).unwrap_err(); + assert!(error.to_string().contains("timed out")); + assert!(started.elapsed() < Duration::from_secs(1)); + assert!(sidecar.request(GeneratedAcpxSidecarCommand::SessionOpen, json!({})).unwrap_err().to_string().contains("unavailable")); + sidecar.shutdown().expect("timeout cleanup remains idempotent"); +} 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 81916edf59c..921babf1fca 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1,3 +1,4 @@ +import { coldAdmissionTimeoutMs, waitForRunnerCommand } from "./runnerd-codex-transport.js"; import { chmod, cp, @@ -22,7 +23,7 @@ import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; -import { expect, it, vi } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import type { ControlPlanePort } from "../contracts/control-plane-port.js"; import { bindAcpxAgentFiles } from "../drivers/acpx/agent-files-binding.js"; import type { NativeExecutionInputV1 } from "../contracts/native-execution.js"; @@ -7797,3 +7798,69 @@ it.each([undefined, "medium", "minimal", "xhigh", null])("rejects non-exact Pi t it("rejects Pi thinking settings on a different transport provider", () => { expect(() => createCapabilityRunnerdCodexTransport({ provider: "codex", piThinkingLevel: "low" })).toThrow(/only supported/); }); + + +describe("cold process admission budget", () => { + it("bounds Pi cold startup and recovery at 60 seconds", () => { + expect(coldAdmissionTimeoutMs("acpx", "pi", true)).toBe(60_000); + }); + it("preserves live adoption and every other provider at 30 seconds", () => { + expect(coldAdmissionTimeoutMs("acpx", "pi", false)).toBe(30_000); + for (const agent of ["codex", "claude", "cursor", "copilot", "grok", undefined]) { + expect(coldAdmissionTimeoutMs("acpx", agent, true)).toBe(30_000); + } + for (const provider of ["codex", "opencode", undefined]) { + expect(coldAdmissionTimeoutMs(provider, "pi", true)).toBe(30_000); + } + }); +}); + + +it("shares Pi's absolute cold deadline across recovery barriers and rejects a late ACK", async () => { + vi.useFakeTimers(); + try { + const deadline = Date.now() + coldAdmissionTimeoutMs("acpx", "pi", true); + let status = "pending"; + const wait = (type: string) => waitForRunnerCommand({ + type, deadline, abortOnClose: true, isClosed: () => false, + throwIfFailed: () => undefined, command: () => ({ status }), + runnerHasExited: async () => false, failureCode: () => "startup_failed", + }); + const attached = wait("run.attach"); + await vi.advanceTimersByTimeAsync(40_000); + status = "completed"; + await vi.advanceTimersByTimeAsync(10); + await attached; + status = "pending"; + const drain = wait("runner.drain"); + const rejection = expect(drain).rejects.toThrow("runner.drain timed out"); + await vi.advanceTimersByTimeAsync(20_000); + await rejection; + status = "completed"; + await expect(wait("session.open")).rejects.toThrow("session.open timed out"); + } finally { vi.useRealTimers(); } +}); + +it("aborts startup promptly on close while allowing the owned cleanup command to drain", async () => { + vi.useFakeTimers(); + try { + let closed = false; + let status = "pending"; + const input = { + deadline: Date.now() + 60_000, isClosed: () => closed, + throwIfFailed: () => undefined, command: () => ({ status }), + runnerHasExited: async () => false, failureCode: () => "startup_failed", + }; + const opening = waitForRunnerCommand({ ...input, type: "session.open", abortOnClose: true }); + const rejection = expect(opening).rejects.toThrow("closed while waiting for session.open"); + closed = true; + await vi.advanceTimersByTimeAsync(10); + await rejection; + // A late open ACK cannot revive the cancelled waiter, but cleanup still owns the process. + const drain = waitForRunnerCommand({ ...input, type: "runner.drain", abortOnClose: false }); + await vi.advanceTimersByTimeAsync(10); + status = "completed"; + await vi.advanceTimersByTimeAsync(10); + await drain; + } finally { vi.useRealTimers(); } +}); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 591d97a2f37..e2360cc4151 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -4843,6 +4843,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#controlPlaneCheckpoint = registration.checkpoint ?? null; this.#controlPlaneRelease = registration.release; } + const admissionDeadline = this.#coldAdmissionDeadline(); const handle = spawnRunner({ connection: registration?.connection ?? { mode: "connect", @@ -4897,9 +4898,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#evidence.runnerProcessGroupId = handle.processGroupId ?? null; this.#publish(); this.#pump = setInterval(() => this.#pumpEventsSafely(), 5); - await this.#waitCommand("run.prepare"); - await this.#waitCommand("session.open"); - await this.#waitForProviderIdentity(); + await this.#waitCommand("run.prepare", undefined, undefined, true); + await this.#waitCommand("session.open", undefined, admissionDeadline, true); + await this.#waitForProviderIdentity(undefined, admissionDeadline); this.#startupComplete = true; this.#diagnostic("runnerd authenticated to the durable PRP control plane"); return this.#openedThreadResponse(params); @@ -5463,6 +5464,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (warmRecovery) { await this.options.authorizeWarmTransitionRecovery?.("before_spawn"); } + // A replacement executor restores its cold provider before acknowledging recovery. + const admissionDeadline = this.#coldAdmissionDeadline(adoptedRunner === undefined); const handle = adoptedRunner ? null : spawnRunner({ @@ -5561,11 +5564,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { registration?.ready, ); } + // Only a replacement executor can be restoring a cold Pi process. Share + // one deadline across every recovery barrier; live adoption stays at 30 s. if (runAttachment) { - await this.#waitCommand("run.attach", runAttachment.commandId); + await this.#waitCommand("run.attach", runAttachment.commandId, admissionDeadline, true); } if (recoveryProbeCommandId !== null) { - await this.#waitCommand("runner.drain", recoveryProbeCommandId); + await this.#waitCommand("runner.drain", recoveryProbeCommandId, admissionDeadline, true); if (this.#checkpointProviderIdentityExpectation !== null) { // A replacement runner can restore the exact provider while its fresh // session.resumed event is compacted or delayed behind the completed @@ -5573,7 +5578,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // waiting only on the bounded event replay. The authenticated command // result is still checked against the exact database checkpoint, so a // missing or changed provider identity continues to fail closed. - const snapshot = await this.#commandResult("session.snapshot", {}); + const snapshot = await this.#commandResult("session.snapshot", {}, admissionDeadline); this.#confirmCheckpointProviderIdentity( snapshot, "authenticated recovery session.snapshot", @@ -5591,6 +5596,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { !this.#checkpointProviderIdentityConfirmed ? "session.resumed" : undefined, + admissionDeadline, ); this.#startupComplete = true; this.#diagnostic( @@ -5918,12 +5924,19 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { return result; } + #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; + } + async #waitForProviderIdentity( expectedEventType?: "harness.ready" | "session.started" | "session.resumed", + deadline = Date.now() + 30_000, ): Promise { - const deadline = Date.now() + 30_000; while (Date.now() < deadline) { this.#throwIfFailed(); + if (this.#closed) throw new Error("runnerd transport closed during provider startup"); this.#pumpEvents(); if ( this.#threadId.length > 0 && @@ -5945,28 +5958,20 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { type: string, commandId?: string, deadline = Date.now() + 30_000, + abortOnClose = false, ): Promise { - while (Date.now() < deadline) { - this.#throwIfFailed(); - const command = - commandId === undefined - ? this.#core?.store.state.commands.find( - (candidate) => candidate.type === type, - ) - : this.#core?.getCommand(commandId); - if (command?.status === "completed") return; - if (command !== undefined && command.status !== "pending") { - throw new Error( - `PRP command ${type} ${command.status}: ${JSON.stringify(command.result)}`, - ); - } - if (await this.#runnerHasExited()) - throw new Error(`runnerd exited while waiting for ${type}`); - await new Promise((resolveWait) => setTimeout(resolveWait, 10)); - } - throw new Error( - `${this.#startupComplete ? "provider_transport_failed" : this.#startupFailureCode}: PRP command ${type} timed out`, - ); + await waitForRunnerCommand({ + type, + deadline, + abortOnClose, + isClosed: () => this.#closed, + throwIfFailed: () => this.#throwIfFailed(), + command: () => commandId === undefined + ? this.#core?.store.state.commands.find((candidate) => candidate.type === type) + : this.#core?.getCommand(commandId), + runnerHasExited: () => this.#runnerHasExited(), + failureCode: () => this.#startupComplete ? "provider_transport_failed" : this.#startupFailureCode, + }); } @@ -6915,3 +6920,37 @@ export function resolveRunnerdCodexSkillInputs( }; }); } + +/** Controller admission budget; ordinary command and turn deadlines are independent. */ +export function coldAdmissionTimeoutMs(provider: string | undefined, agent: string | undefined, coldProcess: boolean): number { + return coldProcess && provider === "acpx" && agent === "pi" ? 60_000 : 30_000; +} + +/** Shared polling path keeps startup cancellation separate from owned cleanup commands. */ +export async function waitForRunnerCommand(input: { + type: string; + deadline: number; + abortOnClose: boolean; + isClosed: () => boolean; + throwIfFailed: () => void; + command: () => { status: string; result?: unknown } | undefined; + runnerHasExited: () => Promise; + failureCode: () => string; +}): Promise { + while (Date.now() < input.deadline) { + input.throwIfFailed(); + if (input.abortOnClose && input.isClosed()) { + throw new Error(`runnerd transport closed while waiting for ${input.type}`); + } + const command = input.command(); + if (command?.status === "completed") return; + if (command !== undefined && command.status !== "pending") { + throw new Error(`PRP command ${input.type} ${command.status}: ${JSON.stringify(command.result)}`); + } + if (await input.runnerHasExited()) { + throw new Error(`runnerd exited while waiting for ${input.type}`); + } + await new Promise((resolveWait) => setTimeout(resolveWait, 10)); + } + throw new Error(`${input.failureCode()}: PRP command ${input.type} timed out`); +} diff --git a/packages/paperclip-runner/test/pi-closed-startup.test.mjs b/packages/paperclip-runner/test/pi-closed-startup.test.mjs index f2908f2780e..378e166acf3 100644 --- a/packages/paperclip-runner/test/pi-closed-startup.test.mjs +++ b/packages/paperclip-runner/test/pi-closed-startup.test.mjs @@ -19,7 +19,7 @@ if (process.argv[2] === "--pi-no-key-probe") { environment: { PATH: "/usr/bin:/bin", LANG: "en_US.UTF-8" }, onEvidence: value => evidence.push(value), }); - let error = null; let settledMs; + let error = null; let closeError = null; let settledMs; try { await bundle.transport.request("initialize", { clientInfo: { name: "pi-closed-startup-regression", version: "1" } }); await bundle.transport.request("thread/start", { @@ -28,13 +28,17 @@ if (process.argv[2] === "--pi-no-key-probe") { permissions: "paperclip-runner-workspace-read-only", dynamicTools: [], }); } catch (failure) { error = String(failure); } - finally { settledMs = Math.round(performance.now() - started); await bundle.transport.close(); } - await writeFile(join(root, "report.json"), JSON.stringify({ error, settledMs, durationMs: Math.round(performance.now() - started), evidence }), { mode: 0o600 }); + finally { + settledMs = Math.round(performance.now() - started); + try { await bundle.transport.close(); } catch (failure) { closeError = String(failure); } + } + await writeFile(join(root, "report.json"), JSON.stringify({ error, closeError, settledMs, durationMs: Math.round(performance.now() - started), evidence }), { mode: 0o600 }); } else { const packageRoot = process.env.PAPERCLIP_TEST_PI_STARTUP_PACKAGE_ROOT; const daemon = process.env.PAPERCLIP_TEST_PI_STARTUP_RUNNER_BINARY; + // 60 s cold admission + 20 s cleanup watchdog + 10 s test supervision. test("closed Pi Runner startup rejects admission and closes without a prompt", { - skip: !packageRoot || !daemon, timeout: 60_000, + skip: !packageRoot || !daemon, timeout: 90_000, }, async () => { const root = await mkdtemp(join(tmpdir(), "pi-closed-startup-")); await mkdir(join(root, "home")); @@ -43,16 +47,17 @@ if (process.argv[2] === "--pi-no-key-probe") { env: { HOME: join(root, "home"), PATH: "/usr/bin:/bin", LANG: "en_US.UTF-8" }, stdio: ["ignore", "pipe", "pipe"], }); let stderr = ""; child.stdout.resume(); child.stderr.on("data", chunk => { stderr += chunk; }); - const timer = setTimeout(() => child.kill("SIGTERM"), 55_000); + const timer = setTimeout(() => child.kill("SIGTERM"), 80_000); try { const exitCode = await new Promise((resolve, reject) => { child.once("error", reject); child.once("close", resolve); }); assert.equal(exitCode, 0, stderr); } finally { clearTimeout(timer); } const report = JSON.parse(await readFile(join(root, "report.json"), "utf8")); + assert.equal(report.closeError, null, report.closeError); assert.match(report.error, /session\.open failed/); assert.match(report.error, /retryable=false, classification=session_ensure_failed/); assert.doesNotMatch(report.error, /timed out/); - assert.ok(report.settledMs < 30_000, `session.open took ${report.settledMs} ms`); + assert.ok(report.settledMs < 60_000, `session.open took ${report.settledMs} ms`); assert.ok(report.evidence.some(item => item.runnerExited === true && item.runnerExitCode === 0)); for (const item of report.evidence) assert.deepEqual(item.childEnvironmentKeys, ["LANG", "PATH"]); const state = JSON.parse(await readFile(join(root, "state/runner/acpx-provider-state.json"), "utf8")); @@ -60,6 +65,10 @@ if (process.argv[2] === "--pi-no-key-probe") { assert.equal(state.activeTurnId, null); assert.equal(state.identity, null); assert.equal(state.providerExitUnconfirmed, false); console.log(JSON.stringify({ settledMs: report.settledMs, durationMs: report.durationMs, credentials: "none", promptCalls: 0 })); - } finally { await rm(root, { recursive: true, force: true }); } + } catch (error) { + console.error(`Pi startup failure evidence retained at ${root}`); + throw error; + } + await rm(root, { recursive: true, force: true }); }); }