diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index 10e6f86282f..10b19629cb7 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -606,6 +606,14 @@ requires separate artifact and ownership admission; this path does not qualify that migration. Pending warm-transition receipts require their separate exact recovery admission, not an ordinary bootstrap. +A hosted external route keeps the source connection until an independent +physical-state read proves the target authority through the existing warm +transition inspector. Recovery first activates the proven source registration +so the Runner can consume its old receipt acknowledgement. It then activates +the target registration. Ordinary attachment uses the same bounded barrier. +External receipt recovery requires exact staged adoption and a physical reader; +connect routes keep their existing activation order. + ## Backpressure and bounded storage The runner has a byte limit and a reserved P0 region. diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 67f21ab4fd3..32b35cebcce 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -4396,7 +4396,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { activationStarted = true; core.rotateRunIdentity(desired, runAttachTemplate); - if (core.store.state.stagedRunAttachment && registration?.connection?.mode === "listen" && this.options.readRunnerState) { + if (core.store.state.stagedRunAttachment && this.options.readRunnerState + && (registration?.connection?.mode === "listen" || registration?.connection?.mode === "external")) { const deadline = Math.min(Date.now() + (this.options.runnerReconnectGraceMs ?? 5_000), this.#sourceAttachmentDeadline ?? Infinity); const artifact = this.options.runnerArtifact ?? approvedRunnerArtifact(this.options.runnerBinary ?? defaultCapabilityRunnerdBinary()); for (;;) { @@ -5697,7 +5698,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ? new URL(connection.connectUrl) : null; if ( - (connection.mode !== "connect" && !(connection.mode === "listen" && staged + (connection.mode !== "connect" && !((connection.mode === "listen" || connection.mode === "external" && this.options.readRunnerState) && staged && this.options.adoptExistingRunner && this.options.controlPlaneRegistration && this.options.warmTransitionRegistrationMode === "routed_same_endpoint")) || (connection.mode === "connect" && (endpoint === null || endpoint.search || endpoint.hash || endpoint.username || endpoint.password)) || @@ -6332,15 +6333,16 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#assertStartupOpen(); let targetRecoveryActivated = registration === newTransitionRegistration; if (oldTransitionRegistration && newTransitionRegistration) { - // A guest listener still serves only its actual physical run. Open that - // route first; probing the future/retired path cannot prove readiness. + // Guest listeners and hosted external ingress serve only their physical + // run. Open that route first so its receipt ACK can activate the target. await registration?.activate?.(); this.#assertStartupOpen(); if (registration === oldTransitionRegistration) { // Source authentication can consume its receipt and disappear before // a sampled count sees it. The durable/physical proof loop below owns // that progress; only target authentication completes recovery. - if (expectedRecoveryConnection?.mode !== "listen") { + if (expectedRecoveryConnection?.mode !== "listen" + && !(expectedRecoveryConnection?.mode === "external" && this.options.readRunnerState)) { await newTransitionRegistration.activate?.(); targetRecoveryActivated = true; } } @@ -6377,8 +6379,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { const connections = core.activeRunnerConnectionCount(); if (connections === 0 && previousConnections > 0) nextPhysicalInspection = 0; previousConnections = connections; - if (!sourceRouteRetired && expectedRecoveryConnection?.mode === "listen" && oldTransitionRegistration - && newTransitionRegistration && this.options.readRunnerState && Date.now() >= nextPhysicalInspection) { + if (!sourceRouteRetired && (expectedRecoveryConnection?.mode === "listen" || expectedRecoveryConnection?.mode === "external") + && oldTransitionRegistration && newTransitionRegistration && this.options.readRunnerState && Date.now() >= nextPhysicalInspection) { nextPhysicalInspection = Date.now() + 250; const physical = await this.options.readRunnerState(); const proof = inspectWarmRunTransition({ controlPlaneState: core.store.state, runnerState: physical, diff --git a/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts b/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts index 7749ce4d927..4e16fd7a3b7 100644 --- a/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts @@ -6,7 +6,7 @@ import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { expect, it, vi } from "vitest"; import { - DurablePrpControlPlane, inspectStagedRunAttachmentHeader, + DurablePrpControlPlane, inspectStagedRunAttachmentHeader, spawnRunner, type DurablePrpCheckpoint, type RunnerProcessConnection, type PrpWireConnection, } from "../control-plane/durable-prp-control-plane.js"; import { @@ -99,7 +99,7 @@ it("delivers a WebSocket message coalesced with the real HTTP upgrade response", }); // Real encrypted PRP and real Runner/provider processes; no model or server imports. -it.each([["connect", "owned"], ["listen", "owned"], ["connect", "legacy"], ["listen", "legacy"], ["connect", "cleanup"]] as const)("recovers a staged %s %s handoff at every durable boundary without another process", async (mode, policy) => { +it.each([["connect", "owned"], ["listen", "owned"], ["external", "owned"], ["connect", "legacy"], ["listen", "legacy"], ["connect", "cleanup"]] as const)("recovers a staged %s %s handoff at every durable boundary without another process", async (mode, policy) => { const root = await mkdtemp(join(tmpdir(), "runner-staged-handoff-")); const shared = join(root, "shared"), calls = join(root, "calls.log"); const source = { runnerInstanceId: "runner-staged", environmentLeaseId: "lease-staged", runId: "source-run", @@ -137,7 +137,7 @@ it.each([["connect", "owned"], ["listen", "owned"], ["connect", "legacy"], ["lis const atBoundary = new Promise(done => { reached = done; }); const cores = new Set(); const registered = new WeakSet(); - let generation = 0; + let generation = 0, authorityRunId = target.runId; const registration = async (core: DurablePrpControlPlane, identity = core.store.state.identity) => { currentCore = core; if (!registered.has(core)) { @@ -157,6 +157,7 @@ it.each([["connect", "owned"], ["listen", "owned"], ["connect", "legacy"], ["lis const path = `/api/runner/v1/connect/${identity.runId}`, token = Symbol(path); const connection: RunnerProcessConnection = mode === "connect" ? { mode: "connect", connectUrl: `ws://127.0.0.1:${address.port}${path}` } + : mode === "external" ? { mode: "external" } : { mode: "listen", listenAddress: "0.0.0.0", listenPort: guestPort, listenPath: path }; if (mode === "connect") { routes.set(path, { core, token }); @@ -170,7 +171,7 @@ it.each([["connect", "owned"], ["listen", "owned"], ["connect", "legacy"], ["lis const deadline = Date.now() + 8_000; while (!stopped && Date.now() < deadline) { try { - const peer = await openOwnedWire(`ws://127.0.0.1:${guestPort}${path}`); socket = peer; + const peer = await openOwnedWire(`ws://127.0.0.1:${guestPort}${mode === "external" ? `/api/runner/v1/connect/${source.runId}` : path}`); socket = peer; if (stopped) { peer.close(); return; } core.attachWireConnection(peer.wire); readyResolve(); await peer.closed; } catch { /* The target listener appears only after the old receipt ACK. */ } @@ -179,13 +180,24 @@ it.each([["connect", "owned"], ["listen", "owned"], ["connect", "legacy"], ["lis } if (!stopped) readyReject(new Error("Owned listener did not become available")); }; - return { connection, activate: () => { void connect(); }, ready: () => ready, + return { connection, activate: async () => { + if (mode === "external" && (generation > 0 || identity.runId !== source.runId)) { + // Hosted ingress admits only the physical authority currently served. + const physical = JSON.parse(await readFile(join(directory, "runner/runner-state.json"), "utf8")); + for (const [key, value] of Object.entries(identity)) expect(physical[key]).toBe(value); + } + void connect(); + }, ready: () => ready, release: () => { stopped = true; socket?.close(); } }; }; const options = { runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"), codexArgs: ["--call-log", callLog, "--record-process-start", "--state-file", join(directory, "fake.json")], stateDirectory: directory, readRunnerState: async () => JSON.parse(await readFile(join(directory, "runner/runner-state.json"), "utf8")), lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 }, runnerReconnectGraceMs: 5_000, prpIdentity: source, controlPlaneRegistration: registration, + ...(mode === "external" ? { runnerProcessLauncher: (spec: import("../control-plane/durable-prp-control-plane.js").RunnerProcessLaunchSpec) => + spawnRunner({ ...spec.nativeRunner!, runnerBinaryPath: defaultCapabilityRunnerdBinary(), environment: spec.environment, + connection: { mode: "listen", listenAddress: "0.0.0.0", listenPort: guestPort, + listenPath: `/api/runner/v1/connect/${source.runId}` } }) } : {}), onStagedRunAttachment: async () => undefined, onRecoveryAccountingEvent: async () => undefined }; const first = createCapabilityRunnerdCodexTransport(options); let replacement: ReturnType | undefined; @@ -209,11 +221,11 @@ it.each([["connect", "owned"], ["listen", "owned"], ["connect", "legacy"], ["lis checkpoint.restoreSessionDirectory(next); const noSpawn = vi.fn(() => { throw new Error("Duplicate Runner launch"); }); const recoveryOptions: Parameters[0] = { ...options, stateDirectory: next, - authorityRunId: policy === "legacy" ? undefined : target.runId, + get authorityRunId() { return policy === "legacy" ? undefined : authorityRunId; }, recoveryPurpose: policy === "cleanup" ? "stop_cleanup" : undefined, prpIdentity: target, runnerArtifact: readRunnerdArtifactBinding(options.runnerBinary), runnerStateDirectory: join(directory, "runner"), readRunnerState: async () => JSON.parse(await readFile(join(directory, "runner/runner-state.json"), "utf8")), prepareExternalRunnerState: async () => { throw new Error("Verified live handoff must not restore a backup"); }, - runnerProcessLauncher: noSpawn, warmTransitionRegistrationMode: mode === "listen" ? "routed_same_endpoint" : "routed_connect", + runnerProcessLauncher: noSpawn, warmTransitionRegistrationMode: mode !== "connect" ? "routed_same_endpoint" : "routed_connect", resumeProviderSession: { driverSessionId: String((opened.thread as { id: string }).id), providerSessionId: String((opened.thread as { sessionId: string }).sessionId) }, adoptExistingRunner: { pid: runnerPid, processGroupId: runnerPid, startedAt, isAlive: () => !dead(runnerPid!), signal: signal => { if (dead(runnerPid!)) return true; process.kill(-runnerPid!, signal); return true; } } }; @@ -263,12 +275,20 @@ it.each([["connect", "owned"], ["listen", "owned"], ["connect", "legacy"], ["lis else { await within(replacement.transport.request("turn/start", { input: [{ type: "text", text: "Work after recovered attachment" }] }), "target turn"); for await (const event of replacement.transport.notifications()) if (event.method === "turn/completed") break; + if (mode === "external") { + const next = { ...target, runId: "next-run", turnId: "next-turn", itemId: "next-item" }; + authorityRunId = next.runId; + await within(replacement.transport.attachRun!(next), "ordinary external warm followup"); + expect(currentCore.store.state.identity).toEqual(next); + await within(replacement.transport.request("turn/start", { input: [{ type: "text", text: "Work after ordinary external attachment" }] }), "ordinary target turn"); + for await (const event of replacement.transport.notifications()) if (event.method === "turn/completed") break; + } } expect(noSpawn).not.toHaveBeenCalled(); const effects = (await readFile(callLog, "utf8")).trim().split(/\r?\n/); expect(effects.filter(line => line === "process-start")).toHaveLength(1); expect(effects.filter(line => line === "thread/start")).toHaveLength(1); - expect(effects.filter(line => line === "turn/start")).toHaveLength(policy === "cleanup" ? 0 : 1); + expect(effects.filter(line => line === "turn/start")).toHaveLength(policy === "cleanup" ? 0 : mode === "external" ? 2 : 1); } catch (error) { const physical = JSON.parse(await readFile(join(directory, "runner/runner-state.json"), "utf8").catch(() => "{}")); throw new Error(`${mode}/${boundary}: ${String(error)}; ${JSON.stringify({