diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index c69c928eb65..fccd60629d4 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1863,7 +1863,10 @@ All of these must be true: - The prior run is terminal and its coordinator committed a result. - The coordinator has no lease owner and no recovery state. -- The prior controller is proven stopped. +- The prior controller is proven stopped. The controller runs inside the + webserver, so when this same server process finished the owner, its live + PID proves nothing. The owner's exact controller boot identity, with no + in-process session or startup left for that run, proves it stopped instead. - The journal has a delivered terminal, only completed commands, and an answer for each tool input. - No local Runner lifecycle evidence exists for the old owner. diff --git a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts index 071fdf49e8d..0e5ff1b62e9 100644 --- a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -1,3 +1,4 @@ +import { spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; import { existsSync } from "node:fs"; import { cp, mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises"; @@ -16,6 +17,8 @@ import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; import { buildNativeExecutionInput } from "./native-execution-input.js"; import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; import { createNativeControllerRecovery } from "./native-controller-recovery.js"; +import { currentNativeControllerIdentity } from "./native-restart-recovery.js"; +import { readProcessStartedAt } from "../hot-restart.js"; import { buildNativeExecutionWithCheckpoint, findNativeSessionResumeRun, nativeToolContractFingerprintForTarget, } from "./native-session-resume.js"; @@ -83,7 +86,8 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered * later committed its result and released the controller lease, but no * suspended Runner receipt or failover backup ever reached the controller. * Only the database controller checkpoint and recovery pointer survive. */ - async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" = "settled") { + async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" = "settled", + options: { controller?: { bootId: string; pid: number; processStartedAt: Date } } = {}) { const issueId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID(); await db.insert(issues).values({ id: issueId, companyId, title: "Continue after recovery", assigneeAgentId: agentId, status: "in_progress" }); const [row] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", @@ -104,10 +108,12 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered const execution = executionFor(runId); const identity = { runId, normalizedSessionId: native.normalizedSessionId, runnerInstanceId: native.runnerInstanceId, environmentLeaseId: native.environmentLeaseId, turnId: native.turnId, itemId: native.itemId }; - const owner = { companyId, issueId, runId, authorityRunId: runId, bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; + const priorController = options.controller ?? { bootId: randomUUID(), pid: 2_147_483_000, + processStartedAt: new Date(Date.now() - 60_000) }; + const owner = { companyId, issueId, runId, authorityRunId: runId, bootId: priorController.bootId, generation: 1, leaseOwner: randomUUID() }; await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "observed", - controllerBootId: owner.bootId, controllerGeneration: 1, controllerPid: 2_147_483_000, - controllerProcessStartedAt: new Date(Date.now() - 60_000), leaseOwner: owner.leaseOwner, + controllerBootId: owner.bootId, controllerGeneration: 1, controllerPid: priorController.pid, + controllerProcessStartedAt: priorController.processStartedAt, leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }); const scratch = await mkdtemp(join(tmpdir(), "native-recovered-owner-controller-")); const controller = new DurablePrpControlPlane({ stateDirectory: scratch, identity, @@ -214,6 +220,57 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(followUp.logs.join("")).toContain(`Run ${prior.runId} finished without a saved runner checkpoint`); }, 60_000); + async function keepLocalRootWithoutPointer(prior: Awaited>, + followUp: Awaited>) { + await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + expect(await followUp.start()).toBe(STRANDED); + const [copy] = await quarantined(); + await rename(join(stateDirectory, "quarantine", copy!), join(stateDirectory, copy!.split(".")[0]!)); + await rm(join(stateDirectory, "quarantine"), { recursive: true, force: true }); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, prior.runId)); + const { nativeControllerRecovery: _pointer, ...profile } = run.runnerProfileJson!; + await db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, prior.runId)); + await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + } + + it.each(["recovery pointer", "local journal only"] as const)( + "admits a follow-up in the process that finished the owner: %s", async evidence => { + const prior = await seedRecoveredPriorRun("settled", { controller: await currentNativeControllerIdentity() }); + const followUp = await admitFollowUp(prior); + if (evidence === "local journal only") await keepLocalRootWithoutPointer(prior, followUp); + expect(await followUp.start()).toBe(ADMITTED); + expect(await quarantined()).toEqual([]); + expect(await archivedOwners(await sessionRoot())).toEqual([prior.runId]); + }, 60_000); + + it("admits a local journal whose controller process has exited", async () => { + const prior = await seedRecoveredPriorRun(); + const followUp = await admitFollowUp(prior); + await keepLocalRootWithoutPointer(prior, followUp); + expect(await followUp.start()).toBe(ADMITTED); + }, 60_000); + + it.each(["another live controller", "an earlier controller in this process"] as const)( + "keeps a local journal held while %s may still own it", async scenario => { + const child = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60_000)"], { stdio: "ignore" }); + try { + await new Promise(resolve => child.once("spawn", resolve)); + const self = await currentNativeControllerIdentity(); + const startedAt = await readProcessStartedAt(child.pid!); + expect(startedAt).toBeTruthy(); + const controller = scenario === "another live controller" + ? { bootId: randomUUID(), pid: child.pid!, processStartedAt: new Date(startedAt!) } + : { ...self, bootId: randomUUID() }; + const prior = await seedRecoveredPriorRun("settled", { controller }); + const followUp = await admitFollowUp(prior); + await keepLocalRootWithoutPointer(prior, followUp); + expect(await followUp.start()).toBe(STRANDED); + expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); + } finally { + child.kill("SIGKILL"); + } + }, 60_000); + /** The state left by a follow-up admitted before retirement existed. Its * admission quarantined the rebuilt projection and failed before provider * work; it keeps only the provider checkpoint inherited from the owner. */ diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index d8359ffd31f..79ca54fc5a8 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -179,6 +179,13 @@ export async function currentNativeControllerIdentity(): Promise isCurrentNativeControllerProcess(prior!.owner) && + !activeNativeSessions.has(identity.runId) && !nativeSessionStartups.has(identity.runId); const profile = record(prior.run.runnerProfileJson); - const stopped = await evaluateNativeControllerTakeover({ owner: prior.owner, now: new Date(), - requireStoppedController: true, + const stopped = finishedHere() ? { allowed: true } : await evaluateNativeControllerTakeover({ owner: prior.owner, + now: new Date(), requireStoppedController: true, remoteLeaseFenced: readNativeControllerRecovery(profile.nativeControllerRecovery, profile.sessionCheckpoint) !== null }); if (!stopped.allowed) return false; + const provenFinishedHere = finishedHere(); // Earlier rejected admissions can quarantine the same projection. Each copy // must pass the same proof: a rebuilt root never carries the local Runner // evidence that held an earlier attempt, so that evidence stays decisive. @@ -5266,7 +5270,8 @@ async function retireFinishedPerTurnRunnerdOwner(input: { } } // Database reads can await. Retire only the exact evidence proven above. - if (readSettledRunnerdProjection(input.root, identity) !== controlBytes) return false; + if (readSettledRunnerdProjection(input.root, identity) !== controlBytes || + (provenFinishedHere && !finishedHere())) return false; const archivedAt = Date.now(); const archiveRoot = resolve(input.root, "continuity-breaks", `${archivedAt}-${randomUUID()}`); mkdirSync(archiveRoot, { recursive: true, mode: 0o700 });