fix(native): roll back a warm rotation cut off before its new control plane once its runner is proven dead (#20218)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A warm follow-up rotates authority before it attaches. It archives
the previous epoch (suspended runner state and settled controller) under
`authority-epochs/`, then writes its own control plane
> - #20146 taught recovery to roll back a rotation whose new control
plane held only an undelivered `run.attach`, once its runner was proven
dead
> - A hard kill can land one step earlier, after the archive and before
the new control plane exists. The root then holds only the archive, with
no active control plane and no runner state. The executor read "no
durable identity" and failed with `runner_state_identity_mismatch:
durable_identity_unreadable`
> - This pull request applies the same proven rollback to that earlier
point of the rotation
> - The benefit is that a hard kill anywhere in the rotation window
resumes the follow-up from the previous turn

## Linked Issues or Issue Description

Refs #20146

**What happened?**
Local mixed restart-chaos wave w-20261109T081456-s9901 (master
09e24dd12c), t15 (codex M1):
- Run `4b85497f` started at 08:29:32 as a warm follow-up and requested
its process, then K1 killed the server.
- Recovery stopped the previous run's warm runner (pid 60249) and
claimed `resume_dead_runner`.
- The executor failed with `runner_state_identity_mismatch:
durable_identity_unreadable`.

The quarantined root held:
- `authority-epochs/epoch-c2842347…` with the previous owner `dc7d447b`:
runner `suspended`, every command completed;
- no `control-plane/` state and no `runner/runner-state.json`, only
provider state.

A Retry ran fresh. Phase 3 also saw the same signature on Daytona;
#20214 covers that remote shape.

**Expected behavior**
For a local runner proven dead (`resume_dead_runner`), a root that holds
only the archive of a suspended previous owner of this session restores
that epoch, and the follow-up continues from the previous turn. Without
a dead-runner claim it still fails closed.

**Steps to reproduce**
1. Start a warm follow-up and hard-kill the server after the rotation
archives the previous epoch, before it writes the new control plane.
2. Restart. Before this change, the follow-up fails with
`durable_identity_unreadable`.

**Paperclip version or commit**
09e24dd12c (wave). Branch rebased on current master.

## What Changed

- `native-session-executor.ts`:
- `undeliveredAttachRotationSource` becomes `interruptedRotationSource`,
and the new `latestSuspendedArchivedEpoch` shares the epoch checks:
named by its identity digest, a matching suspended runner state.
- Besides #20146's undelivered-attach shape, it accepts a root with no
`runner-state.json` and no active control plane (absent or empty). In
that case the epoch must belong to this session (`normalizedSessionId`)
and to a different run.
- The rollback moves an active control plane aside only when one exists.
- `native-recovered-owner-followup.test.ts`: #20146's rotation test adds
the "cut off before its new control plane" shape. With a dead-runner
claim it is admitted from the restored previous owner; without the claim
it still fails closed.

## Verification

- Red/green: on master, the new admitted case fails with exactly
`durable_identity_unreadable`. With the change it is admitted. The
fail-closed cases are unchanged.
- `vitest run` of the recovered-owner, restart-recovery and bootstrap
test files and `native-session-executor.test.ts`: 8 files, 1627 tests
passed (14 skipped).
- `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm typecheck` (`server`): passed.

## Risks

- Low risk:
- The rollback still requires a `resume_dead_runner` claim on a local
runner, an absent runner state, and an archived epoch of this session
whose runner is suspended and whose name matches its identity.
- Restoring only moves files back within the root, and nothing is
deleted.

## 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

@@ -904,10 +904,13 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered
       }
     }, 60_000);
   it.each([
-    { label: "never delivered to a dead runner", connectionCount: 0, claim: true, admitted: true },
-    { label: "possibly delivered", connectionCount: 1, claim: true, admitted: false },
-    { label: "without a dead-runner claim", connectionCount: 0, claim: false, admitted: false },
-  ])("resumes from the previous turn when a warm attach rotation was $label", async ({ connectionCount, claim, admitted }) => {
+    { label: "never delivered to a dead runner", connectionCount: 0, claim: true, admitted: true, archiveOnly: false },
+    { label: "possibly delivered", connectionCount: 1, claim: true, admitted: false, archiveOnly: false },
+    { label: "without a dead-runner claim", connectionCount: 0, claim: false, admitted: false, archiveOnly: false },
+    { label: "cut off before its new control plane was written", connectionCount: 0, claim: true, admitted: true, archiveOnly: true },
+    { label: "cut off before its new control plane, without a dead-runner claim", connectionCount: 0, claim: false, admitted: false,
+      archiveOnly: true },
+  ])("resumes from the previous turn when a warm attach rotation was $label", async ({ connectionCount, claim, admitted, archiveOnly }) => {
     const prior = await seedRecoveredPriorRun();
     const followUp = await admitFollowUp(prior);
     const root = await restoreHeldRoot(prior, followUp);
@@ -921,22 +924,28 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered
     await mkdir(join(root, "runner"), { recursive: true, mode: 0o700 });
     await writeFile(join(root, "runner", "codex-provider-state.json"), "{}");
     const identity = { ...priorIdentity, runId: followUp.runId, turnId: `turn-${followUp.runId}`, itemId: `item-${followUp.runId}` };
-    await mkdir(join(root, "control-plane"), { mode: 0o700 });
+    if (!archiveOnly) await mkdir(join(root, "control-plane"), { mode: 0o700 });
     const undelivered = JSON.stringify({ schema: "paperclip.runner.durable.control-plane-state.v2", identity,
       bootstrapHistory: { schema: "paperclip.runner.bootstrap-history.v1", originIdentity: identity, credentialsIssued: 0, adopted: false },
       runAttachTemplate: {}, tickets: {}, leases: {},
       commands: [{ commandId: "attach", controllerSeq: 1, type: "run.attach", status: "pending", payload: {} }],
       committedEvents: [], ackedSourceSeq: 0, connectionCount, commandDeliveryCounts: {} });
-    await writeFile(join(root, "control-plane", "control-plane-state.json"), undelivered);
+    if (!archiveOnly) await writeFile(join(root, "control-plane", "control-plane-state.json"), undelivered);
     const outcome = await followUp.startLocal(claim ? { kind: "resume_dead_runner", runId: followUp.runId, leaseOwner: "successor",
       controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null } as never : undefined);
     if (!admitted) {
-      expect(outcome).toBe("runner_state_identity_mismatch: authority_indeterminate");
+      expect(outcome).toBe(archiveOnly ? "runner_state_identity_mismatch: durable_identity_unreadable"
+        : "runner_state_identity_mismatch: authority_indeterminate");
       return;
     }
     expect(outcome).toBe(ADMITTED);
     expect(JSON.parse(await readFile(join(root, "control-plane", "control-plane-state.json"), "utf8")).identity).toEqual(priorIdentity);
     expect(existsSync(join(root, "authority-epochs"))).toBe(false);
+    if (archiveOnly) {
+      expect(existsSync(join(root, "continuity-breaks"))).toBe(false);
+      expect(await quarantined()).toEqual([]);
+      return;
+    }
     const [archive] = await readdir(join(root, "continuity-breaks"));
     expect(await readFile(join(root, "continuity-breaks", archive!, "control-plane", "control-plane-state.json"), "utf8")).toBe(undelivered);
     expect(await quarantined()).toEqual([]);

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

@@ -5680,8 +5680,34 @@ async function retireFinishedPerTurnRunnerdOwner(input: Parameters<typeof inspec
   return { retired: true };
 }
 
-function undeliveredAttachRotationSource(root: string, execution: NativeExecutionInput): string | null {
+function latestSuspendedArchivedEpoch(root: string, matches: (prior: Record<string, unknown>) => boolean): string | null {
+  const archives = resolve(root, "authority-epochs");
+  if (!isSafeNativeStateDirectory(archives)) return null;
+  const candidates = readdirSync(archives).map(name => resolve(archives, name)).filter(epoch => {
+    const state = resolve(epoch, "control-plane", "control-plane-state.json");
+    return isSafeNativeStateDirectory(epoch) && existsSync(state) && existsSync(resolve(epoch, "runner-state.json"));
+  }).sort((left, right) => lstatSync(resolve(right, "control-plane", "control-plane-state.json")).mtimeMs
+    - lstatSync(resolve(left, "control-plane", "control-plane-state.json")).mtimeMs);
+  const latest = candidates[0];
+  if (!latest) return null;
+  const prior = record(readRunnerdControlPlaneState(latest).identity);
+  const runner = record(JSON.parse(readBoundedNativeFile(resolve(latest, "runner-state.json"),
+    NATIVE_RUNNER_STATE_MAX_BYTES, "runner_state_too_large").toString("utf8")));
+  const digest = createHash("sha256").update(JSON.stringify(prior)).digest("hex").slice(0, 24);
+  if (basename(latest) !== `epoch-${digest}` || !matches(prior) ||
+      runner.schema !== RUNNERD_STATE_SCHEMA || runner.lifecycle !== "suspended" ||
+      ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId"].some(key => runner[key] !== prior[key])) return null;
+  return latest;
+}
+
+function interruptedRotationSource(root: string, execution: NativeExecutionInput): string | null {
   try {
+    if (existsSync(resolve(root, "runner", "runner-state.json"))) return null;
+    if (!existsSync(resolve(root, "control-plane", "control-plane-state.json"))) {
+      if (existsSync(resolve(root, "control-plane")) && readdirSync(resolve(root, "control-plane")).length > 0) return null;
+      return latestSuspendedArchivedEpoch(root, prior => typeof prior.runId === "string" &&
+        prior.runId !== execution.binding.runId && prior.normalizedSessionId === nativeSessionKey(execution));
+    }
     const control = readRunnerdControlPlaneState(root);
     const identity = record(control.identity);
     const commands = Array.isArray(control.commands) ? control.commands.map(record) : [];
@@ -5693,26 +5719,9 @@ function undeliveredAttachRotationSource(root: string, execution: NativeExecutio
         Object.keys(record(control.commandDeliveryCounts)).length !== 0 ||
         control.warmTransition != null || control.completedWarmTransition != null || control.stagedRunAttachment != null ||
         history.adopted === true || history.credentialsIssued !== 0 ||
-        commands.length !== 1 || commands[0]!.type !== "run.attach" || commands[0]!.status !== "pending" ||
-        existsSync(resolve(root, "runner", "runner-state.json"))) return null;
-    const archives = resolve(root, "authority-epochs");
-    if (!isSafeNativeStateDirectory(archives)) return null;
-    const candidates = readdirSync(archives).map(name => resolve(archives, name)).filter(epoch => {
-      const state = resolve(epoch, "control-plane", "control-plane-state.json");
-      return isSafeNativeStateDirectory(epoch) && existsSync(state) && existsSync(resolve(epoch, "runner-state.json"));
-    }).sort((left, right) => lstatSync(resolve(right, "control-plane", "control-plane-state.json")).mtimeMs
-      - lstatSync(resolve(left, "control-plane", "control-plane-state.json")).mtimeMs);
-    const latest = candidates[0];
-    if (!latest) return null;
-    const prior = record(readRunnerdControlPlaneState(latest).identity);
-    const runner = record(JSON.parse(readBoundedNativeFile(resolve(latest, "runner-state.json"),
-      NATIVE_RUNNER_STATE_MAX_BYTES, "runner_state_too_large").toString("utf8")));
-    const digest = createHash("sha256").update(JSON.stringify(prior)).digest("hex").slice(0, 24);
-    if (basename(latest) !== `epoch-${digest}` || prior.runId === identity.runId ||
-        ["runnerInstanceId", "environmentLeaseId", "normalizedSessionId"].some(key => prior[key] !== identity[key]) ||
-        runner.schema !== RUNNERD_STATE_SCHEMA || runner.lifecycle !== "suspended" ||
-        ["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId"].some(key => runner[key] !== prior[key])) return null;
-    return latest;
+        commands.length !== 1 || commands[0]!.type !== "run.attach" || commands[0]!.status !== "pending") return null;
+    return latestSuspendedArchivedEpoch(root, prior => prior.runId !== identity.runId &&
+      ["runnerInstanceId", "environmentLeaseId", "normalizedSessionId"].every(key => prior[key] === identity[key]));
   } catch {
     return null;
   }
@@ -5721,13 +5730,15 @@ function undeliveredAttachRotationSource(root: string, execution: NativeExecutio
 async function rollBackUndeliveredAttachRotation(root: string, execution: NativeExecutionInput,
   onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>): Promise<void> {
   if (!existsSync(root) || !isSafeNativeStateDirectory(root)) return;
-  const epoch = undeliveredAttachRotationSource(root, execution);
+  const epoch = interruptedRotationSource(root, execution);
   if (!epoch) return;
   const archivedAt = Date.now();
-  const breakRoot = resolve(root, "continuity-breaks", `${archivedAt}-${randomUUID()}`);
-  mkdirSync(breakRoot, { recursive: true, mode: 0o700 });
   noteNativeRunnerEvidence(archivedAt);
-  renameSync(resolve(root, "control-plane"), resolve(breakRoot, "control-plane"));
+  if (existsSync(resolve(root, "control-plane"))) {
+    const breakRoot = resolve(root, "continuity-breaks", `${archivedAt}-${randomUUID()}`);
+    mkdirSync(breakRoot, { recursive: true, mode: 0o700 });
+    renameSync(resolve(root, "control-plane"), resolve(breakRoot, "control-plane"));
+  }
   mkdirSync(resolve(root, "runner"), { recursive: true, mode: 0o700 });
   renameSync(resolve(epoch, "runner-state.json"), resolve(root, "runner", "runner-state.json"));
   renameSync(resolve(epoch, "control-plane"), resolve(root, "control-plane"));