diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 84141d4f0d5..5ee7e4e098d 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2362,6 +2362,22 @@ All of these must be true: - Each quarantined root for the scope passes the same journal and local Runner evidence checks for the same owner. +A local per-turn Runner can end the same way: the result was accepted, but its +`runner.suspend` command did not finish, so its own receipt is still +unsuspended. The next local run retires that owner with the same checks and +these differences: + +- The Runner's own receipt may be unsuspended. It may not be unreadable or + name another owner. +- Only the Runner's own lifecycle commands (`runner.suspend`, `runner.drain`, + `runner.shutdown`) may be unfinished. All other commands must be complete. +- The prior run's recorded Runner process is proven stopped by exact PID and + start time, as in restart recovery. A PID that now belongs to another + process counts as stopped. A live recorded process, a live process group, a + live recorded provider process, a missing identity, or a start time that + cannot be compared keeps the rejection, for example + `retirement_refused=runner_process_alive`. A later Retry checks again. + The provider session then follows the normal resume-or-replace policy. A resumed sandbox, a warm session, a plugin task, a restart claim, or any missing proof keeps the existing `prior_owner_terminal_state_indeterminate` 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 f12aad1871f..634bdbd2b77 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 @@ -185,7 +185,11 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered runner: { execute: vi.fn(async () => { throw new Error("admission must not reach the sandbox"); }) }, sandboxLeaseAcquisition: { outcome, providerLeaseId: `sandbox-${runId}` } } as never, }).then(() => "constructed", (error: Error) => error.message); - return { runId, logs, start, selectedRunId: selected?.id ?? null }; + // A local Runner keeps its controller root on this host between turns. + const startLocal = () => createRunnerdBackend({ + db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); }, + }).then(() => "constructed", (error: Error) => error.message); + return { runId, logs, start, startLocal, selectedRunId: selected?.id ?? null }; } async function quarantined() { @@ -465,6 +469,78 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); }, 60_000); + /** A spawned process that stands in for the prior local Runner. */ + async function runnerProcess() { + const child = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60_000)"], { stdio: "ignore" }); + await new Promise(resolve => child.once("spawn", resolve)); + const startedAt = await readProcessStartedAt(child.pid!); + expect(startedAt).toBeTruthy(); + return { child, pid: child.pid!, processStartedAt: new Date(startedAt!) }; + } + + async function stoppedRunnerProcess() { + const runner = await runnerProcess(); + const exited = new Promise(resolve => runner.child.once("exit", resolve)); + runner.child.kill("SIGKILL"); + await exited; + return runner; + } + + /** The state a local per-turn Runner leaves when its result was accepted but + * its suspend command never finished: the controller journal holds a pending + * runner.suspend and the Runner's own receipt is still `ready`. */ + async function seedUnsuspendedLocalOwner(runner: { pid: number; processStartedAt: Date }) { + const prior = await seedRecoveredPriorRun("pending-command"); + await db.update(heartbeatRuns).set({ processPid: runner.pid, processGroupId: null, processStartedAt: runner.processStartedAt }) + .where(eq(heartbeatRuns.id, prior.runId)); + const followUp = await admitFollowUp(prior); + // Learn the scope key from a held attempt, then keep its root in place. + expect(await followUp.start()).toMatch(/^runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate/); + const [copy] = await quarantined(); + const root = join(stateDirectory, copy!.split(".")[0]!); + await rename(join(stateDirectory, "quarantine", copy!), root); + await rm(join(stateDirectory, "quarantine"), { recursive: true, force: true }); + await mkdir(join(root, "runner"), { recursive: true }); + await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.state.v1", lifecycle: "ready", ...prior.identity })); + return { prior, followUp, root }; + } + + it("retires a finished local owner whose Runner stopped before it proved suspension", async () => { + const { prior, followUp, root } = await seedUnsuspendedLocalOwner(await stoppedRunnerProcess()); + expect(await followUp.startLocal()).toBe(ADMITTED); + expect(await quarantined()).toEqual([]); + expect(existsSync(join(root, "runner"))).toBe(false); + expect(await archivedOwners(root)).toEqual([prior.runId]); + expect(followUp.logs.join("")).toContain(`Run ${prior.runId} finished without a saved runner checkpoint`); + }, 60_000); + + it("keeps a finished local owner held while its recorded Runner process is alive", async () => { + const runner = await runnerProcess(); + try { + const { followUp } = await seedUnsuspendedLocalOwner(runner); + expect(await followUp.startLocal()).toBe(stranded("runner_process_alive")); + expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); + } finally { + runner.child.kill("SIGKILL"); + } + }, 60_000); + + it("keeps a finished local owner held without a recorded Runner process identity", async () => { + const { prior, followUp } = await seedUnsuspendedLocalOwner(await stoppedRunnerProcess()); + await db.update(heartbeatRuns).set({ processPid: null, processStartedAt: null }).where(eq(heartbeatRuns.id, prior.runId)); + expect(await followUp.startLocal()).toBe(stranded("runner_process_identity_missing")); + }, 60_000); + + it("keeps a stopped local owner held by an unfinished non-lifecycle command", async () => { + const { followUp, root } = await seedUnsuspendedLocalOwner(await stoppedRunnerProcess()); + const path = join(root, "control-plane", "control-plane-state.json"); + const state = JSON.parse(await readFile(path, "utf8")); + state.commands.find((command: { type: string }) => command.type === "turn.start").status = "pending"; + await writeFile(path, JSON.stringify(state)); + expect(await followUp.startLocal()).toBe(stranded("journal_commands_unsettled(turn.start.pending)")); + }, 60_000); + it.each(["uncommitted result", "committed without result", "live controller lease", "unexpired controller lease", "recovery pending", "pending command", "unanswered tool input", "foreign quarantine"] as const)("keeps the prior owner held with %s", async scenario => { const prior = await seedRecoveredPriorRun(scenario === "pending command" ? "pending-command" diff --git a/server/src/services/native-runtime/native-restart-recovery.test.ts b/server/src/services/native-runtime/native-restart-recovery.test.ts index fd79245b6e5..d1d8bdae7c5 100644 --- a/server/src/services/native-runtime/native-restart-recovery.test.ts +++ b/server/src/services/native-runtime/native-restart-recovery.test.ts @@ -5,8 +5,44 @@ import { evaluateNativeControllerTakeover, evaluateNativeProviderProcesses, nextNativeProviderAttempt, + proveNativeLocalRunnerStopped, } from "./native-restart-recovery.js"; +describe("local runner stop proof", () => { + const startedAt = new Date("2026-11-06T17:58:54.000Z"); + const base = { processPid: 72454, processGroupId: 72454, processStartedAt: startedAt }; + const probe = (alive: number[], starts: Record, groups: number[] = []) => ({ + isProcessAlive: (pid: number) => alive.includes(pid), + isProcessGroupAlive: (group: number) => groups.includes(group), + readProcessStartedAt: async (pid: number) => starts[pid] ?? null, + }); + + it.each([ + ["the runner and its group exited", base, probe([], {}), { stopped: true }], + ["the runner PID now belongs to another process", base, + probe([72454], { 72454: new Date(startedAt.getTime() + 60_000) }), { stopped: true }], + ["the recorded runner is alive", base, probe([72454], { 72454: startedAt }), + { stopped: false, reason: "runner_process_alive" }], + ["the live PID start time is unknown", base, probe([72454], {}), + { stopped: false, reason: "runner_process_identity_unverifiable" }], + ["the runner group is alive", base, probe([], {}, [72454]), + { stopped: false, reason: "runner_process_group_alive" }], + ["no runner PID was recorded", { ...base, processPid: null }, probe([], {}), + { stopped: false, reason: "runner_process_identity_missing" }], + ["no runner start time was recorded", { ...base, processStartedAt: null }, probe([], {}), + { stopped: false, reason: "runner_process_identity_missing" }], + ] as const)("decides when %s", async (_case, identity, checks, expected) => { + await expect(proveNativeLocalRunnerStopped({ ...identity, ...checks })).resolves.toEqual(expected); + }); + + it("refuses while a recorded provider process is alive", async () => { + const providerStartedAt = new Date("2026-11-06T17:58:56.000Z"); + await expect(proveNativeLocalRunnerStopped({ ...base, + providerProcesses: [{ pid: 72460, processStartedAt: providerStartedAt }], + ...probe([72460], { 72460: providerStartedAt }) })).resolves.toEqual({ stopped: false, reason: "provider_process_alive" }); + }); +}); + describe("native restart recovery classification", () => { it.each([false, true])("recognizes verified staged remote evidence with target snapshot=%s", hasCheckpoint => { const evidence = { remoteSandbox: true, runnerPidAlive: false, runnerGroupAlive: false, diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index efb32887539..3f53cba59b1 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -335,6 +335,59 @@ export async function evaluateNativeProviderProcesses(input: { }; } +export type NativeLocalRunnerStopProof = + | { stopped: true } + | { + stopped: false; + reason: + | "runner_process_identity_missing" + | "runner_process_alive" + | "runner_process_identity_unverifiable" + | "runner_process_group_alive" + | "provider_process_alive" + | "provider_process_identity_unverifiable"; + }; + +/** + * Proves that a finished local runner and its recorded provider processes + * stopped, with the same exact PID and start-time checks as restart recovery. + * A PID that now belongs to a different process counts as stopped. A live + * process with the recorded identity, a live process group, or a live PID + * whose start time cannot be compared is not proof. + */ +export async function proveNativeLocalRunnerStopped(input: { + processPid: number | null; + processGroupId: number | null; + processStartedAt: Date | null; + providerProcesses?: readonly NativeProviderProcessIdentity[]; + isProcessAlive?: (pid: number) => boolean; + isProcessGroupAlive?: (processGroupId: number) => boolean; + readProcessStartedAt?: (pid: number) => Promise; +}): Promise { + const refuse = (reason: Extract["reason"]) => + ({ stopped: false as const, reason }); + if (!input.processPid || !Number.isInteger(input.processPid) || input.processPid <= 0 || !input.processStartedAt) { + return refuse("runner_process_identity_missing"); + } + const runner = await evaluateNativeProviderProcesses({ + identities: [{ pid: input.processPid, processStartedAt: input.processStartedAt }], + isProcessAlive: input.isProcessAlive, + readProcessStartedAt: input.readProcessStartedAt, + }); + if (runner.livePids.length > 0) return refuse("runner_process_alive"); + if (runner.ambiguousLivePids.length > 0) return refuse("runner_process_identity_unverifiable"); + const groupAlive = input.isProcessGroupAlive ?? processGroupIsAlive; + if (input.processGroupId && groupAlive(input.processGroupId)) return refuse("runner_process_group_alive"); + const providers = await evaluateNativeProviderProcesses({ + identities: (input.providerProcesses ?? []).filter(identity => identity.pid !== input.processPid), + isProcessAlive: input.isProcessAlive, + readProcessStartedAt: input.readProcessStartedAt, + }); + if (providers.livePids.length > 0) return refuse("provider_process_alive"); + if (providers.ambiguousLivePids.length > 0) return refuse("provider_process_identity_unverifiable"); + return { stopped: true }; +} + export function classifyNativeRunnerRecoveryEvidence(input: { runnerPidAlive: boolean; runnerGroupAlive: boolean; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 27f2abb0856..c49bcdec555 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -290,6 +290,7 @@ import { import { currentNativeControllerIdentity, evaluateNativeControllerTakeover, + proveNativeLocalRunnerStopped, isCurrentNativeControllerProcess, nextNativeProviderAttempt, type NativeControllerIdentity, @@ -5288,15 +5289,25 @@ function moveNativeContinuityState(root: string, archiveRoot: string): void { * Runner outbox can hold only post-terminal telemetry. Local lifecycle * evidence, even an unsuspended receipt, and unknown entries are owner * claims that this proof never overrides. + * + * `localRunner` names a local Runner that wrote its own lifecycle receipt in + * this root. Its unsuspended receipt and its own unfinished lifecycle commands + * (suspend, drain, shutdown) are claims of that one process. The caller must + * separately prove that process stopped before it relies on this projection. */ type SettledRunnerdProjection = { bytes: string; refused?: never } | { bytes?: never; refused: string }; -function readSettledRunnerdProjection(root: string, identity: Record): SettledRunnerdProjection { +const LOCAL_RUNNER_LIFECYCLE_COMMANDS = new Set(["runner.suspend", "runner.drain", "runner.shutdown"]); + +function readSettledRunnerdProjection(root: string, identity: Record, + options: { localRunner?: boolean } = {}): SettledRunnerdProjection { const refuse = (refused: string): SettledRunnerdProjection => ({ refused }); try { if (!isSafeNativeStateDirectory(root)) return refuse("root_unsafe"); const lifecycle = runnerdAuthorityLifecycle(root, identity as RunnerdDurableIdentity); - if (lifecycle !== "absent") return refuse(`runner_lifecycle_${lifecycle}`); + if (lifecycle !== "absent" && !(options.localRunner && lifecycle === "not_suspended")) { + return refuse(`runner_lifecycle_${lifecycle}`); + } // Every legitimate writer's entry is classified in native-session-root-entries.ts. // Held entries are known evidence that needs its own settlement; anything // else is writer drift. Name each kind separately. @@ -5308,6 +5319,11 @@ function readSettledRunnerdProjection(root: string, identity: Record !LOCAL_RUNNER_LIFECYCLE_COMMANDS.has(String(command.type)) || command.status === "completed") + : commands; + const settledControl = settledCommands === commands ? control : { ...control, commands: settledCommands }; const answered = new Set(commands .filter(command => command.type === "semantic_tool.result" && command.status === "completed") .map(command => record(command.payload).sourceEventId)); @@ -5317,11 +5333,11 @@ function readSettledRunnerdProjection(root: string, identity: Record record(control.identity)[key] !== value)) { return refuse("journal_identity_mismatch"); } - if (!isSettledRemoteWarmControlState(control, identity)) { + if (!isSettledRemoteWarmControlState(settledControl, identity)) { // Closed vocabulary only: known command types and statuses, never payloads. - const pending = commands.filter(command => command.status !== "completed") + const unsettled = settledCommands.filter(command => command.status !== "completed") .map(command => `${retirementReasonName("commandType", command.type)}.${retirementReasonName("commandStatus", command.status)}`); - return refuse(pending.length > 0 ? `journal_commands_unsettled(${[...new Set(pending)].slice(0, 4).join(",")})` + return refuse(unsettled.length > 0 ? `journal_commands_unsettled(${[...new Set(unsettled)].slice(0, 4).join(",")})` : "journal_terminal_unsettled"); } if ((control.committedEvents as unknown[]).map(record).some(event => @@ -5365,6 +5381,12 @@ function retirementReasonName(kind: keyof typeof RETIREMENT_REASON_NAMES, value: * that authority as a continuity break instead. This run then bootstraps its * own authority, and the provider session follows the ordinary * resume-or-replace policy. Any missing proof keeps the original rejection. + * + * A local per-turn Runner can end the same way: the result is accepted but its + * suspend command does not finish, so its own receipt stays unsuspended. That + * receipt is a claim of one recorded Runner process. Once that process and its + * recorded provider processes are proven stopped by exact PID and start time, + * the claim has no owner left and the same retirement applies. */ async function retireFinishedPerTurnRunnerdOwner(input: { db: Db; @@ -5379,11 +5401,14 @@ async function retireFinishedPerTurnRunnerdOwner(input: { // existing recovery path reads. Plugin tasks own a separate saved checkpoint. const refuse = (refused: string) => ({ retired: false as const, refused }); if (input.execution.session.lifecyclePolicy.mode !== "per_turn") return refuse("lifecycle_not_per_turn"); - if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner || target.runnerTask) { - return refuse("target_not_sandbox_runner"); - } - if (!["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) { - return refuse(`sandbox_${retirementReasonName("sandboxOutcome", target.sandboxLeaseAcquisition?.outcome)}`); + const localRunner = !target || target.kind === "local"; + if (!localRunner) { + if (target.kind !== "remote" || target.transport !== "sandbox" || !target.runner || target.runnerTask) { + return refuse("target_not_sandbox_runner"); + } + if (!["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) { + return refuse(`sandbox_${retirementReasonName("sandboxOutcome", target.sandboxLeaseAcquisition?.outcome)}`); + } } const identity: Record = {}; for (const key of ["runId", "normalizedSessionId", "runnerInstanceId", "environmentLeaseId", "turnId", "itemId"]) { @@ -5391,7 +5416,7 @@ async function retireFinishedPerTurnRunnerdOwner(input: { if (typeof value !== "string" || !value) return refuse("durable_identity_incomplete"); identity[key] = value; } - const projection = readSettledRunnerdProjection(input.root, identity); + const projection = readSettledRunnerdProjection(input.root, identity, { localRunner }); if (projection.refused !== undefined) return refuse(projection.refused); const controlBytes = projection.bytes; const { binding } = input.execution; @@ -5423,6 +5448,22 @@ async function retireFinishedPerTurnRunnerdOwner(input: { remoteLeaseFenced: readNativeControllerRecovery(profile.nativeControllerRecovery, profile.sessionCheckpoint) !== null }); if (!stopped.allowed) return refuse(`controller_${retirementReasonName("controllerReason", stopped.reason)}`); const provenFinishedHere = finishedHere(); + const proveLocalRunnerStopped = async () => { + const checkpointProcess = record(record(profile.sessionCheckpoint).process); + const providerProcesses = [["providerPid", "providerProcessStartedAt"], ["codexPid", "codexProcessStartedAt"], + ["sidecarPid", "sidecarProcessStartedAt"], ["agentPid", "agentProcessStartedAt"]].flatMap(([pidField, startedField]) => { + const pid = checkpointProcess[pidField!]; + if (typeof pid !== "number" || !Number.isInteger(pid) || pid <= 0) return []; + const startedAt = typeof checkpointProcess[startedField!] === "string" ? new Date(String(checkpointProcess[startedField!])) : null; + return [{ pid, processStartedAt: startedAt && !Number.isNaN(startedAt.getTime()) ? startedAt : null }]; + }); + return proveNativeLocalRunnerStopped({ processPid: prior!.run.processPid, processGroupId: prior!.run.processGroupId, + processStartedAt: prior!.run.processStartedAt, providerProcesses }); + }; + if (localRunner) { + const runnerStopped = await proveLocalRunnerStopped(); + if (!runnerStopped.stopped) return refuse(runnerStopped.reason); + } // 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. @@ -5431,13 +5472,14 @@ async function retireFinishedPerTurnRunnerdOwner(input: { if (!isSafeNativeStateDirectory(quarantine)) return refuse("quarantine_unsafe"); for (const name of readdirSync(quarantine)) { if (!name.startsWith(`${basename(input.root)}.`)) continue; - const copy = readSettledRunnerdProjection(resolve(quarantine, name), identity); + const copy = readSettledRunnerdProjection(resolve(quarantine, name), identity, { localRunner }); if (copy.refused !== undefined) return refuse(`quarantine_${copy.refused}`); } } // Database reads can await. Retire only the exact evidence proven above. - if (readSettledRunnerdProjection(input.root, identity).bytes !== controlBytes) return refuse("journal_changed"); + if (readSettledRunnerdProjection(input.root, identity, { localRunner }).bytes !== controlBytes) return refuse("journal_changed"); if (provenFinishedHere && !finishedHere()) return refuse("controller_restarted_here"); + if (localRunner && !(await proveLocalRunnerStopped()).stopped) return refuse("runner_process_restarted"); const archivedAt = Date.now(); const archiveRoot = resolve(input.root, "continuity-breaks", `${archivedAt}-${randomUUID()}`); mkdirSync(archiveRoot, { recursive: true, mode: 0o700 });