fix(runner): bound Pi cold session admission at sixty seconds

This commit is contained in:
Dotta committed 2026-10-02 10:08:39 -05:00
1 parent 47cfeb137b
commit 76e591ab04
6 files changed
+227 -38

No files matched your search

@@ -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.
@@ -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<AcpxSidecarEvent>,
@@ -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, LocalRunnerError> {
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]
@@ -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");
}
@@ -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(); }
});
@@ -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<void> {
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<void> {
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<boolean>;
failureCode: () => string;
}): Promise<void> {
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`);
}
@@ -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 });
});
}