From 7a203274b73c0c08c281b97b3002d8c90cdc8665 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 18:44:45 -0500 Subject: [PATCH] fix: retain PRP attachment identity during active recovery Carry all six verified durable identity fields into remote reattachment after a healthy warm handoff. Preserve the Runner mismatch guard. Cover opaque attachment IDs in the server factory and real Runner authentication after restoring an active second run, with no duplicate launch or uncertain tool redispatch. Co-Authored-By: Paperclip --- doc/DEVELOPING.md | 2 + doc/execution-semantics.md | 10 +- .../src/live/runnerd-codex-transport.test.ts | 116 ++++++++++++++++++ .../native-session-executor.test.ts | 2 +- .../native-runtime/native-session-executor.ts | 10 ++ 5 files changed, 137 insertions(+), 3 deletions(-) diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index adef62fa6af..d23bd27c1d4 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2783,6 +2783,8 @@ Control-port tests also cover terminal snapshots before accepted results and result-replay contention with the status writer's coordinator-first lock order. Remote adoption tests must invoke actual external-state preparation. Inspecting transport options alone does not exercise the resumed-lease harness checks. +Include a healthy warm handoff before the crash so recovery exercises the +handoff's turn and item attachment IDs as well as first-run identities. Run it against an isolated database with `PAPERCLIP_TEST_DATABASE_URL` when the host cannot start embedded Postgres: diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 6239b21ca02..be3b5acb9ff 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2707,8 +2707,9 @@ public run projections and contains no session parameters or credentials. A terminal snapshot can arrive before result acceptance. That snapshot publishes neither a continuation pointer nor a handled publication receipt. The existing post-acceptance checkpoint completes publication in its normal transaction. -The first production release containing recovery readers and redaction for both -private recovery fields is the disk-independent rollback floor. Earlier +The first production release containing recovery readers, complete attachment +identity handling and redaction for both private recovery fields is the +disk-independent rollback floor. Earlier development-reader probes do not establish a published rollback target. Active remote adoption and compatible idle warm reuse carry an existing verified @@ -2718,6 +2719,11 @@ authenticates the durable PRP authority. A provider lease acquisition described `resumed` does not establish that this already-running harness was suspended. Ordinary resumed leases retain their verified suspension and backup checks. +Reattachment keeps the complete durable PRP binding, including its turn and item +attachment IDs. A healthy warm handoff can create new attachment IDs within the +retained Runner. Replacing that run's controller must preserve those IDs; deriving +them again from the business run ID would request an invalid authority rotation. + For per-turn direct sandbox recovery, the executor reads the retained Runner's suspended lifecycle through the recorded sandbox lease and validates its complete identity and lack of pending delivery/cleanup. Only a lifecycle projection is diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 68fc16f53f7..37c6011ffe2 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -7646,6 +7646,122 @@ it("rotates PRP authority in place for a warm cross-run attachment", async () => } }, 30_000); +it("recovers an active second warm business run with its opaque attachment identity", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-warm-second-active-")); + const firstRoot = join(root, "first-controller"), nextRoot = join(root, "next-controller"); + const server = createServer(); + let authority: DurablePrpControlPlane | null = null, generation = 0; + let checkpoint: durableControlPlane.DurablePrpCheckpoint | undefined; + const registered = new WeakSet(); + server.on("upgrade", (request, socket, head) => authority + ? authority.handleUpgrade(request, socket, "/runner", head) : socket.destroy()); + await new Promise(done => server.listen(0, "127.0.0.1", done)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing owned Runner route"); + const registration = async (core: DurablePrpControlPlane) => { + if (!registered.has(core)) { + core.setCheckpointPersistence(async value => { checkpoint = value; }); + registered.add(core); + } + authority = core; const current = ++generation; + return { connectUrl: `ws://127.0.0.1:${address.port}/runner`, release: async () => { if (generation === current) authority = null; } }; + }; + const initialIdentity = { runnerInstanceId: "runner-warm-second-active", environmentLeaseId: "lease-warm-second-active", + runId: "run-warm-first-healthy", normalizedSessionId: "session-warm-second-active", turnId: "turn-first", itemId: "item-first" }; + const options = { runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(root, "--emit-tool-call", "--durable-tool-ids"), stateDirectory: firstRoot, + prpIdentity: initialIdentity, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 }, controlPlaneRegistration: registration }; + const first = createCapabilityRunnerdCodexTransport(options); + const within = async (label: string, promise: Promise) => { + let timer: ReturnType | undefined; + try { return await Promise.race([promise, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`warm active recovery stage timed out: ${label}`)), 5_000); })]); } + finally { clearTimeout(timer); } + }; + const waitComplete = async (transport: typeof first.transport) => { + for await (const notification of transport.notifications()) if (notification.method === "turn/completed") return; + }; + let calls = 0, runnerPid: number | null = null, providerPid: number | null = null; + let recovered: ReturnType | undefined; + const result = { ok: true, task: { id: "task-1" } }; + first.transport.setServerRequestHandler(async () => ++calls === 1 + ? { success: true, contentItems: [{ type: "inputText", text: JSON.stringify(result) }] } + : new Promise>(() => undefined)); + try { + const opened = await within("first thread/start", first.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [{ name: "get_task_context", description: "Read current task", + inputSchema: { type: "object", properties: {}, additionalProperties: false } }] })); + runnerPid = first.evidence().runnerPid; + providerPid = first.evidence().codexPid; + expect(providerPid).toEqual(expect.any(Number)); + const providerStart = execFileSync("ps", ["-p", String(providerPid), "-o", "lstart="], { encoding: "utf8" }).trim(); + const startedAt = first.transport.processInfo().startedAt; + const firstCompleted = waitComplete(first.transport); + await within("first turn/start", first.transport.request("turn/start", { input: [{ type: "text", text: "first healthy business run" }] })); + await within("healthy first turn completion", firstCompleted); + const attachment = { runId: "run-warm-second-active", turnId: `turn_attachment_${randomUUID()}`, itemId: `item_attachment_${randomUUID()}` }; + await within("warm second business attachment", first.transport.attachRun!(attachment)); + const active = await within("second active turn/start", first.transport.request("turn/start", { input: [{ type: "text", text: "second business run held at tool" }] })); + await vi.waitFor(() => expect(calls).toBe(2), { timeout: 5_000 }); + const activeCore = authority!; + const savedIdentity = { ...activeCore.store.state.identity }; + expect(savedIdentity).toEqual({ ...initialIdentity, ...attachment }); + await activeCore.flushCheckpoint(); + await within("controller detach", first.detachControllerForRestart()); + expect(() => process.kill(runnerPid!, 0)).not.toThrow(); + checkpoint!.restoreSessionDirectory(nextRoot); + const noSpawn = vi.fn(() => { throw new Error("Duplicate Runner launch"); }); + const openedThread = opened.thread as Record; + const activeTurn = active.turn as Record; + expect(openedThread.id).toEqual(expect.any(String)); + expect(activeTurn.id).toEqual(expect.any(String)); + const shared = { ...options, stateDirectory: nextRoot, + resumeProviderSession: { driverSessionId: String(openedThread.id), providerSessionId: String(openedThread.sessionId) }, + resumeActiveTurnId: String(activeTurn.id), runnerArtifact: readRunnerdArtifactBinding(options.runnerBinary), + runnerStateDirectory: join(firstRoot, "runner"), readRunnerState: async () => JSON.parse(await readFile(join(firstRoot, "runner/runner-state.json"), "utf8")), + prepareExternalRunnerState: async () => {}, archiveExternalRunnerState: async () => { throw new Error("Active authority must not rotate"); }, + runnerProcessLauncher: noSpawn, adoptExistingRunner: { pid: runnerPid!, processGroupId: runnerPid, startedAt, + isAlive: () => { try { process.kill(runnerPid!, 0); return true; } catch { return false; } }, + signal: (signal: NodeJS.Signals) => { try { process.kill(runnerPid!, signal); return true; } catch { return false; } } } }; + // The old server's synthetic pair must continue failing the strict Runner guard. + const wrong = createCapabilityRunnerdCodexTransport({ ...shared, prpIdentity: { ...savedIdentity, + turnId: `turn-${attachment.runId}`, itemId: `item-${attachment.runId}` } }); + await expect(wrong.transport.request("thread/read", {})).rejects.toThrow("PRP recovery identity does not match the durable session binding"); + await wrong.detachControllerForRestart(); + // Leave the pending tool outcome unknown. Reattachment must authenticate + // and preserve it without fabricating a receipt or redispatching its handler. + recovered = createCapabilityRunnerdCodexTransport({ ...shared, prpIdentity: savedIdentity }); + await within("replacement thread/read", recovered.transport.request("thread/read", {})); + expect(recovered.evidence()).toMatchObject({ runnerPid, runnerExited: false }); + expect(authority!.activeRunnerConnectionCount()).toBe(1); + expect(() => process.kill(providerPid!, 0)).not.toThrow(); + expect(execFileSync("ps", ["-p", String(providerPid), "-o", "lstart="], { encoding: "utf8" }).trim()).toBe(providerStart); + expect(calls).toBe(2); + expect(authority!.store.state.commands.filter(command => command.type === "semantic_tool.result")).toHaveLength(0); + expect(recovered.transport.processInfo().startedAt).toBe(startedAt); + expect(authority!.store.state.identity).toEqual(savedIdentity); + expect(authority!.store.state.commands.filter(command => command.type === "turn.start")).toHaveLength(1); + expect(noSpawn).not.toHaveBeenCalled(); + } catch (error) { + const saved = JSON.parse(await readFile(join(firstRoot, "runner/runner-state.json"), "utf8").catch(() => "{}")); + const stderr = await readFile(join(firstRoot, "diagnostics/runnerd.stderr.log"), "utf8").catch(() => ""); + throw new Error(`${String(error)}; owned protocol evidence ${JSON.stringify({ + transport: recovered?.evidence(), connections: authority?.activeRunnerConnectionCount(), + identity: authority?.store.state.identity, commands: authority?.store.state.commands.map(c => ({type:c.type,status:c.status})), + events: authority?.store.state.committedEvents.map(e => e.eventType).slice(-8), + runner: { lifecycle:saved.lifecycle, runId:saved.runId, turnId:saved.turnId, itemId:saved.itemId, + outbox:saved.outbox?.length }, stderr:stderr.slice(-1500), + })}`); + } finally { + await (recovered ? within("recovered cleanup", recovered.transport.close()).catch(() => undefined) : undefined); + await within("first cleanup", first.transport.close()).catch(() => undefined); + if (providerPid) { try { process.kill(providerPid, "SIGKILL"); } catch {} } + await (authority as DurablePrpControlPlane | null)?.stop(); + if (runnerPid) { try { process.kill(-runnerPid, "SIGKILL"); } catch {} } + server.closeAllConnections(); + if (server.listening) await new Promise(done => server.close(() => done())); + await rm(root, { recursive: true, force: true }); + } +}, 30_000); + it("waits for a warm runner to re-authenticate before probing attachment readiness", async () => { const stateDirectory = await mkdtemp( join(tmpdir(), "runnerd-warm-reattach-before-probe-"), diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index c0cc1cf41d0..ebe0791ffc6 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -12581,7 +12581,7 @@ describe("runnerd provider runtime wiring", () => { } as NativeExecutionInputV1; const identity = { runId: current.binding.runId, normalizedSessionId: current.session.normalizedSessionId!, runnerInstanceId: `admitted-${provider}`, environmentLeaseId: "admitted-lease", - turnId: `turn-${current.binding.runId}`, itemId: `item-${current.binding.runId}` }; + turnId: `turn_attachment_${randomUUID()}`, itemId: `item_attachment_${randomUUID()}` }; const root = await initializeScopedRunnerRoot(current, identity.runnerInstanceId); await mkdir(join(root, "control-plane"), { recursive: true }); await writeFile(join(root, "control-plane", "control-plane-state.json"), JSON.stringify(durableControlPlaneState(identity))); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index e6418e282ce..650a5d8b3b1 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -13910,6 +13910,16 @@ async function createRunnerdBackendWithinSessionClaim( normalizedSessionId: input.remoteWarmRecovery.identity.normalizedSessionId, turnId: input.remoteWarmRecovery.identity.turnId, itemId: input.remoteWarmRecovery.identity.itemId, + } : input.restartRecovery?.kind === "reattach_remote_runner" ? { + // Reattachment verified all six durable fields against the surviving + // Runner. Warm handoff uses opaque attachment turn/item IDs; deriving + // replacements from runId would request a different same-run authority. + runnerInstanceId: durableIdentity!.runnerInstanceId as string, + environmentLeaseId: durableIdentity!.environmentLeaseId as string, + runId: durableIdentity!.runId as string, + normalizedSessionId: durableIdentity!.normalizedSessionId as string, + turnId: durableIdentity!.turnId as string, + itemId: durableIdentity!.itemId as string, } : { runnerInstanceId: effectiveRunnerInstanceId, environmentLeaseId: effectiveEnvironmentLeaseId,