fix(native): retire a finished local Runner that stopped before suspending

A local per-turn turn can have its result accepted while runner.suspend does
not finish under load. The Runner's own receipt then stays unsuspended, and
the next run failed with prior_owner_terminal_state_indeterminate;
retirement_refused=target_not_sandbox_runner and quarantined the root.

Retirement now also applies to a local Runner when the prior run's recorded
Runner process (and any recorded provider process) is proven stopped by exact
PID and start time, the same checks restart recovery uses. Only the Runner's
own lifecycle commands may be unfinished; every other proof is unchanged. A
live or unverifiable process keeps the named rejection.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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`

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"

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<number, Date | null>, 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,

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<Date | null>;
+}): Promise<NativeLocalRunnerStopProof> {
+  const refuse = (reason: Extract<NativeLocalRunnerStopProof, { stopped: false }>["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;

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<string, string>): SettledRunnerdProjection {
+const LOCAL_RUNNER_LIFECYCLE_COMMANDS = new Set(["runner.suspend", "runner.drain", "runner.shutdown"]);
+
+function readSettledRunnerdProjection(root: string, identity: Record<string, string>,
+  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<string, str
       NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8");
     const control = record(JSON.parse(bytes));
     const commands = (Array.isArray(control.commands) ? control.commands : []).map(record);
+    // A stopped local Runner can no longer act on its own lifecycle commands.
+    const settledCommands = options.localRunner && lifecycle === "not_suspended" && Array.isArray(control.commands)
+      ? commands.filter(command => !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<string, str
     if (Object.entries(identity).some(([key, value]) => 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) {
+  const localRunner = !target || target.kind === "local";
-    return refuse("target_not_sandbox_runner");
+  if (!localRunner) {
-  }
+    if (target.kind !== "remote" || target.transport !== "sandbox" || !target.runner || target.runnerTask) {
-  if (!["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) {
+      return refuse("target_not_sandbox_runner");
-    return refuse(`sandbox_${retirementReasonName("sandboxOutcome", target.sandboxLeaseAcquisition?.outcome)}`);
+    }
+    if (!["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) {
+      return refuse(`sandbox_${retirementReasonName("sandboxOutcome", target.sandboxLeaseAcquisition?.outcome)}`);
+    }
   }
   const identity: Record<string, string> = {};
   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 });