diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 37092285f2b..44ff7b33b11 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3173,6 +3173,13 @@ configuration, and controller-owned credential or instruction resources use verified suspension and saved-history resume. This recovery can restart a provider; it does not suspend every healthy warm turn. +A cold continuation can retain the completed receipt from a prior warm handoff. +Before inspecting that receipt against physical Runner state, the transport +materializes the admitted external backup. A refused restore or mismatched +session identity prevents provider startup. Adoption of a live Runner, pending +handoffs and active warm transitions still require their original physical state +and ownership checks without restoring a backup. + A different task can reuse the same physical sandbox after its prior task finishes. Provider ingress has one listener per allocation. Before a new logical session starts, Runner retires a locally retained idle session that still owns diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-cold-restore.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-cold-restore.test.ts new file mode 100644 index 00000000000..d7598b004fa --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-cold-restore.test.ts @@ -0,0 +1,107 @@ +import { cp, mkdir, mkdtemp, readFile, rename, 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"; + +it.each(["restore", "refuse", "foreign-state"] as const)( + "continues a completed warm handoff through external cold-state preparation: %s", + async (mode) => { + const root = await mkdtemp(join(tmpdir(), "runnerd-warm-cold-restore-")); + const saved = await mkdtemp(join(tmpdir(), "runnerd-warm-saved-")); + const callsPath = join(root, "calls.log"); + const identity = { + runnerInstanceId: "cold-restore-runner", + environmentLeaseId: "cold-restore-lease", + runId: "first-run", + normalizedSessionId: "cold-restore-session", + turnId: "first-turn", + itemId: "first-item", + }; + const options = { + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(root, "--call-log", callsPath, "--record-process-start"), + stateDirectory: root, + lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 }, + runnerReconnectGraceMs: 2_000, + }; + 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.attachRun!({ runId: "second-run", turnId: "second-turn", itemId: "second-item" }); + expect((await first.transport.request("thread/read", {})).thread).toMatchObject(thread); + await first.transport.close(); + const controlPath = join(root, "control-plane", "control-plane-state.json"); + const controlBytes = await readFile(controlPath); + const control = JSON.parse(controlBytes.toString()); + expect(control.completedWarmTransition.receipt.newIdentity.runId).toBe("second-run"); + const savedRunner = join(saved, "runner"); + await rename(join(root, "runner"), savedRunner); + const savedState = JSON.parse(await readFile(join(savedRunner, "runner-state.json"), "utf8")); + expect(savedState).toMatchObject({ lifecycle: "suspended", runId: "second-run" }); + const callsBefore = await readFile(callsPath, "utf8"); + const sentinel = "workspace content must survive cold provider recovery"; + await writeFile(join(root, "workspace-sentinel.txt"), sentinel); + let prepared = false; + const prepare = vi.fn(async () => { + if (prepared) return; + if (mode === "refuse") throw new Error("fixture cold restore authority refused"); + await cp(savedRunner, join(root, "runner"), { recursive: true, errorOnExist: true }); + if (mode === "foreign-state") { + await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({ + ...savedState, normalizedSessionId: "another-session", + })); + } + prepared = true; + }); + const read = vi.fn(async () => JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8"))); + const archive = vi.fn(async (request: { archiveKey: string }) => { + const state = await read(); + const directory = join(root, "authority-epochs", request.archiveKey); + await mkdir(directory, { recursive: true }); + await rename(join(root, "runner", "runner-state.json"), join(directory, "runner-state.json")); + return state; + }); + resumed = createCapabilityRunnerdCodexTransport({ + ...options, + prpIdentity: { ...identity, runId: "third-run", turnId: "third-turn", itemId: "third-item" }, + resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId }, + readRunnerState: read, + prepareExternalRunnerState: prepare, + archiveExternalRunnerState: archive, + }); + if (mode === "refuse") { + await expect(resumed.transport.request("thread/read", {})).rejects.toThrow("fixture cold restore authority refused"); + expect(read).not.toHaveBeenCalled(); + expect(archive).not.toHaveBeenCalled(); + expect(await readFile(callsPath, "utf8")).toBe(callsBefore); + expect(await readFile(controlPath)).toEqual(controlBytes); + } else if (mode === "foreign-state") { + await expect(resumed.transport.request("thread/read", {})).rejects.toThrow("native_runner_authority_rotation_requires_settled_state"); + expect(read).toHaveBeenCalled(); + expect(archive).not.toHaveBeenCalled(); + expect(await readFile(callsPath, "utf8")).toBe(callsBefore); + expect(await readFile(controlPath)).toEqual(controlBytes); + } else { + expect((await resumed.transport.request("thread/read", {})).thread).toMatchObject(thread); + expect(prepare.mock.invocationCallOrder[0]).toBeLessThan(read.mock.invocationCallOrder[0]); + const calls = (await readFile(callsPath, "utf8")).trim().split(/\r?\n/); + expect(calls.filter(call => call === "thread/start")).toHaveLength(1); + expect(calls.filter(call => call === "thread/resume")).toHaveLength(1); + expect(calls.filter(call => call === "turn/start")).toHaveLength(0); + expect(JSON.parse(await readFile(controlPath, "utf8")).identity.runId).toBe("third-run"); + expect(await readFile(join(root, "workspace-sentinel.txt"), "utf8")).toBe(sentinel); + } + } finally { + await Promise.allSettled([first.transport.close(), resumed?.transport.close()]); + await rm(root, { recursive: true, force: true }); + await rm(saved, { recursive: true, force: true }); + } + }, + 30_000, +); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 82e23dfa1ae..f699208da3d 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -5526,23 +5526,30 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { let identity = controlPlaneState ? controlPlaneIdentity(controlPlaneState) : desiredIdentity; + const hasControlPlaneWarmBoundary = + controlPlaneState?.warmTransition !== undefined || + controlPlaneState?.schema === + "paperclip.runner.durable.control-plane-state.warm-transition.v1"; + const hasPendingWarmAttachment = + Array.isArray(controlPlaneState?.commands) && + controlPlaneState.commands.some((entry) => { + const command = record(entry); + return ( + command.type === "run.attach" && + command.status === "pending" && + record(command.payload).paperclipNextAuthority !== undefined + ); + }); if ( this.options.readRunnerState && controlPlaneState && - (controlPlaneState.warmTransition !== undefined || + (hasControlPlaneWarmBoundary || controlPlaneState.completedWarmTransition !== undefined || - controlPlaneState.schema === - "paperclip.runner.durable.control-plane-state.warm-transition.v1" || - (Array.isArray(controlPlaneState.commands) && - controlPlaneState.commands.some((entry) => { - const command = record(entry); - return ( - command.type === "run.attach" && - command.status === "pending" && - record(command.payload).paperclipNextAuthority !== undefined - ); - }))) + hasPendingWarmAttachment) ) { + if (!this.options.adoptExistingRunner && !hasControlPlaneWarmBoundary && !hasPendingWarmAttachment) { + await this.options.prepareExternalRunnerState?.(); + } candidateRunnerState = await this.options.readRunnerState(); } const staged = inspectStagedRunAttachment(controlPlaneState, desiredIdentity); @@ -5558,9 +5565,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { port?: number; } | null = null; const hasWarmBoundary = - controlPlaneState?.warmTransition !== undefined || - controlPlaneState?.schema === - "paperclip.runner.durable.control-plane-state.warm-transition.v1" || + hasControlPlaneWarmBoundary || hasRunnerWarmBoundary || candidateRunnerState?.warmTransition !== undefined || candidateRunnerState?.schema ===