From fb2e07ebdd17a813965c4e415696e72cc6f9eed2 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 5 Nov 2026 13:05:56 -0800 Subject: [PATCH] Wait for physical authority before activating hosted warm routes (#19277) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Core can give two consecutive runs, A and B, the same hosted Runner so B can continue A’s conversation. > - During that handoff, Core can record that B is next while the Runner still answers as A. > - If Core restarts at that point and tries to connect for B, the Runner rejects the connection. B’s next turn stalls. > - This PR makes Core finish the handoff over A’s connection and check which run the Runner now reports. > - Once the Runner reports B and authenticates B’s connection, B can continue on the same Runner. ## Linked Issues or Issue Description - Refs #17390 for hosted Runner reuse. - Refs #19157 for recovering a stopped Runner session before attaching the next run. ## What Changed - If the Runner still reports A, Core uses A’s connection to acknowledge the result of `run.attach`. This tells the Runner that Core saved the handoff and that it can switch to B. - Core reads the Runner’s saved state and checks it against the recorded handoff. If the Runner already reports B for that handoff, Core connects for B directly. Core waits for B’s connection to authenticate before sending its work. This applies both during an ordinary handoff and after Core restarts. ## Verification - Run `pnpm --filter @paperclipai/paperclip-runner exec vitest run src/live/runnerd-staged-handoff.test.ts` with the native test binaries built. The regression replaces Core at four points in the handoff. Before the fix, Core tried to connect for B while the Runner still reported A. - With the fix, B and a third run each start and finish one turn. The test observes one Runner process, one controlled Codex provider process, and one conversation thread throughout. - All seven handoff cases passed. GitHub CI passed after retrying an unrelated port collision and interrupted jobs. Standard Greptile scored the commit 5/5. [CI run](https://github.com/paperclipai/paperclip/actions/runs/42070818838) ## Risks - The host must be able to read the Runner’s state. If Core cannot verify the switch before the existing deadline, it leaves the handoff pending and sends no work for B. Live hosted verification is still separate. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), high reasoning effort, through Codex with repository tools and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- .../paperclip-runner/docs/durable-recovery.md | 8 +++++ .../src/live/runnerd-codex-transport.ts | 16 +++++---- .../src/live/runnerd-staged-handoff.test.ts | 36 ++++++++++++++----- 3 files changed, 45 insertions(+), 15 deletions(-) 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({