fix(native): restart a remote run whose controller never sent its runner a command (#20153)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A follow-up run on a remote sandbox inherits the previous run's
session checkpoint at admission, rebound to the new run's identity
> - A remote runner receives work only through the controller's Postgres
checkpoint, which the controller writes before every command. A run with
no checkpoint rows, no runner events and no recorded runner process
never sent its runner anything
> - Restart recovery still read the rebound admission checkpoint's
provider session id as provider evidence, and classified such a run
`reattach_remote_runner`. The executor then found the root still owned
by the previous run and failed with
`native_remote_recovery_lease_mismatch: stage=runner_verification
field=identity_run_id`
> - This pull request classifies that run as an incomplete bootstrap,
and lets the executor start it from the previous owner's settled root
> - The benefit is that a follow-up interrupted before its runner ever
got a command restarts on its own

## Linked Issues or Issue Description

Refs #20137

**What happened?**
Daytona per-turn restart-chaos M1 rerun w-20261109T013733-s9501 (master
8e3b64a257, image 34bc12bf), t02 (codex M1 parent):
1. Wake run `a867cb18` was admitted at 01:48:31 with a rebound
checkpoint inherited from run `dfeddafa`: identity of the new run,
`cursor: null`, no terminal turns, no active turn.
2. It recorded `native.process_start_requested` at 01:48:36, then a K4
hold froze the control plane.
3. At 01:53:11 recovery classified it `reattach_remote_runner`
(`hasCheckpoint: true`, `hasProviderEvidence: true`, from the rebound
checkpoint), although the run had no `native_runner_checkpoints` rows,
no runner-originated events and no recorded process identity.
4. The executor found the session root still owned by `dfeddafa` and
failed with `native_remote_recovery_lease_mismatch:
stage=runner_verification field=identity_run_id`. The task needed a
Retry.

**Expected behavior**
A remote run whose runner provably never received a command restarts its
bootstrap. The executor treats the previous owner's root as it would for
any follow-up start.

**Steps to reproduce**
1. Claim a remote sandbox follow-up that recorded
`native.process_start_requested` but has no controller checkpoint rows,
no runner events and no `native.process_identity_recorded`, with only
the admission-rebound session checkpoint.
2. Restart. Before this change, recovery claims
`reattach_remote_runner`, and the executor fails with `identity_run_id`.

**Paperclip version or commit**
8e3b64a257 (wave). Branch rebased on current master.

## What Changed

- `native-restart-recovery.ts`: the new
`remoteRunnerNeverReceivedCommands` condition.
- It applies to a remote sandbox run (not a plugin task) with a prepared
remote workspace when all of these hold:
    - no durable controller record, and no staged attachment;
- only the admission-rebound checkpoint (or none), and no provider
process events;
- no runner-originated events, and no
`native.process_identity_recorded`;
    - no run process fields;
- the same claimed and dead previous-controller conditions as
`remoteRunnerNeverLaunched`.
- It yields `bootstrap_incomplete` with the reason
`remote_runner_never_received_commands`, and does not count the rebound
checkpoint as this run's checkpoint.
- `native-session-executor.ts`:
- a `bootstrap_incomplete` claim requires the incomplete-bootstrap proof
only when the root carries this run's own identity;
- a root still owned by the previous run takes the normal previous-owner
path, and may retire a finished previous owner, as any follow-up start
does.
- Policy change, called out for review: a claimed remote run that
recorded only `native.process_start_requested` was held as
`ambiguous_remote_recovery_evidence`. Under the conditions above it now
bootstraps again.
- The controller writes every command to the Postgres checkpoint before
sending it, so with no checkpoint rows no command reached any runner,
and no provider work can exist.
- A runner process that was spawned but never commanded is stopped
before the new bootstrap by the existing stale-runner stop.
- A recorded process identity, an adapter invocation, a runner event,
process fields or a provider session still hold the run.
- Tests:
- `native-runner-restart-recovery.integration.test.ts`: the
classification for a follow-up with an inherited checkpoint and for a
first run without one, plus a held case when a runner process identity
was recorded. The old held expectation for "a runner process start was
requested" moved here;
- `native-recovered-owner-followup.test.ts`: the executor admitting a
`bootstrap_incomplete` follow-up from the previous owner's settled root.

## Verification

- Red/green:
- On master, the classification test claims `reattach_remote_runner`, as
in production. With the change it claims `bootstrap_incomplete` with
`remote_runner_never_received_commands`.
  - A recorded runner process identity keeps the run off this path.
  - The executor test is admitted with the change. On master it fails.
- `vitest run` over the restart-recovery, bootstrap, recovered-owner and
failed-workspace-recovery test files and
`native-session-executor.test.ts`: 9 files, all passing after the policy
test moved; `native-runner-restart-recovery.integration.test.ts` alone:
144 passed (14 skipped).
- `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm typecheck` (`server`): passed.

## Risks

- Low risk:
- The new path requires positive proof that nothing reached a runner: no
controller record, no runner events and no recorded process. Any of
these keeps the existing classification.
- In the executor, a root that carries this run's identity still needs
the full bootstrap proof.
- A root that carries the previous run's identity goes through the same
previous-owner verification as any follow-up start.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use
(shell, file edits, test execution).

## 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
- [ ] All Paperclip CI gates are green
- [ ] 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>

server/src/services/native-runtime/native-recovered-owner-followup.test.ts

@@ -202,8 +202,10 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered
     const remoteExecute = vi.fn(async (_request: unknown): Promise<{ exitCode: number; timedOut: boolean; stdout: string; stderr: string }> => {
       throw new Error("admission must not reach the sandbox");
     });
-    const start = (outcome: "created" | "replacement" | "resumed" = "created") => createRunnerdBackend({
+    const start = (outcome: "created" | "replacement" | "resumed" = "created",
+      restartRecovery?: Parameters<typeof createRunnerdBackend>[0]["restartRecovery"]) => createRunnerdBackend({
       db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); },
+      ...(restartRecovery ? { restartRecovery } : {}),
       runnerExecutionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", environmentId,
         leaseId, remoteCwd: "/workspace", reusableLeaseConfigured: true,
         runner: { execute: remoteExecute },
@@ -658,6 +660,16 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered
     expect(await followUp.start()).toBe(outcome);
   }, 60_000);
 
+  it("starts a follow-up whose own bootstrap never reached its root from the previous owner's settled root", async () => {
+    const prior = await seedRecoveredPriorRun();
+    const followUp = await admitFollowUp(prior);
+    const root = await restoreHeldRoot(prior, followUp);
+    expect(await followUp.start("created", { kind: "bootstrap_incomplete", runId: followUp.runId, leaseOwner: "successor",
+      controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null } as never)).toBe(ADMITTED);
+    expect(await quarantined()).toEqual([]);
+    expect(await archivedOwners(root)).toEqual([prior.runId]);
+  }, 60_000);
+
   it("admits again when a retry rebuilds the same projection from the database", async () => {
     const prior = await seedRecoveredPriorRun();
     const followUp = await admitFollowUp(prior);

server/src/services/native-runtime/native-restart-recovery.ts

@@ -678,7 +678,7 @@ export function classifyNativeRunnerRecoveryEvidence(input: {
   /** Durable evidence proves a claimed remote bootstrap reached no provider work. */
   remoteBootstrapProvenIncomplete?: boolean;
   remoteBootstrapIncompleteReason?: "remote_runner_never_launched" | "remote_session_opened_without_turn"
-    | "remote_runner_exited_before_provider_work";
+    | "remote_runner_exited_before_provider_work" | "remote_runner_never_received_commands";
   processStartMatches: boolean;
   knownProviderProcessAlive?: boolean;
   knownProviderProcessIdentityAmbiguous?: boolean;
@@ -793,6 +793,16 @@ async function hasNativeRunnerOriginatedEvidence(
   return event !== undefined;
 }
 
+async function hasNativeRunEvent(
+  tx: Pick<Db, "select">, run: { companyId: string; id: string }, eventType: string,
+): Promise<boolean> {
+  const [event] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents)
+    .where(and(eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
+      eq(heartbeatRunEvents.eventType, eventType)))
+    .limit(1);
+  return event !== undefined;
+}
+
 export async function hasNativeExecutionStartEvidence(
   tx: Pick<Db, "select">, run: { companyId: string; id: string },
 ): Promise<boolean> {
@@ -1408,6 +1418,24 @@ export async function claimNativeRestartRecoveries(input: {
       const remoteBootstrapProvenIncomplete = remoteBootstrapProof !== null
         && (!hasProviderEvidence || remoteBootstrapProof === "session_opened_without_turn");
 
+      const admissionCheckpointOnly = !hasCheckpoint || (checkpointIdentityMatches && checkpointRecord.cursor == null
+        && (!Array.isArray(checkpointRecord.terminalTurns) || checkpointRecord.terminalTurns.length === 0)
+        && checkpointRecord.activeTurnId == null && checkpointRecord.semanticResult == null && checkpointRecord.terminal == null);
+      const remoteRunnerNeverReceivedCommands = !unclaimedBootstrap && !remoteRunnerNeverLaunched
+        && !remoteRunnerExitedBeforeProviderWork && !remoteBootstrapProvenIncomplete
+        && remoteSandbox && remoteWorkspace !== null && !pluginTask && !durableAuthority && stagedAttachment === null
+        && admissionCheckpointOnly && providerEvents.length === 0
+        && row.run.runnerProfileJson?.nativePreparationFencingVersion === NATIVE_PREPARATION_FENCING_VERSION
+        && row.coordinator.phase === "observed" && Boolean(row.coordinator.leaseOwner)
+        && row.coordinator.attempt >= 1 && row.coordinator.controllerGeneration >= 1
+        && Boolean(row.coordinator.controllerBootId && row.coordinator.controllerPid && row.coordinator.controllerProcessStartedAt)
+        && !row.coordinator.failureDetail?.successorRunId
+        && !row.run.processPid && !row.run.processGroupId && !row.run.processStartedAt
+        && !row.run.sessionIdBefore && !row.run.sessionIdAfter
+        && Boolean(row.run.nativeSessionId && row.run.runnerInstanceId)
+        && !(await hasNativeRunnerOriginatedEvidence(tx, row.run))
+        && !(await hasNativeRunEvent(tx, row.run, PROCESS_IDENTITY_RECORDED));
+
       // Read before this claim appends its own history entry.
       const cleanDrainSuspension = suspendedByCleanDrain(row.coordinator);
       const resumeKeepsAttempt = cleanDrainSuspension || await restartResumeKeepsAttempt(tx, {
@@ -1417,9 +1445,10 @@ export async function claimNativeRestartRecoveries(input: {
         remoteSandbox: remoteSandbox || Boolean(pluginTask && durableAuthority),
         unclaimedBootstrap,
         remoteBootstrapProvenIncomplete: remoteBootstrapProvenIncomplete || remoteRunnerNeverLaunched
-          || remoteRunnerExitedBeforeProviderWork,
+          || remoteRunnerExitedBeforeProviderWork || remoteRunnerNeverReceivedCommands,
         remoteBootstrapIncompleteReason: remoteRunnerNeverLaunched ? "remote_runner_never_launched"
           : remoteRunnerExitedBeforeProviderWork ? "remote_runner_exited_before_provider_work"
+          : remoteRunnerNeverReceivedCommands ? "remote_runner_never_received_commands"
           : remoteBootstrapProof === "session_opened_without_turn" ? "remote_session_opened_without_turn" : undefined,
         runnerPidAlive,
         runnerGroupAlive,
@@ -1427,7 +1456,7 @@ export async function claimNativeRestartRecoveries(input: {
         knownProviderProcessAlive: providerProcesses.livePids.length > 0,
         knownProviderProcessIdentityAmbiguous:
           providerProcesses.ambiguousLivePids.length > 0,
-        hasCheckpoint,
+        hasCheckpoint: hasCheckpoint && !remoteRunnerNeverReceivedCommands,
         stagedAttachmentVerified: stagedAttachment !== null,
         checkpointIdentityMatches:
           checkpointIdentityMatches &&

server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts

@@ -2690,6 +2690,34 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
     return fixture;
   }
 
+  it.each([
+    { label: "restarts a remote follow-up", identityRecorded: false, inherited: true },
+    { label: "restarts a first remote run", identityRecorded: false, inherited: false },
+    { label: "keeps a recorded runner process held instead of restarting", identityRecorded: true, inherited: true },
+  ])("$label whose controller never sent its runner a command", async ({ identityRecorded, inherited }) => {
+    const fixture = await seedUnlaunchedRemoteClaim(`NEVER-COMMANDED-${identityRecorded}-${inherited}`,
+      { event: { eventType: "native.process_start_requested" } });
+    const [seeded] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
+    if (inherited) await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...seeded.runnerProfileJson, sessionCheckpoint: {
+      driverKind: "codex_app_server", backendKind: "runner", sessionId: "inherited-thread", cursor: null,
+      identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId },
+      providerSessionId: "inherited-thread", activeTurnId: null, terminal: null, terminalTurns: [], semanticResult: null,
+      pendingRuntimeRequests: [], lineage: [],
+    } } }).where(eq(heartbeatRuns.id, fixture.runId));
+    if (identityRecorded) await fixture.db.insert(heartbeatRunEvents).values({ companyId, runId: fixture.runId, agentId, seq: 901,
+      stream: "system", level: "info", eventType: "native.process_identity_recorded" });
+    const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
+      restartKind: "hard", runIds: [fixture.runId] });
+    if (identityRecorded) {
+      expect(claim).not.toMatchObject({ kind: "bootstrap_incomplete" });
+      return;
+    }
+    expect(claim).toMatchObject({ kind: "bootstrap_incomplete", runId: fixture.runId, controllerGeneration: 2 });
+    const [coordinator] = await fixture.db.select().from(nativeRunFinalizations)
+      .where(eq(nativeRunFinalizations.runId, fixture.runId));
+    expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ reason: "remote_runner_never_received_commands" });
+  });
+
   it("restarts a remote bootstrap retry that shutdown cut before any runner was requested", async () => {
     const fixture = await seedUnlaunchedRemoteClaim("BOOTSTRAP-RETRY", { bootstrapRetry: true });
     expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
@@ -2865,8 +2893,6 @@ process.on("SIGTERM", () => child.kill("SIGTERM"));
   });
 
   it.each<[string, Parameters<typeof seedUnlaunchedRemoteClaim>[1], { kind: string; reason: string }]>([
-    ["a runner process start was requested", { event: { eventType: "native.process_start_requested" } },
-      { kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }],
     ["a runner process identity was recorded", { event: { eventType: "native.process_identity_recorded" } },
       { kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }],
     ["the adapter was invoked", { event: { eventType: "adapter.invoke" } },

server/src/services/native-runtime/native-session-executor.ts

@@ -5808,7 +5808,7 @@ async function migrateRunnerdStateRootForExecution(input: {
       if (result.kind !== "state") throw new Error("Plugin Runner checkpoint unavailable");
       savePluginRunnerCheckpoint(scoped, result.state, identity);
     }
-    if (input.restartRecovery?.kind === "bootstrap_incomplete") {
+    if (input.restartRecovery?.kind === "bootstrap_incomplete" && identity.runId === input.execution.binding.runId) {
       // A local runner's process ids belong to this host, so a fresh provider
       // session it opened before any turn can be proven gone here.
       const localProcesses = !input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local";
@@ -5883,7 +5883,8 @@ async function migrateRunnerdStateRootForExecution(input: {
       // Name the failed retirement proof: the shared error otherwise hides
       // which of several independent fences held the follow-up.
       let retirementRefusal = "";
-      if (verification === "terminal_state_indeterminate" && !input.restartRecovery && !input.retainedWarmOwnerRunId) {
+      if (verification === "terminal_state_indeterminate" && !input.retainedWarmOwnerRunId
+        && (!input.restartRecovery || input.restartRecovery.kind === "bootstrap_incomplete")) {
         const retirement = await retireFinishedPerTurnRunnerdOwner({ ...input, root: scoped, identity });
         if (retirement.retired) return;
         if (retirement.refused !== "lifecycle_not_per_turn") retirementRefusal = `; retirement_refused=${retirement.refused}`;