diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 5c71fd55f82..e5451eb7ace 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2379,6 +2379,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 8185661619d..e969b20d0c2 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,104 @@ 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: runner.pid, 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.each(["no identity", "no process group"] as const)("keeps a finished local owner held with %s", async scenario => { + const { prior, followUp } = await seedUnsuspendedLocalOwner(await stoppedRunnerProcess()); + await db.update(heartbeatRuns).set(scenario === "no identity" + ? { processPid: null, processGroupId: null, processStartedAt: null } : { processGroupId: null }) + .where(eq(heartbeatRuns.id, prior.runId)); + expect(await followUp.startLocal()).toBe(stranded("runner_process_identity_missing")); + }, 60_000); + + it("retires a finished local owner from its saved stop receipt after settlement cleared the process fields", async () => { + const runner = await stoppedRunnerProcess(); + const { prior, followUp, root } = await seedUnsuspendedLocalOwner(runner); + await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: prior.runId, seq: 900, eventType: "native.local_process_stopped", + stream: "system", payload: { processPid: runner.pid, processGroupId: runner.pid, processStartedAt: runner.processStartedAt.toISOString() } }); + await db.update(heartbeatRuns).set({ processPid: null, processGroupId: null, processStartedAt: null }) + .where(eq(heartbeatRuns.id, prior.runId)); + expect(await followUp.startLocal()).toBe(ADMITTED); + expect(await archivedOwners(root)).toEqual([prior.runId]); + }, 60_000); + + it("keeps a stopped local owner held while a provider process from its run events is alive", async () => { + const provider = await runnerProcess(); + try { + const { prior, followUp } = await seedUnsuspendedLocalOwner(await stoppedRunnerProcess()); + await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: prior.runId, seq: 901, eventType: "provider.event", + stream: "system", payload: { prpEvent: { payload: { processId: provider.pid, + processStartedAt: provider.processStartedAt.toISOString() } } } }); + expect(await followUp.startLocal()).toBe(stranded("provider_process_alive")); + } finally { + provider.child.kill("SIGKILL"); + } + }, 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..52d8d15c9b7 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,48 @@ 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" }], + ["no runner process group was recorded", { ...base, processGroupId: null }, probe([], {}), + { stopped: false, reason: "runner_process_identity_missing" }], + ["a stop receipt proved the runner absent", { ...base, processStartedAt: null, runnerStopRecorded: true }, + probe([], {}), { stopped: true }], + ] 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 d750a83a141..089d421450e 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -336,6 +336,118 @@ export async function evaluateNativeProviderProcesses(input: { }; } +/** Run event types that can carry a provider process identity. */ +export const NATIVE_PROVIDER_PROCESS_EVENT_TYPES = [ + "harness.ready", + "session.started", + "session.resumed", + "session.updated", + "turn.started", + "provider.event", + "provider.rpc_result", +] as const; + +function parsedProcessStart(value: unknown): Date | null { + const parsed = typeof value === "string" ? new Date(value) : null; + return parsed && !Number.isNaN(parsed.getTime()) ? parsed : null; +} + +/** Provider process identities from a saved checkpoint and recent run events. */ +export function nativeProviderProcessIdentities( + checkpoint: Record, + providerEvents: ReadonlyArray<{ payload: Record | null }>, +): NativeProviderProcessIdentity[] { + const identities: NativeProviderProcessIdentity[] = []; + const checkpointProcess = + checkpoint.process && typeof checkpoint.process === "object" && !Array.isArray(checkpoint.process) + ? (checkpoint.process as Record) + : {}; + for (const [field, startedAtField] of [ + ["providerPid", "providerProcessStartedAt"], + ["codexPid", "codexProcessStartedAt"], + ["sidecarPid", "sidecarProcessStartedAt"], + ["agentPid", "agentProcessStartedAt"], + ] as const) { + const value = checkpointProcess[field]; + if (typeof value === "number" && Number.isInteger(value) && value > 0) { + identities.push({ pid: value, processStartedAt: parsedProcessStart(checkpointProcess[startedAtField]) }); + } + } + for (const providerEvent of providerEvents) { + const prpEvent = providerEvent.payload?.prpEvent; + const event = prpEvent && typeof prpEvent === "object" && !Array.isArray(prpEvent) + ? (prpEvent as Record) : {}; + const eventPayload = event.payload && typeof event.payload === "object" && !Array.isArray(event.payload) + ? (event.payload as Record) : {}; + const processId = eventPayload.processId; + if (typeof processId === "number" && Number.isInteger(processId) && processId > 0) { + identities.push({ pid: processId, + processStartedAt: parsedProcessStart(eventPayload.processStartedAt ?? eventPayload.providerProcessStartedAt) }); + } + } + return identities; +} + +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[]; + /** A current, server-authored stop receipt already proved this exact Runner and group absent. */ + runnerStopRecorded?: boolean; + isProcessAlive?: (pid: number) => boolean; + isProcessGroupAlive?: (processGroupId: number) => boolean; + readProcessStartedAt?: (pid: number) => Promise; +}): Promise { + const refuse = (reason: Extract["reason"]) => + ({ stopped: false as const, reason }); + // A local Runner leads its own process group. Without both identities, a + // live child in that group could outlive the Runner unseen. + if (!input.processPid || !Number.isInteger(input.processPid) || input.processPid <= 1 || !input.processGroupId + || (!input.processStartedAt && !input.runnerStopRecorded)) { + return refuse("runner_process_identity_missing"); + } + if (!input.runnerStopRecorded) { + 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 (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; @@ -898,79 +1010,12 @@ export async function claimNativeRestartRecoveries(input: { .where( and( eq(heartbeatRunEvents.runId, row.run.id), - inArray(heartbeatRunEvents.eventType, [ - "harness.ready", - "session.started", - "session.resumed", - "session.updated", - "turn.started", - "provider.event", - "provider.rpc_result", - ]), + inArray(heartbeatRunEvents.eventType, [...NATIVE_PROVIDER_PROCESS_EVENT_TYPES]), ), ) .orderBy(desc(heartbeatRunEvents.id)) .limit(100); - const checkpointProcess = - checkpointRecord.process && - typeof checkpointRecord.process === "object" && - !Array.isArray(checkpointRecord.process) - ? (checkpointRecord.process as Record) - : {}; - const providerProcessIdentities: NativeProviderProcessIdentity[] = []; - const checkpointProcessFields = [ - ["providerPid", "providerProcessStartedAt"], - ["codexPid", "codexProcessStartedAt"], - ["sidecarPid", "sidecarProcessStartedAt"], - ["agentPid", "agentProcessStartedAt"], - ] as const; - for (const [field, startedAtField] of checkpointProcessFields) { - const value = checkpointProcess[field]; - if (typeof value === "number" && Number.isInteger(value) && value > 0) { - const rawStartedAt = checkpointProcess[startedAtField]; - const parsedStartedAt = - typeof rawStartedAt === "string" ? new Date(rawStartedAt) : null; - providerProcessIdentities.push({ - pid: value, - processStartedAt: - parsedStartedAt && !Number.isNaN(parsedStartedAt.getTime()) - ? parsedStartedAt - : null, - }); - } - } - for (const providerEvent of providerEvents) { - const prpEvent = providerEvent.payload?.prpEvent; - const event = - prpEvent && typeof prpEvent === "object" && !Array.isArray(prpEvent) - ? (prpEvent as Record) - : {}; - const eventPayload = - event.payload && - typeof event.payload === "object" && - !Array.isArray(event.payload) - ? (event.payload as Record) - : {}; - const processId = eventPayload.processId; - if ( - typeof processId === "number" && - Number.isInteger(processId) && - processId > 0 - ) { - const rawStartedAt = - eventPayload.processStartedAt ?? - eventPayload.providerProcessStartedAt; - const parsedStartedAt = - typeof rawStartedAt === "string" ? new Date(rawStartedAt) : null; - providerProcessIdentities.push({ - pid: processId, - processStartedAt: - parsedStartedAt && !Number.isNaN(parsedStartedAt.getTime()) - ? parsedStartedAt - : null, - }); - } - } + const providerProcessIdentities = nativeProviderProcessIdentities(checkpointRecord, providerEvents); const providerProcesses = await evaluateNativeProviderProcesses({ identities: (remoteProcess ? [] : providerProcessIdentities).filter( (identity) => identity.pid !== row.run.processPid, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index fd1216fe57c..98fee67a06c 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -292,6 +292,10 @@ import { import { currentNativeControllerIdentity, evaluateNativeControllerTakeover, + proveNativeLocalRunnerStopped, + nativeProviderProcessIdentities, + NATIVE_PROVIDER_PROCESS_EVENT_TYPES, + type NativeLocalRunnerStopProof, isCurrentNativeControllerProcess, nextNativeProviderAttempt, type NativeControllerIdentity, @@ -5286,15 +5290,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. @@ -5306,6 +5320,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)); @@ -5315,11 +5334,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 => @@ -5363,6 +5382,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; @@ -5377,11 +5402,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"]) { @@ -5389,7 +5417,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; @@ -5421,6 +5449,29 @@ 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 (): Promise => { + const run = prior!.run; + // Settlement may already have cleared the process fields after it saved a + // server-authored stop receipt; that receipt is the identity to check. + const released = run.processPid === null && run.processGroupId === null && run.processStartedAt === null; + const receipt = released ? await readNativeLocalProcessStop(input.db, run.companyId, run.id) : null; + if (released && !receipt) return { stopped: false, reason: "runner_process_identity_missing" }; + const providerEvents = await input.db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, [...NATIVE_PROVIDER_PROCESS_EVENT_TYPES]))) + .orderBy(desc(heartbeatRunEvents.id)).limit(100); + return proveNativeLocalRunnerStopped({ + processPid: receipt?.processPid ?? run.processPid, + processGroupId: receipt?.processGroupId ?? run.processGroupId, + processStartedAt: receipt ? null : run.processStartedAt, + runnerStopRecorded: receipt !== null, + providerProcesses: nativeProviderProcessIdentities(record(profile.sessionCheckpoint), providerEvents), + }); + }; + 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. @@ -5429,13 +5480,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 });