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,