diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b418585f432..7f74393e1f9 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2305,8 +2305,20 @@ existing bounded disposition-only recovery policy. A bootstrap retry requires durable evidence that no provider session, provider event, or checkpoint exists. A runner report that `session.open` failed or was -rejected qualifies, because that command runs before any prompt. An -indeterminate `session.open` outcome does not qualify. Bootstrap retries back +rejected qualifies, because that command runs before any prompt. A +`session.open` that the controller stopped waiting for during runner startup +(`: PRP command session.open timed out`) also qualifies +as `provider_session_open_timeout`: the controller sends a prompt only after +`session.open` completes, and closing the failed transport stops that runner +before the retry starts a new one. The runner can still open an empty provider +session after the controller stops waiting, so the retry starts from clean +state: a local runner root is quarantined only when the restart-recovery +bootstrap proof shows that no turn began and its processes are gone, and a +fresh ACPX open replaces a saved session record that holds no conversation. A +root that fails the proof, a remote runner, and a provider session or provider +event that became durable before the failure settled keep the state ambiguous. +A runner-reported indeterminate `session.open` outcome, and a timeout after +startup completed, do not qualify. Bootstrap retries back off: about 30 seconds after the first attempt and about 60 seconds after the second, with jitter that is persisted with the failure. Before it schedules a bootstrap retry, the executor looks for a known provider limit for the agent. diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 13c354a89c0..dee5e1b9bc4 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -403,7 +403,9 @@ and both must succeed before artifact metadata and source-bound hashes are recorded. Run Docker and native qualification on hosted CI rather than a local developer machine. -Pi cold provider admission has an absolute 60-second budget. Warm run attachment +Pi and Claude cold provider admission have an absolute 60-second budget, and +the runner's ACPX sidecar and the sidecar's provider session handshake bound their `session.open` at 60 seconds too. A cold +Claude `session.open` timed out at the earlier 30-second bound on a loaded host. Warm run attachment checkpoints the old ACPX sidecar and starts a new one, so it uses that same budget while retaining the existing Runner authority. Live adoption and ordinary commands retain their 30-second bounds. Closing the transport cancels admission; 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 07282d24adb..8db06a56c36 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 @@ -91,8 +91,10 @@ pub struct AcpxSidecarTransport { // 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. +// A cold Claude session.open exceeded 30 s on a loaded host; it shares Pi's +// 60 s bound, matching the controller's cold admission budget. fn session_open_timeout(agent: &str, ordinary: Duration) -> Duration { - if agent == "pi" { + if agent == "pi" || agent == "claude" { Duration::from_secs(60) } else { ordinary @@ -1202,13 +1204,15 @@ mod tests { } #[test] - fn only_pi_cold_open_gets_the_longer_admission_budget() { + fn only_pi_and_claude_cold_open_get_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"] { + for agent in ["pi", "claude"] { + assert_eq!( + session_open_timeout(agent, ordinary), + Duration::from_secs(60) + ); + } + for agent in ["codex", "grok", "cursor", "copilot"] { assert_eq!(session_open_timeout(agent, ordinary), ordinary); } } diff --git a/packages/paperclip-runner/src/drivers/acpx/abandoned-bootstrap-record.test.ts b/packages/paperclip-runner/src/drivers/acpx/abandoned-bootstrap-record.test.ts new file mode 100644 index 00000000000..dd501d90213 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/abandoned-bootstrap-record.test.ts @@ -0,0 +1,107 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createAcpRuntime, createAgentRegistry, createRuntimeStore } from "acpx/runtime"; +import { describe, expect, it } from "vitest"; +import { resetAbandonedBootstrapRecord } from "./codex-runtime-adapter.js"; + +// A controlled ACP peer that behaves like Claude: a session that never received +// a prompt was never saved, so session/load (and session/resume) of it fails. +const peer = String.raw` +const send = x => process.stdout.write(JSON.stringify({ jsonrpc: '2.0', ...x }) + '\n'); +let next = 0; +const prompted = new Set(JSON.parse(process.env.PROMPTED || '[]')); +require('node:readline').createInterface({ input: process.stdin }).on('line', line => { + const m = JSON.parse(line); + if (m.method === 'initialize') send({ id: m.id, result: { protocolVersion: 1, agentCapabilities: { loadSession: true }, authMethods: [] } }); + else if (m.method === 'session/new') send({ id: m.id, result: { sessionId: process.env.SESSION_PREFIX + (next++) } }); + else if (m.method === 'session/load') { + if (prompted.has(m.params.sessionId)) send({ id: m.id, result: {} }); + else send({ id: m.id, error: { code: -32002, message: 'No conversation found with session ID: ' + m.params.sessionId } }); + } + else if (m.method === 'session/prompt') send({ id: m.id, result: { stopReason: 'end_turn' } }); +});`; + +async function withRuntime(dir: string, prefix: string, run: (input: { + runtime: ReturnType; methods: string[]; +}) => Promise): Promise { + const children: ChildProcess[] = []; + const methods: string[] = []; + const runtime = createAcpRuntime({ + cwd: dir, + agentRegistry: createAgentRegistry({ overrides: { claude: "fixture" } }), + sessionStore: createRuntimeStore({ stateDir: join(dir, "state") }), + permissionMode: "deny-all", + onAcpMessage: (direction, value) => { if (direction === "outbound") methods.push((value as { method: string }).method); }, + spawnAgent: () => { + const child = spawn(process.execPath, ["-e", peer], { cwd: dir, env: { ...process.env, SESSION_PREFIX: prefix }, stdio: ["pipe", "pipe", "pipe"] }); + children.push(child); + return child; + }, + }); + try { + return await run({ runtime, methods }); + } finally { + for (const child of children) if (child.exitCode === null) child.kill("SIGKILL"); + } +} + +async function prompt(runtime: ReturnType, handle: Awaited["ensureSession"]>>) { + const turn = runtime.startTurn({ handle, text: "fixture", mode: "prompt", requestId: "first-turn", timeoutMs: 5_000 }); + const drain = (async () => { for await (const _event of turn.events) { /* drain */ } })(); + const result = await turn.result; + await drain; + return result; +} + +describe("a saved ACPX record from a bootstrap that stopped before any prompt", () => { + async function abandonBootstrap(dir: string) { + // The first bootstrap opens a session, then its runner stops before a turn. + await withRuntime(dir, "abandoned-", async ({ runtime, methods }) => { + const handle = await runtime.ensureSession({ sessionKey: "paperclip-session-key", agent: "claude", mode: "persistent", cwd: dir }); + expect(handle.backendSessionId).toBe("abandoned-0"); + expect(methods).toContain("session/new"); + await runtime.close({ handle, reason: "runner stopped before a turn" }); + }); + } + + it("is reused by a fresh open without a reset, so the first turn cannot load it", async () => { + const dir = await mkdtemp(join(tmpdir(), "acpx-abandoned-record-")); + try { + await abandonBootstrap(dir); + await withRuntime(dir, "retry-", async ({ runtime, methods }) => { + const handle = await runtime.ensureSession({ sessionKey: "paperclip-session-key", agent: "claude", mode: "persistent", cwd: dir }); + expect(handle.backendSessionId).toBe("abandoned-0"); + const result = await prompt(runtime, handle); + expect(result.status).toBe("failed"); + expect(JSON.stringify(result)).toMatch(/SESSION_RESUME_REQUIRED|could not be resumed|No conversation found/); + expect(methods).toContain("session/load"); + expect(methods).not.toContain("session/prompt"); + }); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + it("is replaced by a new provider session after the reset, and the first turn runs", async () => { + const dir = await mkdtemp(join(tmpdir(), "acpx-abandoned-record-")); + try { + await abandonBootstrap(dir); + expect(await resetAbandonedBootstrapRecord(createRuntimeStore({ stateDir: join(dir, "state") }), "paperclip-session-key")).toBe(true); + await withRuntime(dir, "retry-", async ({ runtime, methods }) => { + const handle = await runtime.ensureSession({ sessionKey: "paperclip-session-key", agent: "claude", mode: "persistent", cwd: dir }); + expect(handle.backendSessionId).toBe("retry-0"); + expect(await prompt(runtime, handle)).toMatchObject({ status: "completed" }); + expect(methods).toContain("session/new"); + expect(methods).not.toContain("session/load"); + expect(methods).toContain("session/prompt"); + await runtime.close({ handle, reason: "fixture cleanup" }); + }); + // The used session now holds a conversation, so a later reset leaves it alone. + expect(await resetAbandonedBootstrapRecord(createRuntimeStore({ stateDir: join(dir, "state") }), "paperclip-session-key")).toBe(false); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts index 467f2d11ee9..09af405e9b0 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts @@ -12,13 +12,14 @@ import type { AcpRuntime, AcpRuntimeHandle, AcpRuntimeOptions, + AcpSessionRecord, AcpSessionStore, } from "acpx/runtime"; import { decodeAcpxRuntimeHandleState } from "acpx/runtime"; import { describe, expect, it, vi } from "vitest"; import type { VerifiedAcpxCommandLease } from "./installation-integrity.js"; -import { claudeExitPlanModeProposal, openCodexAcpxRuntime } from "./codex-runtime-adapter.js"; +import { acpxSessionHandshakeTimeoutMs, claudeExitPlanModeProposal, openCodexAcpxRuntime, resetAbandonedBootstrapRecord } from "./codex-runtime-adapter.js"; import { cursorInstructionBinding } from "./cursor-instructions.js"; import { persistedAcpxTurnUsage } from "./usage-accounting.js"; import { createAcpxCommandLeaseOwner } from "./command-lease-owner.js"; @@ -2510,6 +2511,101 @@ describe("Codex ACPX runtime adapter", () => { } }); + it.each([ + { agent: "claude" as const, admitted: true }, + { agent: "codex" as const, admitted: false }, + ])("bounds a 45 s cold $agent session handshake by the agent's admission budget", async ({ agent, admitted }) => { + expect(acpxSessionHandshakeTimeoutMs(agent)).toBe(admitted ? 60_000 : 30_000); + // Pi shares Claude's budget; its admission needs Pi-only options, so only the bound is checked here. + expect(acpxSessionHandshakeTimeoutMs("pi")).toBe(60_000); + vi.useFakeTimers(); + try { + const child = fakeChild(); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child); + const runtime = fakeRuntime(); + const options = openOptions(command); + const opening = openCodexAcpxRuntime({ ...options, profile: { ...options.profile, agent } }, { + createRegistry: () => registry(), + createStore: () => store(), + awaitProviderOwnership: providerOwnershipEstablished, + awaitProviderExit: providerOwnershipEstablished, + createRuntime: (runtimeOptions) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + runtimeOptions.spawnAgent?.({ command: "ignored", args: ["--stdio"], options: {} }); + await new Promise((resolve) => setTimeout(resolve, 45_000)); + return HANDLE; + }); + return runtime; + }, + }); + const settled = opening.then(() => "admitted", (error: unknown) => error instanceof Error ? error.message : String(error)); + await vi.advanceTimersByTimeAsync(46_000); + const outcome = await settled; + if (admitted) expect(outcome).toBe("admitted"); + else expect(outcome).toContain("handshake exceeded its admission deadline"); + } finally { + vi.useRealTimers(); + } + }); + + describe("a record left by a bootstrap that stopped before any prompt", () => { + const savedRecord = (overrides: Partial = {}): AcpSessionRecord => ({ + schema: "acpx.session.v1", acpxRecordId: "provider-key", acpSessionId: "never-prompted", agentCommand: "agent", + cwd: "/workspace", name: "provider-key", createdAt: "2026-11-07T00:32:24.053Z", lastUsedAt: "2026-11-07T00:32:24.053Z", + lastSeq: 0, eventLog: {} as AcpSessionRecord["eventLog"], messages: [], updated_at: "2026-11-07T00:32:24.053Z", + cumulative_token_usage: {}, request_token_usage: {}, acpx: { current_model_id: "claude-sonnet-5" }, + ...overrides, + }) as AcpSessionRecord; + + it("asks ACPX for a new session instead of reusing a never-prompted record", async () => { + const sessions = store(); + vi.mocked(sessions.load).mockResolvedValue(savedRecord()); + expect(await resetAbandonedBootstrapRecord(sessions, "provider-key")).toBe(true); + expect(sessions.save).toHaveBeenCalledWith(expect.objectContaining({ + acpSessionId: "never-prompted", acpx: { current_model_id: "claude-sonnet-5", reset_on_next_ensure: true }, + })); + }); + + it.each([ + ["holds a conversation", { messages: [{ User: { id: "m", content: [] } }] as unknown as AcpSessionRecord["messages"] }], + ["recorded a prompt", { lastPromptAt: "2026-11-07T00:33:09.553Z" }], + ["recorded provider events", { lastSeq: 3 }], + ])("keeps a record that %s", async (_label, overrides) => { + const sessions = store(); + vi.mocked(sessions.load).mockResolvedValue(savedRecord(overrides)); + expect(await resetAbandonedBootstrapRecord(sessions, "provider-key")).toBe(false); + expect(sessions.save).not.toHaveBeenCalled(); + }); + + it.each([false, true])("resets it before a fresh open only (restoring=%s)", async (restoring) => { + const sessions = store(); + vi.mocked(sessions.load).mockResolvedValue(savedRecord()); + const child = fakeChild(); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child); + const runtime = fakeRuntime(); + const order: string[] = []; + vi.mocked(sessions.save).mockImplementation(async () => { order.push("save"); }); + const port = await openCodexAcpxRuntime({ ...openOptions(command), restoringSession: restoring }, { + createRegistry: () => registry(), + createStore: () => sessions, + awaitProviderOwnership: providerOwnershipEstablished, + awaitProviderExit: providerOwnershipEstablished, + createRuntime: (runtimeOptions) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + order.push("ensure"); + runtimeOptions.spawnAgent?.({ command: "ignored", args: ["--stdio"], options: {} }); + return HANDLE; + }); + return runtime; + }, + }); + expect(order).toEqual(restoring ? ["ensure"] : ["save", "ensure"]); + await port.close({ reason: "test complete" }); + }); + }); + it("terminates a provider spawned before the session handshake rejects", async () => { const child = fakeChild(); const command = fakeCommand(); diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts index 745736fde5a..987b6faa253 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts @@ -77,6 +77,15 @@ const activeCodexRuntimeCleanupOwners = new Set>(); // qualified runtime enough time to complete that local handshake. const SESSION_HANDSHAKE_TIMEOUT_MS = 30_000; +/** + * Pi and Claude get the same 60 s cold session.open bound as the runner's + * sidecar and the controller's cold admission budget. A cold Claude handshake + * exceeded 30 s on a loaded host; a shorter inner bound would fail first. + */ +export function acpxSessionHandshakeTimeoutMs(agent: string): number { + return agent === "pi" || agent === "claude" ? 60_000 : SESSION_HANDSHAKE_TIMEOUT_MS; +} + interface GoalAwareAcpRuntime extends AcpRuntime { requestExtension?(input: { handle: AcpRuntimeHandle; @@ -543,6 +552,7 @@ export async function openQualifiedAcpxRuntime( let handle: AcpRuntimeHandle | null = null; const ensuredSession = Promise.resolve() + .then(() => options.restoringSession ? undefined : resetAbandonedBootstrapRecord(baseStore, options.providerSessionKey)) .then(() => runtime.ensureSession({ sessionKey: options.providerSessionKey, @@ -647,7 +657,7 @@ export async function openQualifiedAcpxRuntime( try { const boundedHandshake = boundedSessionHandshake( handshake, - dependencies.sessionHandshakeTimeoutMs ?? SESSION_HANDSHAKE_TIMEOUT_MS, + dependencies.sessionHandshakeTimeoutMs ?? acpxSessionHandshakeTimeoutMs(options.profile.agent), ); handle = options.signal ? await raceRuntimeHandshakeWithAbort(boundedHandshake, options.signal) @@ -1116,6 +1126,24 @@ function nonEmptyRuntimeIdentity( return typeof value === "string" && value.length > 0 ? value : undefined; } +/** + * A fresh open has no runner-recorded provider identity, so a saved record for + * its session key was left by an earlier bootstrap of the same session that + * stopped before any prompt (for example a session.open the controller gave up + * on). Its provider session was never used and may not be resumable: Claude + * saves nothing until the first prompt. Ask ACPX to start a new session instead + * of reusing it. A record that holds any conversation is left unchanged. + */ +export async function resetAbandonedBootstrapRecord(store: AcpSessionStore, sessionKey: string): Promise { + const record = await store.load(sessionKey); + if (!record || record.acpx?.reset_on_next_ensure === true) return false; + // Only a record that provably holds no conversation qualifies. + if (!Array.isArray(record.messages) || record.messages.length > 0 || record.lastSeq !== 0 || + record.lastPromptAt !== undefined) return false; + await store.save({ ...record, acpx: { ...record.acpx, reset_on_next_ensure: true } }); + return true; +} + async function boundedSessionHandshake( handshake: Promise, timeoutMs: number, 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 b16668d84f1..6b229b66174 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -6183,13 +6183,17 @@ describe("cold process admission budget", () => { it("bounds Pi cold startup and recovery at 60 seconds", () => { expect(coldAdmissionTimeoutMs("acpx", "pi", true)).toBe(60_000); }); + it("bounds a cold ACPX Claude session.open at 60 seconds", () => { + expect(coldAdmissionTimeoutMs("acpx", "claude", true)).toBe(60_000); + expect(coldAdmissionTimeoutMs("acpx", "claude", false)).toBe(30_000); + }); it("bounds a cold OpenCode restore at 60 seconds", () => { expect(coldAdmissionTimeoutMs("opencode", undefined, true)).toBe(60_000); }); it("preserves live adoption and every other provider at 30 seconds", () => { expect(coldAdmissionTimeoutMs("acpx", "pi", false)).toBe(30_000); expect(coldAdmissionTimeoutMs("opencode", undefined, false)).toBe(30_000); - for (const agent of ["codex", "claude", "cursor", "copilot", "grok", undefined]) { + for (const agent of ["codex", "cursor", "copilot", "grok", undefined]) { expect(coldAdmissionTimeoutMs("acpx", agent, true)).toBe(30_000); } for (const provider of ["codex", "cursor_cloud", undefined]) { diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 23e28c7bd03..99313961fb3 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -8502,10 +8502,15 @@ export function resolveRunnerdCodexSkillInputs( }); } -/** Controller admission budget; ordinary command and turn deadlines are independent. */ +/** + * Controller admission budget; ordinary command and turn deadlines are independent. + * A cold ACPX Claude session.open timed out at the old 30 s bound on a loaded + * host (1-minute load 17-43), so it shares Pi's and OpenCode's 60 s cold budget. + * The runner's sidecar session.open bound matches. + */ export function coldAdmissionTimeoutMs(provider: string | undefined, agent: string | undefined, coldProcess: boolean): number { if (!coldProcess) return 30_000; - return (provider === "acpx" && agent === "pi") || provider === "opencode" ? 60_000 : 30_000; + return (provider === "acpx" && (agent === "pi" || agent === "claude")) || provider === "opencode" ? 60_000 : 30_000; } export interface RunnerCommandFailure extends Error { diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index d6dd1111d75..04eef87a7d0 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -566,6 +566,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti "provider_transport_failed", "native_runner_process_exited", "provider_session_open_rejected", + "provider_session_open_timeout", "local_runtime_api_listener_unavailable", ]; if ( diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 664e41b7ffb..0cfb4377f7d 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -19108,6 +19108,25 @@ it("classifies only a definite session.open outcome as a bounded pre-session fai expect(nativeSessionFailureSourceCode(new Error(`PRP command session.open indeterminate: ${modelDetail}`))) .toBe("native_session_interrupted"); }); +it("classifies a startup session.open timeout as a bounded transient startup failure", () => { + for (const startupCode of ["runner_local_connect_failed", "runner_direct_wss_failed", "runner_ingress_unavailable"]) { + const code = nativeSessionFailureSourceCode(new Error(`${startupCode}: PRP command session.open timed out`)); + expect(code).toBe("provider_session_open_timeout"); + expect(nativeSessionFailureDisposition(1, new Date(), code)).toMatchObject({ phase: "retryable_failure" }); + expect(nativeSessionFailureDisposition(3, new Date(), code)).toMatchObject({ + phase: "terminal_failure", failureCode: "native_session_retry_exhausted", nextAttemptAt: null, + }); + } + // After startup completed, or for any other command, the provider may have done work. + expect(nativeSessionFailureSourceCode(new Error("provider_transport_failed: PRP command session.open timed out"))) + .toBe("provider_transport_failed"); + for (const message of [ + "runner_local_connect_failed: PRP command turn.start timed out", + "runner_local_connect_failed: PRP command session.open timed out after turn", + ]) { + expect(nativeSessionFailureSourceCode(new Error(message))).toBe("native_session_interrupted"); + } +}); describe("HTTP Stop during remote runner staging", () => { it("does not dispatch after Stop and bounded bootstrap cleanup", async () => { const { createCapabilityRunnerdCodexTransport } = await import("@paperclipai/paperclip-runner/live"); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 3fdd6a830a7..9c1f1cb0732 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -2119,7 +2119,7 @@ function runnerdStateBase(): string { ); } -function scopedRunnerdStateRoot(execution: NativeExecutionInput): string { +export function scopedRunnerdStateRoot(execution: NativeExecutionInput): string { // Antigravity's first release owns a fresh per-turn daemon. Its native // conversation is stored separately and may load under each new run's grant. const scope = execution.provider.kind === "acpx" && execution.provider.agent === "antigravity" @@ -6075,6 +6075,25 @@ export async function restoreRemoteBootstrapCheckpointForRecovery(input: { await restoreNativeRunnerCheckpoint(input.db, input.execution.binding, input.root, nativeSessionKey(input.execution)); } +/** + * A session.open timeout leaves an indeterminate root: the runner may have + * opened a provider session after the controller stopped waiting. A same-run + * bootstrap retry may start only from an empty root. A local root that proves + * no turn began (the restart-recovery proof) is quarantined; any other root, + * and any remote runner, keeps the run fail-closed. + */ +export function prepareNativeSessionOpenTimeoutRetry(execution: NativeExecutionInput, localRunner: boolean): boolean { + if (!localRunner) return false; + const root = scopedRunnerdStateRoot(execution); + if (!lstatSync(root, { throwIfNoEntry: false })) return true; + if (!isSafeNativeStateDirectory(root)) return false; + const identity = readRunnerdDurableIdentity(root); + if (!identity || identity.runId !== execution.binding.runId) return false; + if (!runnerdStateProvesIncompleteBootstrap(root, { localProcesses: true })) return false; + quarantineRunnerdStateRoot(root, "identity_indeterminate"); + return true; +} + export function runnerdStateProvesIncompleteBootstrap( root: string, options: { localProcesses: boolean } = { localProcesses: false }, @@ -7455,6 +7474,7 @@ export function nativeSessionFailureSourceCode( | typeof NATIVE_CURRENT_WAKE_COMMENTS_UNREAD | typeof NATIVE_CURRENT_WAKE_COMMENTS_CHANGED_AFTER_READ | "provider_session_open_rejected" + | "provider_session_open_timeout" | "runner_ingress_access_rejected" | "runner_ingress_acquisition_failed" | "local_runtime_api_listener_unavailable" @@ -7562,6 +7582,14 @@ export function nativeSessionFailureSourceCode( } return "provider_session_open_rejected"; } + // The controller stopped waiting for session.open while runner startup was + // still incomplete: the transport reports its startup failure code only + // before startup completes, and session.open precedes every prompt. The + // outcome is unknown, so durable evidence (no provider event, no checkpoint) + // still decides whether a same-run bootstrap retry is safe. + if (/^(?:runner_local_connect_failed|runner_direct_wss_failed|runner_ingress_unavailable): PRP command session\.open timed out$/.test(message)) { + return "provider_session_open_timeout"; + } // This process has no HTTP listener (graceful shutdown closes it first), so // the control plane cannot hand a runner its callback origin. The lookup runs // before any runner or provider process is created. @@ -7684,6 +7712,7 @@ export async function nativeProviderRecoveryEvidence(input: { "provider_request_protocol_error", "native_runner_process_exited", "provider_session_open_rejected", + "provider_session_open_timeout", "local_runtime_api_listener_unavailable", ]).has(input.sourceFailureCode); if (!checkpointExists && !providerEventsExist && (definitelyPreSession @@ -10940,7 +10969,13 @@ async function executePaperclipNativeSessionWithinScope( throw error; } const now = new Date(); - const classifiedFailureCode = nativeSessionFailureSourceCode(error); + const startupClassifiedFailureCode = nativeSessionFailureSourceCode(error); + // A timed-out session.open may still have opened a provider session + // before its runner stopped. The same-run retry must not reuse it. + const classifiedFailureCode = startupClassifiedFailureCode === "provider_session_open_timeout" && + !prepareNativeSessionOpenTimeoutRetry(input.execution, input.runnerExecutionTarget?.kind !== "remote") + ? "native_session_interrupted" as const + : startupClassifiedFailureCode; const sourceFailureCode = classifiedFailureCode === "native_event_replay_conflict" ? classifiedFailureCode diff --git a/server/src/services/native-runtime/native-startup-retry.integration.test.ts b/server/src/services/native-runtime/native-startup-retry.integration.test.ts index e847c5d27f8..936a705a4c3 100644 --- a/server/src/services/native-runtime/native-startup-retry.integration.test.ts +++ b/server/src/services/native-runtime/native-startup-retry.integration.test.ts @@ -1,8 +1,9 @@ import { randomUUID } from "node:crypto"; -import { mkdtempSync, rmSync } from "node:fs"; +import { spawn } from "node:child_process"; +import { existsSync, mkdirSync, mkdtempSync, readdirSync, rmSync, writeFileSync } from "node:fs"; import { createServer } from "node:http"; import { tmpdir } from "node:os"; -import { join } from "node:path"; +import { basename, join } from "node:path"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { @@ -28,7 +29,7 @@ vi.mock("../../vendor/paperclip-runner/index.js", async (importOriginal) => ({ executeNativeSession: provider.execute, })); -import { executePaperclipNativeSession } from "./native-session-executor.js"; +import { executePaperclipNativeSession, prepareNativeSessionOpenTimeoutRetry, scopedRunnerdStateRoot } from "./native-session-executor.js"; import { buildNativeCompletionContract } from "./completion-contracts.js"; import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; @@ -480,6 +481,187 @@ describe("native provider startup retry", () => { expect(state.issue.status).toBe("in_progress"); }); + it.each(["runner_local_connect_failed", "runner_direct_wss_failed", "runner_ingress_unavailable"])( + "retries a %s session.open startup timeout on the same run within the bounded budget", + async (startupFailureCode) => { + const seeded = await seedRun(await createAgent()); + // The controller stopped waiting for session.open during startup, before + // any provider identity, event, checkpoint, or prompt existed. + const timeout = () => new Error(`${startupFailureCode}: PRP command session.open timed out`); + + const firstFailedAt = await failStartup(seeded, timeout()); + let state = await read(seeded); + expect(state.coordinator).toMatchObject({ + phase: "retryable_failure", + recoveryState: "resuming_session", + attempt: 1, + failureDetail: expect.objectContaining({ + originalFailureCode: "provider_session_open_timeout", + recoveryMode: "bootstrap_retry", + providerSessionEstablished: false, + providerEventsExist: false, + checkpointExists: false, + }), + }); + const [low, high] = backoffWindow(firstFailedAt, 1); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeGreaterThanOrEqual(low); + expect(state.coordinator.nextAttemptAt!.getTime()).toBeLessThanOrEqual(high + 5_000); + expect(state.issue.status).toBe("in_progress"); + expect(state.issue.executionRunId).toBe(seeded.runId); + expect(state.action).toMatchObject({ + ownerType: "agent", + wakePolicy: { kind: "resume_native_run", runId: seeded.runId }, + }); + + await failStartup(seeded, timeout(), { leaseOwner: await claimRetry(seeded) }); + state = await read(seeded); + expect(state.coordinator).toMatchObject({ phase: "retryable_failure", attempt: 2 }); + expect(state.issue.status).toBe("in_progress"); + + // At most two automatic attempts follow the first; then a person decides. + await failStartup(seeded, timeout(), { leaseOwner: await claimRetry(seeded) }); + state = await read(seeded); + expect(state.coordinator).toMatchObject({ + phase: "terminal_failure", + failureCode: "native_session_retry_exhausted", + attempt: 3, + nextAttemptAt: null, + failureDetail: expect.objectContaining({ originalFailureCode: "provider_session_open_timeout", recoveryMode: "bootstrap_retry" }), + }); + expect(state.issue.status).toBe("blocked"); + expect(state.action).toMatchObject({ ownerType: "board", cause: "native_session_retry_exhausted", wakePolicy: null }); + }, + ); + + it.each([ + { name: "the provider session started before the timeout settled", eventType: "session.started", message: "runner_local_connect_failed: PRP command session.open timed out" }, + { name: "a provider turn started", eventType: "turn.started", message: "runner_local_connect_failed: PRP command session.open timed out" }, + { name: "the timeout came after startup completed", eventType: null, message: "provider_transport_failed: PRP command session.open timed out" }, + { name: "a different startup command timed out", eventType: null, message: "runner_local_connect_failed: PRP command turn.start timed out" }, + ])("never retries a timeout automatically when $name", async ({ eventType, message }) => { + const seeded = await seedRun(await createAgent()); + provider.execute.mockReset().mockImplementation(async () => { + if (eventType) { + await db.insert(heartbeatRunEvents).values({ + companyId, runId: seeded.runId, agentId: seeded.agentId, seq: 900, eventType, stream: "system", + }); + } + throw new Error(message); + }); + await expect(executePaperclipNativeSession({ + db, + execution: seeded.execution, + runnerInstanceId: seeded.runnerInstanceId, + backend: { + descriptor: async () => { throw new Error("Unexpected provider descriptor call"); }, + openSession: async () => { throw new Error("Unexpected provider session call"); }, + }, + })).rejects.toThrow("timed out"); + + const state = await read(seeded); + expect(state.coordinator).toMatchObject({ + phase: "terminal_failure", + nextAttemptAt: null, + failureDetail: expect.objectContaining({ recoveryMode: "ambiguous_state" }), + }); + expect(state.issue.status).toBe("blocked"); + expect(await claimNativeSessionResumptions({ + db, + runnerInstanceId: "startup-retry-test", + runIds: [seeded.runId], + now: new Date(Date.now() + 60 * 60_000), + deferIneligible: false, + })).toEqual([]); + }); + + describe("state left by a timed-out session.open", () => { + let stateBase: string; + let previousStateDir: string | undefined; + beforeAll(() => { + previousStateDir = process.env.PAPERCLIP_RUNNER_STATE_DIR; + stateBase = mkdtempSync(join(tmpdir(), "native-startup-retry-state-")); + process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase; + }); + afterAll(() => { + if (previousStateDir === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previousStateDir; + rmSync(stateBase, { recursive: true, force: true }); + }); + + const deadPid = async () => { + const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + await new Promise(resolveExit => child.once("exit", resolveExit)); + return child.pid!; + }; + const event = (sourceSeq: number, eventType: string, payload: Record = {}) => ({ + sourceSeq, eventType, priority: 0, envelope: { kind: "event", payload: { eventType, payload } }, + }); + /** The runner opened a fresh provider session after the controller stopped waiting, then stopped. */ + async function writeOpenedSessionRoot(seeded: Seeded, options: { turnIssued?: boolean } = {}) { + const pid = await deadPid(); + const root = scopedRunnerdStateRoot(seeded.execution); + mkdirSync(join(root, "control-plane"), { recursive: true }); + mkdirSync(join(root, "runner"), { recursive: true }); + writeFileSync(join(root, "control-plane", "control-plane-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.control-plane-state.v2", + identity: { runId: seeded.runId, normalizedSessionId: seeded.execution.binding.runId }, + bootstrapHistory: { schema: "paperclip.runner.bootstrap-history.v1", credentialsIssued: 1, adopted: false }, + commands: [ + { type: "run.prepare", status: "completed", result: { result: { status: "prepared" } } }, + { type: "session.open", status: "completed", result: { commandType: "session.open", status: "completed", + result: { driver: "acpx_runtime", processId: pid, provider: "acpx", status: "started" } } }, + { type: "runner.drain", status: "completed" }, + { type: "runner.suspend", status: "completed" }, + ...(options.turnIssued ? [{ type: "turn.start", status: "pending" }] : []), + ], + committedEvents: [event(1, "session.started", { processId: pid }), event(2, "session.capabilities.updated")], + ackedSourceSeq: 2, + connectionCount: 1, + })); + writeFileSync(join(root, "runner", "runner-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.state.v1", lifecycle: "suspended", nextSourceSeq: 3, ackedSourceSeq: 2, outbox: [], + processedCommands: { + command_prp_00000001: { commandType: "run.prepare", status: "completed" }, + command_prp_00000002: { commandType: "session.open", status: "completed", result: { processId: pid, status: "started" } }, + }, + })); + return root; + } + const timeout = () => new Error("runner_local_connect_failed: PRP command session.open timed out"); + + it("quarantines an opened session that no turn used, so the same-run retry starts fresh", async () => { + const seeded = await seedRun(await createAgent()); + const root = await writeOpenedSessionRoot(seeded); + await failStartup(seeded, timeout()); + const state = await read(seeded); + expect(state.coordinator).toMatchObject({ + phase: "retryable_failure", + failureDetail: expect.objectContaining({ originalFailureCode: "provider_session_open_timeout", recoveryMode: "bootstrap_retry" }), + }); + expect(existsSync(root)).toBe(false); + expect(readdirSync(join(stateBase, "quarantine")).some(name => name.startsWith(`${basename(root)}.identity_indeterminate.`))).toBe(true); + }); + + it("keeps the run operator-owned and the root intact when a turn may have started", async () => { + const seeded = await seedRun(await createAgent()); + const root = await writeOpenedSessionRoot(seeded, { turnIssued: true }); + await failStartup(seeded, timeout()); + const state = await read(seeded); + expect(state.coordinator).toMatchObject({ + phase: "terminal_failure", + nextAttemptAt: null, + failureDetail: expect.objectContaining({ originalFailureCode: "native_session_interrupted", recoveryMode: "ambiguous_state" }), + }); + expect(existsSync(root)).toBe(true); + }); + + it("never prepares a remote runner's timeout for an automatic retry", async () => { + const seeded = await seedRun(await createAgent()); + expect(prepareNativeSessionOpenTimeoutRetry(seeded.execution, false)).toBe(false); + expect(prepareNativeSessionOpenTimeoutRetry(seeded.execution, true)).toBe(true); + }); + }); + it("retries startup that a draining server interrupted before any provider session", async () => { const seeded = await seedRun(await createAgent()); // A graceful shutdown closes the HTTP listener first; startup that reaches