diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index 040fb5748e2..9b43063fd5c 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -243,6 +243,15 @@ runner refused), and its acknowledgement names the runner's journal status runners accept only `indeterminate` and otherwise exit with "terminal command result acknowledgement changed its durable identity". +That runner exits before it accepts any command of the new attempt. During +startup, the controller restarts it in place once, so the attempt continues. +It does this only when the runner's durable journal proves the exit: its last +command is exactly the suspend that an earlier attempt journaled, the +controller has settled that suspend, the runner rests `suspended` with no +terminal or provider cleanup fence, and no command of this attempt has a +result. The runner file is read only after such an exit. Any other exit keeps +its earlier failure. + 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, diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-carried-suspend.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-carried-suspend.test.ts new file mode 100644 index 00000000000..87cf56ba2e9 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-carried-suspend.test.ts @@ -0,0 +1,160 @@ +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, runnerdRecoveryInternals } from "./runnerd-codex-transport.js"; +import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js"; + +/** + * A server freeze while the controller suspends a finished turn can stop the + * runner inside `runner.suspend`: the runner journaled the command, its + * outcome is not recorded, and the controller still lists it as pending. The + * next same-run attempt starts a new runner. That runner settles the old + * suspend from the welcome (as `execution_indeterminate`; it never runs it + * again), records `suspended` and exits with code 0 before it accepts any of + * this attempt's commands. The attempt restarts the runner in place once, and + * the resume completes in the same attempt without another provider turn. + * + * `fenced` is the state a released build can leave: the controller already + * recorded the suspend as failed and the runner kept its terminal fence. The + * next runner reconciles that fence, keeps a provider cleanup fence and exits. + * That runner would refuse new work, so the attempt is not restarted in place: + * any exit without the exact proof keeps the earlier failure. + */ +it.each(["journaled", "fenced"] as const)("restarts in place only a runner that settled the previous attempt's suspend (%s)", async (mode) => { + const root = await mkdtemp(join(tmpdir(), "runnerd-carried-suspend-")); + const callsPath = join(root, "calls.log"); + const identity = { + runnerInstanceId: "carried-suspend-runner", + environmentLeaseId: "carried-suspend-lease", + runId: "carried-suspend-run", + normalizedSessionId: "carried-suspend-session", + turnId: "carried-suspend-turn", + itemId: "carried-suspend-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 first = createCapabilityRunnerdCodexTransport({ ...options, prpIdentity: identity }); + let resumed: ReturnType | undefined; + try { + const opened = await first.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 first.transport.close(); + + // Rewrite both journals to the moment of the freeze: the runner journaled + // the suspend before its effect; the controller never saw its result. + const controlPath = join(root, "control-plane", "control-plane-state.json"); + const control = JSON.parse(await readFile(controlPath, "utf8")); + const suspend = control.commands.findLast((command: { type: string }) => command.type === "runner.suspend"); + expect(suspend).toMatchObject({ status: "completed" }); + expect(control.commands.at(-1).commandId).toBe(suspend.commandId); + const indeterminate = { code: "execution_indeterminate", message: "runner recovered after journaling this command; it will not execute twice" }; + Object.assign(suspend, mode === "journaled" ? { status: "pending", result: null } : { + status: "failed", + result: { commandId: suspend.commandId, controllerSeq: suspend.controllerSeq, commandType: "runner.suspend", + status: "failed", result: indeterminate }, + }); + await writeFile(controlPath, JSON.stringify(control)); + const runnerPath = join(root, "runner", "runner-state.json"); + const runner = JSON.parse(await readFile(runnerPath, "utf8")); + expect(runner.lastControllerCommandSeq).toBe(suspend.controllerSeq); + runner.processedCommands[suspend.commandId] = { + ...runner.processedCommands[suspend.commandId], + ...(mode === "journaled" + ? { status: "pending", result: { code: "execution_indeterminate", message: "command was journaled before its effect" } } + : { status: "indeterminate", result: indeterminate }), + }; + runner.lifecycle = mode === "journaled" ? "ready" : "suspended"; + runner.pendingTerminalDelivery = mode === "journaled" ? null : { + commandId: suspend.commandId, controllerSeq: suspend.controllerSeq, commandType: "runner.suspend", lifecycle: "suspended", + }; + await writeFile(runnerPath, JSON.stringify(runner)); + const callsBefore = (await readFile(callsPath, "utf8")).trim().split(/\r?\n/); + + resumed = createCapabilityRunnerdCodexTransport({ + ...options, + prpIdentity: identity, + resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId }, + }); + if (mode === "fenced") { + await expect(resumed.transport.request("thread/read", {})).rejects.toThrow( + "runnerd exited unexpectedly with code 0", + ); + expect(resumed.evidence().diagnostics).not.toContain( + "runner settled a suspend left by an earlier attempt and exited; restarting it in place", + ); + return; + } + expect((await resumed.transport.request("thread/read", {})).thread).toMatchObject(thread); + expect(resumed.evidence().diagnostics).toContain( + "runner settled a suspend left by an earlier attempt and exited; restarting it in place", + ); + const recovered = JSON.parse(await readFile(controlPath, "utf8")); + expect(recovered.commands.find((command: { commandId: string }) => command.commandId === suspend.commandId)) + .toMatchObject({ status: "failed", result: { result: { code: "execution_indeterminate" } } }); + const calls = (await readFile(callsPath, "utf8")).trim().split(/\r?\n/).slice(callsBefore.length); + expect(calls.filter(call => call === "turn/start")).toHaveLength(0); + } finally { + await Promise.allSettled([first.transport.close(), resumed?.transport.close()]); + for (const bundle of [first, resumed]) { + const evidence = bundle?.evidence(); + kill(evidence?.runnerPid); + kill(evidence?.providerPid); + } + await vi.waitFor(() => undefined); + await rm(root, { recursive: true, force: true }); + } +}, 60_000); + +it("restarts only for the exact settled suspend left by an earlier attempt", () => { + const prove = runnerdRecoveryInternals.carriedSuspendSettlementProven; + const identity = { + runnerInstanceId: "runner", environmentLeaseId: "lease", runId: "run", + normalizedSessionId: "session", turnId: "turn", itemId: "item", + }; + const runnerState = (overrides: Record = {}) => ({ + schema: "paperclip.runner.durable.state.v1", ...identity, lifecycle: "suspended", + pendingTerminalDelivery: null, pendingProviderCleanup: null, lastControllerCommandSeq: 3, + processedCommands: { suspend: { commandId: "suspend", controllerSeq: 3, commandType: "runner.suspend", status: "indeterminate" } }, + ...overrides, + }); + const commands = (overrides: Record = {}, later: Record[] = []) => [ + { commandId: "prepare", controllerSeq: 1, type: "run.prepare", status: "completed" }, + { commandId: "turn", controllerSeq: 2, type: "turn.start", status: "completed" }, + { commandId: "suspend", controllerSeq: 3, type: "runner.suspend", status: "failed", ...overrides }, + { commandId: "probe", controllerSeq: 4, type: "runner.drain", status: "pending" }, + ...later, + ] as { commandId: string; controllerSeq: number; type: string; status: string }[]; + const proven = (state = runnerState(), journal = commands(), carriedCommandCount = 3) => + prove({ runnerState: state, identity, commands: journal, carriedCommandCount }); + + expect(proven()).toBe(true); + // This attempt queued the suspend, as a normal close does. + expect(proven(runnerState(), commands(), 2)).toBe(false); + // The controller has not settled it yet. + expect(proven(runnerState(), commands({ status: "pending" }))).toBe(false); + // The last command is not a suspend. + expect(proven(runnerState(), commands({ type: "runner.shutdown" }))).toBe(false); + // The runner went on to a command of this attempt. + expect(proven(runnerState({ lastControllerCommandSeq: 4 }))).toBe(false); + expect(proven(runnerState(), commands({}, []).map(command => + command.commandId === "probe" ? { ...command, status: "completed" } : command))).toBe(false); + // The runner did not rest cleanly suspended. + expect(proven(runnerState({ lifecycle: "ready" }))).toBe(false); + expect(proven(runnerState({ pendingTerminalDelivery: { commandId: "suspend" } }))).toBe(false); + expect(proven(runnerState({ pendingProviderCleanup: { commandId: "suspend" } }))).toBe(false); + // The runner's receipt names another command or another authority. + expect(proven(runnerState({ processedCommands: { suspend: { commandId: "suspend", controllerSeq: 3, + commandType: "runner.shutdown", status: "completed" } } }))).toBe(false); + expect(proven(runnerState({ processedCommands: {} }))).toBe(false); + expect(proven(runnerState({ runId: "other-run" }))).toBe(false); +}); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 15cb01e41f8..b5ce4030a0c 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -119,6 +119,8 @@ const RUNNER_CLIENT_VERSION = "0.3.0"; const RUNNER_BOOTSTRAP_TICKET_TTL_MS = 60_000; const RUNNER_RESTORE_STABLE_MS = 60_000; const MAX_UNSTABLE_RUNNER_RESTORES = 3; +// Bounds the one in-place restart after a runner settles an earlier attempt's suspend. +const CARRIED_SUSPEND_RESTART_GRACE_MS = 30_000; const RUNNERD_MAX_OUTBOX_BYTES = 16 * 1024 * 1024; const RUNNERD_P0_RESERVE_BYTES = 1024 * 1024; @@ -4052,6 +4054,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { readonly #failureSignal: Promise; #rejectFailureSignal!: (error: Error) => void; #runnerRecoveryInProgress = false; + /** Controller commands journaled before this attempt queued any. */ + #carriedCommandCount: number | null = null; + /** Each runner exit's carried-suspend classification, read once after exit. */ + readonly #carriedSuspendExits = new WeakMap>(); + #carriedSuspendRestartUsed = false; #lastRunnerRestoreAt: number | null = null; #unstableRunnerRestores = 0; #startupComplete = false; @@ -6081,6 +6088,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { connectionLeaseTtlMs: 60 * 60 * 1_000, }); this.#core = core; + this.#carriedCommandCount = core.store.state.commands.length; // Externally launched runners own their state directory (for example in a // Daytona sandbox). Do not create an empty controller-side placeholder: // prior-run authority checks must be able to distinguish absent remote @@ -6655,6 +6663,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { connectionLeaseTtlMs: 60 * 60 * 1_000, }); this.#core = core; + this.#carriedCommandCount = core.store.state.commands.length; // Reusing resources still requires current environment authorization. The // optional hook avoids re-preparation after controller storage is lost. // Existing consumers retain the preparation callback's authorization path. @@ -8402,8 +8411,16 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #runnerHasExited(): Promise { // A runnerd killed by a signal reports exitCode null and its signalCode. - if (this.#handle) return this.#handle.isExitConfirmed?.() - ?? (this.#handle.child.exitCode !== null || (this.#handle.child.signalCode ?? null) !== null); + if (this.#handle) { + const handle = this.#handle; + const exited = handle.isExitConfirmed?.() + ?? (handle.child.exitCode !== null || (handle.child.signalCode ?? null) !== null); + // A runner that only settled a suspend left by an earlier attempt is + // being restarted in place; startup keeps waiting for its replacement. + if (exited && !this.#startupComplete && await this.#carriedSuspendExits.get(handle)) + return this.#closed || this.#failure !== null; + return exited; + } const adoptedRunner = this.options.adoptExistingRunner; if (!adoptedRunner) return true; return !(await adoptedRunner.isAlive()); @@ -8481,8 +8498,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } #watchRunner(handle: RunnerProcessHandle): void { + const carriedSuspendExit = handle.completion.then( + (result) => this.#settledOnlyCarriedSuspend(handle, result.code), + () => false, + ); + this.#carriedSuspendExits.set(handle, carriedSuspendExit); void handle.completion.then( - (result) => { + async (result) => { const exitConfirmed = handle.isExitConfirmed?.() ?? true; this.#evidence.runnerExited = exitConfirmed; this.#evidence.runnerExitCode = result.code; @@ -8498,6 +8520,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#lostRunnerHandle = handle; if (this.#closed || this.#failure !== null || this.#handle !== handle || (this.#releaseEpisode && !this.#releaseEpisode.claimed)) return; + if (await carriedSuspendExit) { + if (this.#closed || this.#failure !== null || this.#handle !== handle) return; + this.#carriedSuspendRestartUsed = true; + this.#diagnostic("runner settled a suspend left by an earlier attempt and exited; restarting it in place"); + void this.#recoverRunnerProcess(handle, detail, CARRIED_SUSPEND_RESTART_GRACE_MS); + return; + } + if (this.#closed || this.#failure !== null || this.#handle !== handle) return; // A per-turn runner exits after its terminal suffix is durably ACKed. // Drain that suffix into the provider-facing queue before classifying // process completion; clean terminal exit is the expected lifecycle. @@ -8567,6 +8597,37 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ); } + /** + * A freeze or restart during `runner.suspend` leaves that command pending in + * the controller journal. The next attempt's runner settles it from the + * welcome and exits as suspended before it accepts this attempt's work. Only + * that exit is restarted in place, once, and only when the durable runner + * journal proves it: the runner's last command is exactly that earlier + * suspend, settled by the controller, and the runner rests suspended with no + * fence. It then executed nothing this attempt queued, so no turn started and + * no provider work can repeat. The runner file is read only after an exit. + */ + async #settledOnlyCarriedSuspend(handle: RunnerProcessHandle, code: number | null): Promise { + const core = this.#core; + const carried = this.#carriedCommandCount; + if (code !== 0 || handle.isExitConfirmed?.() === false || !handle.restart || core === null || carried === null + || this.#carriedSuspendRestartUsed || this.#startupComplete || this.#closed || this.#failure !== null + || this.options.adoptExistingRunner || this.options.runnerProcessLauncher || this.options.readRunnerState + || this.options.runnerStateDirectory !== undefined) return false; + let state: Record; + try { + state = await this.#readDurableRunnerState(); + } catch { + return false; + } + return carriedSuspendSettlementProven({ + runnerState: state, + identity: core.store.state.identity, + commands: core.store.state.commands, + carriedCommandCount: carried, + }); + } + #recoverOrFailRunnerProcess(handle: RunnerProcessHandle, detail: string): void { this.#unstableRunnerRestores = unstableRunnerRestoreCount( this.#unstableRunnerRestores, @@ -8587,6 +8648,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #recoverRunnerProcess( failedHandle: RunnerProcessHandle, initialDetail: string, + graceOverrideMs?: number, ): Promise { if ( this.#runnerRecoveryInProgress || @@ -8598,7 +8660,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ) return; this.#runnerRecoveryInProgress = true; - const graceMs = this.options.runnerReconnectGraceMs ?? 0; + const graceMs = graceOverrideMs ?? this.options.runnerReconnectGraceMs ?? 0; const deadline = Date.now() + graceMs; const delays = [250, 500, 1_000, 2_000, 5_000] as const; let attempt = 0; @@ -9118,7 +9180,33 @@ function unstableRunnerRestoreCount(previous: number, restoredAt: number | null, return restoredAt !== null && now - restoredAt < RUNNER_RESTORE_STABLE_MS ? previous + 1 : 0; } +/** The durable proof that a runner's clean exit only settled a suspend that an + * earlier attempt journaled: its last command is exactly that suspend, which + * the controller has settled; it rests suspended with no fence; and nothing + * this attempt queued has a result, so no turn or provider work ran. */ +function carriedSuspendSettlementProven(input: { + runnerState: Record; + identity: DurableRecoveryIdentity; + commands: readonly { commandId: string; controllerSeq: number; type: string; status: string }[]; + carriedCommandCount: number; +}): boolean { + const { runnerState: state, commands, carriedCommandCount: carried } = input; + const seq = state.lastControllerCommandSeq; + if (state.schema !== "paperclip.runner.durable.state.v1" + || !recoveryIdentityMatches(state, input.identity) + || state.lifecycle !== "suspended" || state.pendingTerminalDelivery != null + || state.pendingProviderCleanup != null || state.warmTransition != null + || typeof seq !== "number" || !Number.isSafeInteger(seq) || seq < 1 || seq > carried) return false; + const suspend = commands.find((command) => command.controllerSeq === seq); + if (suspend?.type !== "runner.suspend" || suspend.status === "pending") return false; + if (commands.some((command) => command.controllerSeq > carried && command.status !== "pending")) return false; + const processed = record(record(state.processedCommands)[suspend.commandId]); + return processed.commandId === suspend.commandId && processed.commandType === "runner.suspend" + && processed.controllerSeq === seq && processed.status !== "pending"; +} + export const runnerdRecoveryInternals = Object.freeze({ + carriedSuspendSettlementProven, unstableRunnerRestoreCount, projectCopilotRunGrant, completedMaintenanceTerminalReceipt,