Wait for physical authority before activating hosted warm routes (#19277)

## 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 <noreply@paperclip.ing>

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.

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,

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<void>(done => { reached = done; });
       const cores = new Set<DurablePrpControlPlane>();
       const registered = new WeakSet<DurablePrpControlPlane>();
-      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<typeof createCapabilityRunnerdCodexTransport> | 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<typeof createCapabilityRunnerdCodexTransport>[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({