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

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - The native Runner path runs each per-turn agent turn through a
runnerd process, and the next turn must prove that the previous Runner
is finished before it takes over the saved session root
> - Under load, a local turn can have its result accepted while its
`runner.suspend` command does not finish, so the Runner's own receipt
stays unsuspended
> - The retirement proof that handles the same case for sandbox Runners
refused every local Runner with `target_not_sandbox_runner`
> - So the next turn failed with `runner_state_identity_mismatch:
prior_owner_terminal_state_indeterminate` and its session root was
quarantined
> - This pull request extends the retirement to a local Runner when the
prior Runner process is proven stopped by exact PID and start time
> - The benefit is that a follow-up turn (for example, after a user
answers a question) continues instead of failing

## Linked Issues or Issue Description

Refs #19525 (keeps the accepted result in the same incident; this is the
next run).

**What happened?**

In the runner E2E cell
`local-session-integrity.runner-acpx-claude.local.structured-question-restart-resume`,
the first turn asked a question and its result was accepted. The Runner
then logged `runner did not prove durable suspension before checkpoint`
(`runner.suspend` pending, Runner lifecycle `ready`). After a server
restart and the answer, the next run failed with
`runner_state_identity_mismatch:
prior_owner_terminal_state_indeterminate;
retirement_refused=target_not_sandbox_runner`, and its root was
quarantined.

**Expected behavior**

When the prior local Runner is provably stopped and every other
retirement proof holds, the next run retires that owner as a continuity
break and continues. If the stop cannot be proven, the run keeps the
named rejection.

**Steps to reproduce**

1. Run a local per-turn native turn whose `runner.suspend` does not
finish after its result is accepted (host load average ~40).
2. Start the next turn for the same session.
3. See `retirement_refused=target_not_sandbox_runner`.

## What Changed

- `server/src/services/native-runtime/native-restart-recovery.ts`: add
`proveNativeLocalRunnerStopped`. It uses the existing restart-recovery
checks: the recorded Runner PID is dead or now belongs to a process with
a different start time; its recorded process group is not alive; each
recorded provider process is dead or recycled. A missing PID, start time
or process group, a live recorded process, or a start time that cannot
be compared is not proof. When settlement already cleared the run's
process fields, the saved `native.local_process_stopped` receipt is the
Runner proof. Extract `nativeProviderProcessIdentities` (checkpoint plus
run events) so restart recovery and this proof read provider PIDs the
same way.
- `server/src/services/native-runtime/native-session-executor.ts`
(`retireFinishedPerTurnRunnerdOwner`, `readSettledRunnerdProjection`): a
local target is now eligible. For a local Runner, the projection accepts
the Runner's own unsuspended receipt for the same identity and
unfinished Runner lifecycle commands (`runner.suspend`, `runner.drain`,
`runner.shutdown`) only. All other proofs are unchanged: committed
result, released lease, no recovery state, stopped controller, delivered
terminal, other commands complete, tool inputs answered, quarantine
copies. The process proof runs before retirement and again just before
the archive.
- `doc/execution-semantics.md`: describe the local case.
- Tests: seven integration cases in
`native-recovered-owner-followup.test.ts` and ten unit cases in
`native-restart-recovery.test.ts`.

## Verification

- Red before the fix: the four new integration cases failed with
`retirement_refused=target_not_sandbox_runner`, which is the incident
error.
- Green after the fix: `pnpm exec vitest run
src/services/native-runtime/native-recovered-owner-followup.test.ts`
35/35. The new cases prove that a stopped Runner is retired (root
archived under `continuity-breaks`, no quarantine), and that a live
recorded Runner (`runner_process_alive`), a missing identity
(`runner_process_identity_missing`), and a pending non-lifecycle command
(`journal_commands_unsettled(turn.start.pending)`) keep the rejection.
All existing sandbox cases still pass, including "keeps an unsuspended
Runner receipt decisive" for sandbox targets.
- `native-restart-recovery.test.ts` unit cases: recycled PID, live
process, unknown start time, live group, live provider, missing PID or
start time.
- `native-session-executor.test.ts` + `native-restart-recovery.test.ts`:
1190 pass. `native-runner-restart-recovery.integration.test.ts` +
`native-restart-recovery-isolation.test.ts` (shared provider-PID
extraction): 81 pass. `tsc --noEmit -p server/tsconfig.json`: pass.
- Live: `pnpm test:e2e:runner -- --max-parallel 1
--max-automatic-retries 0 --id
local-session-integrity.runner-acpx-claude.local.structured-question-restart-resume`,
3 runs at eec57dd5ed: 3/3 PASS. The host load average was ~15 (the
incident had ~40), and `runner.suspend` finished in all three runs, so
these runs show no regression but did not exercise the new retirement
path. The integration cases cover that path.

## Risks

- The new path lets a follow-up archive a local root whose Runner
receipt is unsuspended. This is safe only if the Runner is stopped, so
the proof uses exact PID and start-time identity and fails closed on any
uncertainty. `ps` start times have one-second precision; a sub-second
difference is treated as the same process (not proof).
- A zombie-only process group on macOS can report as alive. That keeps
the rejection, as before.
- No quarantine behavior changes. A refused retirement still quarantines
the root as before.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, through Claude Code
with tool use (shell, file edits, GitHub CLI). Extended reasoning on.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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`

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"

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<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" }],
+    ["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,

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<string, unknown>,
+  providerEvents: ReadonlyArray<{ payload: Record<string, unknown> | null }>,
+): NativeProviderProcessIdentity[] {
+  const identities: NativeProviderProcessIdentity[] = [];
+  const checkpointProcess =
+    checkpoint.process && typeof checkpoint.process === "object" && !Array.isArray(checkpoint.process)
+      ? (checkpoint.process as Record<string, unknown>)
+      : {};
+  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<string, unknown>) : {};
+    const eventPayload = event.payload && typeof event.payload === "object" && !Array.isArray(event.payload)
+      ? (event.payload as Record<string, unknown>) : {};
+    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<Date | null>;
+}): Promise<NativeLocalRunnerStopProof> {
+  const refuse = (reason: Extract<NativeLocalRunnerStopProof, { stopped: false }>["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, [
+            inArray(heartbeatRunEvents.eventType, [...NATIVE_PROVIDER_PROCESS_EVENT_TYPES]),
-              "harness.ready",
-              "session.started",
-              "session.resumed",
-              "session.updated",
-              "turn.started",
-              "provider.event",
-              "provider.rpc_result",
-            ]),
           ),
         )
         .orderBy(desc(heartbeatRunEvents.id))
         .limit(100);
-      const checkpointProcess =
+      const providerProcessIdentities = nativeProviderProcessIdentities(checkpointRecord, providerEvents);
-        checkpointRecord.process &&
-        typeof checkpointRecord.process === "object" &&
-        !Array.isArray(checkpointRecord.process)
-          ? (checkpointRecord.process as Record<string, unknown>)
-          : {};
-      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<string, unknown>)
-            : {};
-        const eventPayload =
-          event.payload &&
-          typeof event.payload === "object" &&
-          !Array.isArray(event.payload)
-            ? (event.payload as Record<string, unknown>)
-            : {};
-        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 providerProcesses = await evaluateNativeProviderProcesses({
         identities: (remoteProcess ? [] : providerProcessIdentities).filter(
           (identity) => identity.pid !== row.run.processPid,

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<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.
@@ -5306,6 +5320,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));
@@ -5315,11 +5334,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 =>
@@ -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) {
+  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"]) {
@@ -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<NativeLocalRunnerStopProof> => {
+    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 });