diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index 617398eec88..ecda817f372 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -161,6 +161,25 @@ protected launch argument fails with the closed classification controller never opens a replacement session after either class, whatever the checkpoint's recovery policy, and stops the run for an operator. +A restart can kill the runner after it journaled a run's own cold `run.attach` +and before the attachment finished. The restored runner settles that command as +`execution_indeterminate` and never executes it again. When the journal shows no +provider work after it, the +controller's same-run resume queues a new `run.attach` with the same payload. +"No provider work" means that every later command is recovery maintenance as +`isRunnerRecoveryMaintenanceCommand` defines it (`session.snapshot`, +`session.goal.get`, `runner.drain`, `runner.suspend`, `runner.shutdown` and +`turn.stop`, none of which starts a turn), and that no turn, item, tool-input or +runtime-request event is committed. +Attaching starts no turn, and the runner admits the new command only for a +settled session, so the run continues instead of failing on the old command. +Warm transitions and staged attachments keep their exact replay paths. + +A kill during that attachment's provider launch can also leave a `spawned` +startup fence over a `prepared` state that holds a saved thread. The next runner +retires that fence once the recorded provider process and its group have exited, +as it does for an open or active session. + Saved OpenCode and Cursor Cloud provider state binds the digest of its own qualified launch profile (command, proxy script and, for OpenCode, the executable, each by path and SHA-256). Cursor Cloud does not re-bind. A changed diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index b94eac6ab45..b51cfce38a6 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -2418,10 +2418,15 @@ impl CodexCommandExecutor { let Some(attempt) = state.startup_attempt.clone() else { return Ok(()); }; + // A prepared state that already holds a provider thread is the + // settled checkpoint that run.attach restores (see attach_run): the + // launch that a kill cut off was restoring that saved thread. let restorable = matches!( state.lifecycle.as_str(), "session_open" | "turn_active" | "provider_exited" - ); + ) || (state.lifecycle == "prepared" + && state.thread_id.is_some() + && state.provider_process_generation > 0); let orphaned = match attempt.phase { // The runner died while checking the launch, before any spawn. ProviderStartupPhase::Intent => attempt.before_spawn, @@ -6305,6 +6310,7 @@ mod tests { state.config.driver = "codex_app_server".to_owned(); state.config.command = PathBuf::from("codex"); state.lifecycle = lifecycle.to_owned(); + state.provider_process_generation = 1; if lifecycle != "turn_active" { state.active_provider_turn_id = None; } @@ -6352,7 +6358,7 @@ mod tests { #[test] fn restart_retires_a_spawned_startup_whose_provider_process_exited() { - for lifecycle in ["turn_active", "session_open", "provider_exited"] { + for lifecycle in ["turn_active", "session_open", "provider_exited", "prepared"] { let pid = exited_process_group(); let directory = orphaned_startup_fixture("orphaned-startup-exited", lifecycle, spawned(pid)); @@ -6373,6 +6379,21 @@ mod tests { } } + #[test] + fn restart_keeps_the_fence_for_a_prepared_state_without_a_saved_thread() { + // Only a prepared state over a saved thread is a settled checkpoint. + let pid = exited_process_group(); + let directory = + orphaned_startup_fixture("orphaned-startup-unopened", "prepared", |executor| { + executor.state.as_mut().unwrap().thread_id = None; + spawned(pid)(executor); + }); + assert!(reload_and_retire(&directory) + .assert_startup_admitted() + .is_err()); + fs::remove_dir_all(directory).unwrap(); + } + #[test] fn restart_keeps_the_fence_for_a_startup_that_failed() { for (stage, child) in [ diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-cut-attach.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-cut-attach.test.ts new file mode 100644 index 00000000000..5a6722b9b77 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-cut-attach.test.ts @@ -0,0 +1,135 @@ +import { mkdtemp, readFile, rm, writeFile } 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"; + +/** + * A follow-up run (for example the run that carries a person's answer) starts + * a new runner and attaches it to the saved conversation. A restart can kill + * the runner and its provider after the runner journaled `run.attach` and + * before it finished. The restored runner then settles that command as + * `execution_indeterminate` and will never execute it again. No turn ran for + * this run, so the same-run resume attaches again with a new command instead + * of failing on the old one. + * + * With `startupFence`, the kill also landed while the attachment was launching + * the provider: the provider state holds a spawned startup attempt over a + * prepared (saved, settled) thread, and that provider process is gone. + */ +it.each([false, true])("re-attaches a follow-up whose run.attach a restart cut off (startup fence: %s)", async (startupFence) => { + const root = await mkdtemp(join(tmpdir(), "runnerd-cut-attach-")); + const callsPath = join(root, "calls.log"); + const identity = { + runnerInstanceId: "cut-attach-runner", + environmentLeaseId: "cut-attach-lease", + runId: "asking-run", + normalizedSessionId: "cut-attach-session", + turnId: "asking-turn", + itemId: "asking-item", + }; + const answerIdentity = { ...identity, runId: "answer-run", turnId: "answer-turn", itemId: "answer-item" }; + const options = { + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(root, "--call-log", callsPath), + stateDirectory: root, + }; + const kill = (pid: number | null | undefined) => { + if (!pid) return; + try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } + }; + const alive = (pid: number | null | undefined) => { + if (!pid) return false; + try { process.kill(pid, 0); return true; } catch { return false; } + }; + const asking = createCapabilityRunnerdCodexTransport({ ...options, prpIdentity: identity }); + let answer: ReturnType | undefined; + let resumed: ReturnType | undefined; + try { + const opened = await asking.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] }); + const openedThread = opened.thread as { id: string; sessionId: string }; + const thread = { id: openedThread.id, sessionId: openedThread.sessionId }; + await asking.transport.close(); + + const resumeProviderSession = { driverSessionId: thread.id, providerSessionId: thread.sessionId }; + answer = createCapabilityRunnerdCodexTransport({ ...options, prpIdentity: answerIdentity, resumeProviderSession }); + expect((await answer.transport.request("thread/read", {})).thread).toMatchObject(thread); + const owner = answer.evidence(); + await answer.detachControllerForRestart(); + kill(owner.runnerPid); + kill(owner.providerPid); + await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); }); + + // Rewrite both journals to the moment of the kill: run.attach journaled, + // its outcome not yet recorded, nothing after it. + const controlPath = join(root, "control-plane", "control-plane-state.json"); + const control = JSON.parse(await readFile(controlPath, "utf8")); + expect(control.identity.runId).toBe("answer-run"); + const attach = control.commands.findLast((command: { type: string }) => command.type === "run.attach"); + control.commands = control.commands.filter((command: { controllerSeq: number }) => command.controllerSeq <= attach.controllerSeq); + Object.assign(attach, { status: "pending", result: null }); + await writeFile(controlPath, JSON.stringify(control)); + const runnerPath = join(root, "runner", "runner-state.json"); + const runner = JSON.parse(await readFile(runnerPath, "utf8")); + const processed = runner.processedCommands as Record; + for (const [key, command] of Object.entries(processed)) { + if (command.controllerSeq > attach.controllerSeq) delete processed[key]; + else if (command.commandId === attach.commandId) processed[key] = { ...command, status: "pending", result: null } as never; + } + for (const key of Object.keys(runner.processedCommandFingerprints ?? {})) { + if (!(key in processed)) delete runner.processedCommandFingerprints[key]; + } + runner.lastControllerCommandSeq = attach.controllerSeq; + runner.lifecycle = "ready"; + await writeFile(runnerPath, JSON.stringify(runner)); + if (startupFence) { + const providerPath = join(root, "runner", "codex-provider-state.json"); + const provider = JSON.parse(await readFile(providerPath, "utf8")); + provider.lifecycle = "prepared"; + provider.startupAttempt = { + schema: "paperclip.provider_startup.v1", + launchId: "00000000-0000-4000-8000-000000000001", + phase: "spawned", + trigger: "ensure", + attemptedProcessGeneration: provider.providerProcessGeneration + 1, + origin: { runnerInstanceId: answerIdentity.runnerInstanceId, runId: answerIdentity.runId, + normalizedSessionId: answerIdentity.normalizedSessionId, turnId: answerIdentity.turnId, itemId: answerIdentity.itemId }, + command: { commandId: attach.commandId, controllerSeq: attach.controllerSeq, commandType: "run.attach" }, + configurationFingerprint: `sha256:${"0".repeat(64)}`, + requestedThreadId: thread.id, + authenticatedThreadId: null, + processId: owner.providerPid, + processGroupId: owner.providerPid, + failedStage: null, + directChildExitObserved: false, + exitCode: null, + signal: null, + processTreeRetired: false, + }; + await writeFile(providerPath, JSON.stringify(provider)); + } + const callsBefore = (await readFile(callsPath, "utf8")).trim().split(/\r?\n/); + + resumed = createCapabilityRunnerdCodexTransport({ ...options, prpIdentity: answerIdentity, resumeProviderSession }); + expect((await resumed.transport.request("thread/read", {})).thread).toMatchObject(thread); + const recovered = JSON.parse(await readFile(controlPath, "utf8")); + const attachments = recovered.commands.filter((command: { type: string }) => command.type === "run.attach"); + expect(attachments).toHaveLength(2); + expect(attachments[0]).toMatchObject({ commandId: attach.commandId, status: "failed", + result: { result: { code: "execution_indeterminate" } } }); + expect(attachments[1]).toMatchObject({ status: "completed", payload: attach.payload }); + const calls = (await readFile(callsPath, "utf8")).trim().split(/\r?\n/).slice(callsBefore.length); + expect(calls.filter(call => call === "thread/resume")).toHaveLength(1); + expect(calls.filter(call => call === "turn/start")).toHaveLength(0); + } finally { + await Promise.allSettled([asking.transport.close(), answer?.transport.close(), resumed?.transport.close()]); + for (const bundle of [asking, answer, resumed]) { + const evidence = bundle?.evidence(); + kill(evidence?.runnerPid); + kill(evidence?.providerPid); + } + await rm(root, { recursive: true, force: true }); + } +}, 60_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 8a8254cd41e..936eabbf687 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1019,6 +1019,32 @@ it("replays the durable run attachment outcome and latest provider identity", () }); }); +it("re-issues only this run's own cut-off attachment before any provider work", () => { + const unknown = { status: "failed", result: { result: { code: "execution_indeterminate" } } }; + const attach = { type: "run.attach", controllerSeq: 1, payload: { provider: { kind: "codex" } }, ...unknown }; + const state = (overrides: Record = {}, later: Record[] = [], events: string[] = []) => ({ + commands: [{ ...attach, ...overrides }, ...later.map((command, index) => ({ controllerSeq: index + 2, payload: {}, status: "completed", ...command }))], + committedEvents: events.map(eventType => ({ eventType })), + }) as never; + const reissue = runnerdRecoveryInternals.cutOffRunAttachmentPayload; + expect(reissue(state())).toEqual(attach.payload); + expect(reissue(state({ status: "indeterminate", result: null }))).toEqual(attach.payload); + expect(reissue(state({}, [{ type: "runner.drain", status: "failed" }, { type: "runner.suspend" }, { type: "session.goal.get" }, + { type: "turn.stop" }, { type: "session.snapshot" }], ["runner.reconciled", "session.resumed"]))).toEqual(attach.payload); + // A definite answer, a pending command, or a warm or staged attachment keeps its own path. + expect(reissue(state({ status: "completed", result: { status: "completed" } }))).toBeNull(); + expect(reissue(state({ status: "failed", result: { result: { code: "command_execution_failed" } } }))).toBeNull(); + expect(reissue(state({ status: "pending", result: null }))).toBeNull(); + expect(reissue(state({ payload: { paperclipNextAuthority: {} } }))).toBeNull(); + expect(reissue({ ...(state() as object), warmTransition: {} } as never)).toBeNull(); + expect(reissue({ ...(state() as object), stagedRunAttachment: {} } as never)).toBeNull(); + // Any provider work after it means the run moved on. + for (const type of ["turn.start", "turn.steer", "request.resolve", "semantic_tool.result", "session.goal.set", "run.attach"]) { + expect(reissue(state({}, [{ type }]))).toBeNull(); + } + expect(reissue(state({}, [], ["turn.started"]))).toBeNull(); +}); + it("identifies an active provider turn that must stop before suspension", () => { expect( runnerdRecoveryInternals.providerDrainStateFromSnapshot({ diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 6bc1901d05b..7534e459948 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -678,6 +678,41 @@ function recoveredRunAttachment(state: { }; } +// Committed events that show a provider turn or tool work started. Commands +// use the control plane's shared recovery-maintenance vocabulary instead. +const PROVIDER_WORK_EVENT_TYPES = new Set(["turn.accepted", "turn.started", "item.started", "tool.execution.started", + "semantic_tool.input", "runtime_request.created", "mcp_app.tool_input"]); + +/** + * The payload of this run's own `run.attach` when a restart cut it off before + * any provider work: the runner journaled it and then died, so the restored + * runner settled it as `execution_indeterminate` and will not execute it + * again. Attaching binds the run to its saved session and starts no turn, so a + * new attachment with the same payload is safe while nothing after the old one + * started provider work: every later command is recovery maintenance (see + * isRunnerRecoveryMaintenanceCommand), and no committed event shows a turn or + * tool work. Each runner refuses the new one unless its provider + * session is settled. Warm transitions and staged attachments keep their own + * exact replay paths and are never re-issued here. + */ +function cutOffRunAttachmentPayload(state: { + commands: readonly { type: string; status: string; controllerSeq: number; payload: Record; result?: unknown }[]; + committedEvents: readonly { eventType: string }[]; + warmTransition?: unknown; + stagedRunAttachment?: unknown; +}): Record | null { + if (state.warmTransition != null || state.stagedRunAttachment != null) return null; + const attach = state.commands.findLast(command => command.type === "run.attach"); + if (!attach || attach.payload.paperclipNextAuthority !== undefined) return null; + const outcomeUnknown = attach.status === "indeterminate" || (attach.status === "failed" + && record(record(attach.result).result).code === "execution_indeterminate"); + if (!outcomeUnknown) return null; + if (state.commands.some(command => command.controllerSeq > attach.controllerSeq + && !isRunnerRecoveryMaintenanceCommand(command.type))) return null; + if (state.committedEvents.some(event => PROVIDER_WORK_EVENT_TYPES.has(event.eventType))) return null; + return structuredClone(attach.payload); +} + function latestProviderIdentityEventIndex( events: readonly { eventType: string }[], ): number { @@ -7059,7 +7094,19 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // 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, admissionDeadline, true); + try { + await this.#waitCommand("run.attach", runAttachment.commandId, admissionDeadline, true); + } catch (error) { + // A restart can cut off this run's own attachment after the runner + // journaled it. The restored runner settles it as + // execution_indeterminate and never executes it again. Attach once + // more with a new command; the runner admits it only for a settled + // provider session. + const reattachment = cutOffRunAttachmentPayload(core.store.state); + if (!reattachment) throw error; + const retry = core.queueCommand("run.attach", reattachment, undefined, true); + await this.#waitCommand("run.attach", retry.commandId, admissionDeadline, true); + } } // A launched runner normally advanced at authentication, before run.attach // restores the provider. An adopted runner had no launch step. @@ -9021,6 +9068,7 @@ export const runnerdRecoveryInternals = Object.freeze({ providerDrainStateFromSnapshot, providerTurnIsActiveFromCommittedEvents, recoveredRunAttachment, + cutOffRunAttachmentPayload, releaseRunnerProcessOwnership, runnerCloseDeadlines, runnerCloseGraceMs,