fix(runner): admit follow-ups after recovered per-turn owners (#18846)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Native Runner sessions keep one conversation across task turns.
> - After a per-turn sandbox turn, the controller saves a Runner
suspension receipt or a failover backup.
> - A full sandbox disk can stop that save after the turn is complete.
> - Recovery then commits the result, but the controller has no proof
that the Runner suspended.
> - Each follow-up rebuilds the old controller journal from the database
and rejects it again.
> - This pull request lets admission retire that old authority when the
prior owner is provably finished.
> - The benefit is that the task continues with its saved workspace and
does not stay blocked.

## Linked Issues or Issue Description

Refs #18705 (checkpoint save on a full sandbox disk). Refs #18829,
#18718 and #18343 (related admission and controller recovery). Refs
#18811 (a related warm continuation wait).

**What happened?**

A per-turn native run in a remote sandbox finished its turn. The
post-turn `harness_state.backup.persist` and
`session.checkpoint.persist` steps failed because the sandbox disk was
full. The controller got no Runner suspension receipt and no failover
backup. Workspace recovery later saved the files. The run committed its
result and ended as `succeeded`. A board comment then started a
follow-up in a new sandbox. Admission failed after about two minutes
with `runner_state_identity_mismatch:
prior_owner_terminal_state_indeterminate`. Execution recovery then
blocked the task.

The cause is in durable-root admission.
`restorePerTurnRunnerdController` rebuilds the prior controller journal
from the database checkpoint. `verifyPriorRunnerdStateForSessionScope`
finds a terminal owner in the same scope with no suspended lifecycle and
no retained warm owner. It returns `terminal_state_indeterminate`.
Admission quarantines the root and fails. The database checkpoint does
not change, so every later admission rebuilds the same journal and fails
again.

A board Retry after that failed follow-up does not repeat the same
error. Runtime selection picks the failed follow-up as the resume
source, because it carries an inherited provider checkpoint. Controller
recovery then fails with
`native_controller_recovery_unsupported:metadata_format`. The task stays
blocked. This pull request prevents that failed follow-up. It does not
change source selection (see Risks).

**Expected behavior**

When the prior owner is provably finished, the follow-up must start. It
must keep the saved workspace. It must not start while the prior owner
can still be alive, while terminal delivery or provider cleanup is
unsettled, or while quarantine holds evidence from a different owner.

**Steps to reproduce**

1. Complete a per-turn remote sandbox run. Commit its result through
recovery and release its controller lease.
2. Keep its database controller checkpoint and recovery pointer. Do not
save a suspension receipt or a failover backup.
3. Admit a follow-up in a new sandbox with no local controller
projection.
4. Admission rebuilds the old journal, quarantines it, and fails with
the error above.
5. Admit the same run again. It rebuilds the journal and fails the same
way.

The new regression test runs these steps against PostgreSQL.

**Paperclip version or commit**

Reproduced on `e0276e0166`.

**Deployment mode**

Native Runner with a per-turn lifecycle, the built-in remote sandbox
transport, and the ACPX provider. The admission path is the same for
other per-turn sandbox providers.

## What Changed

- Add `retireFinishedPerTurnRunnerdOwner` to durable-root admission in
`native-session-executor.ts`. It runs only when the existing verifier
returns `terminal_state_indeterminate`, and only without a restart claim
or a retained warm owner.
- The function requires all of these. Otherwise the existing quarantine
and rejection apply.
- Per-turn lifecycle on the built-in remote sandbox transport, with a
`created` or `replacement` sandbox. A resumed sandbox, a warm session,
and a plugin task keep their existing paths.
- No local Runner lifecycle evidence for the old owner. The session root
contains only known authority entries.
- A journal for that owner with its terminal delivered and only idle
goal snapshots after it. All commands are completed. Each tool input has
a completed result. No warm transition or staged attachment exists.
- A terminal prior run in the same task. Its coordinator is `committed`
with a result, has no lease owner, and has no recovery state.
- A stopped prior controller, from `evaluateNativeControllerTakeover`
with `requireStoppedController`. A durable controller is fenced by its
released database lease, as in remote warm recovery.
- Each quarantined root for the scope passes the same journal and local
Runner evidence checks for the same owner. A retry rebuilds only the
database journal, so quarantined Runner evidence must stay decisive.
- When the proof holds, the function reads the journal again, moves the
old authority into `continuity-breaks/<time>-<uuid>` in the session
root, and logs one line. The follow-up then starts a new Runner
authority. The provider session follows the existing resume-or-replace
policy.
- Share the list of continuity archive entries with provider replacement
(`archiveContinuityState`). Provider replacement behavior does not
change.
- Add `native-recovered-owner-followup.test.ts`. It uses real
PostgreSQL, a real encrypted controller checkpoint, and the production
source selector and checkpoint rebinder.
- Document the rule in `doc/execution-semantics.md` and the test in
`doc/testing-native-remote-continuity.md`.

### Design choice

I considered two approaches:

1. Start a new Runner authority for the scope at admission (this pull
request).
2. Make the recovery path that committed the prior run also settle its
Runner and controller projection.

I chose the first approach for these reasons:

- The provider history stayed in the unreachable sandbox. A settled
projection cannot make it resumable. Admission would pass, but a
replacement sandbox would then fail on the missing backup, and a new
sandbox would only try to resume a session that is gone.
- A settled receipt written by recovery would claim a suspension that
the Runner never wrote. Other checks rely on that receipt as Runner
evidence.
- Several paths can commit a result. Only runs that commit after a
settle-at-commit fix would benefit. The admission fix also unblocks
tasks that were committed before it.
- Admission already owns this decision. It already has the scope claim,
the journal, and the database rows.

### Safety argument

- **No live prior owner.** The coordinator has no lease owner, and the
controller must be proven stopped. The follow-up runs in a different
sandbox than the old Runner. A resumed sandbox keeps the existing path,
which reads the old Runner's own lifecycle there.
- **No unsettled terminal delivery or effects.** The controller
acknowledged the terminal and completed every command, including the
suspend command. Every tool input has an answer, and the coordinator
committed the result. The unreadable Runner outbox can then hold only
post-terminal telemetry.
- **No unsettled cleanup.** Any local Runner receipt, including an
unsuspended one, keeps the hold. A coordinator recovery state keeps the
hold.
- **No ambiguous quarantine.** A quarantined root from a different owner
keeps the hold. A quarantined root with Runner lifecycle evidence,
unknown entries, or an unsettled journal also keeps the hold.
- **No lost evidence.** The move keeps the old authority in the session
root. Existing retention expires it like other continuity breaks. A
journal change during the database reads cancels the move.

## Verification

- `cd server && npx vitest run
src/services/native-runtime/native-recovered-owner-followup.test.ts`: 12
tests pass.
- Before the change, the four admission tests failed with the exact
error above. This includes a re-admission that rebuilt the journal from
the database and failed again.
- After the change, `created` and `replacement` sandboxes are admitted.
A re-admission is admitted. A task with an earlier quarantined copy of
the same owner is admitted.
- Eight controls keep the hold: uncommitted result, live controller
lease, unexpired controller lease, recovery state, pending command,
unanswered tool input, a different owner in quarantine, and an
unsuspended Runner receipt in quarantine after a retry rebuilds the
root.
- I disabled each new guard in turn. Each change made its matching
control fail.
- An exploratory run of the same fixture before the change answered the
Retry question above. The second follow-up selected the failed follow-up
as its source and failed with
`native_controller_recovery_unsupported:metadata_format`.
- `cd server && npx vitest run
src/services/native-runtime/native-session-executor.test.ts`: all 1,087
tests pass.
- Related suites pass: `task-workspace-binding`,
`native-controller-recovery` (unit and integration),
`native-harness-backup-expiration`, `native-session-resume`, and the
restart and diskless integration suites. Their container and Runner
binary cases skip locally.
- `node scripts/typecheck-low-memory.mjs`, `pnpm
check:module-boundaries`, and `git diff --check` pass.
- CI passes at `879bb6a5c7`. Two attempts lost jobs when CI runners
received a shutdown signal; the third attempt passed every job.
- I did not run the full `pnpm test:run`, `pnpm build`, or the opt-in
Linux Runner container cases locally. CI runs them.
- I did not test a live sandbox provider or a live model. I could not
inspect the original checkpoint contents. If a stranded run's journal
does not show a delivered terminal and completed commands, its follow-up
stays held.

## Risks

- A follow-up after this path cannot use the lost provider history. The
provider session then follows the existing resume-or-replace policy, and
the task context comes from Paperclip. A checkpoint that requires the
same session, such as an unfinished goal, still fails at provider
recovery.
- Post-terminal Runner telemetry in the lost outbox, such as a late
usage report, stays lost. It cannot carry a workspace effect.
- A task that already has a failed follow-up from this error can still
be blocked on a board Retry. Source selection picks that failed run and
fails with `native_controller_recovery_unsupported:metadata_format`.
This pull request does not change source selection. A same-run retry of
the failed follow-up is admitted.
- The change is limited to per-turn runs on the built-in sandbox
transport. Warm sessions, plugin tasks, resumed sandboxes, and restart
recovery do not change.
- No migration or configuration change is required.

## Model Used

- Anthropic Claude Opus 5.5 (model ID `claude-opus-5-5`) in Claude Code,
with extended thinking, repository tools, shell and code execution, and
local test runs. The runtime does not expose the context window size.

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

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

doc/execution-semantics.md

@@ -1610,6 +1610,28 @@ and plugin tasks. The running session then owns authority;
 the old receipt must not conflict with a later warm run's journal identity.
 Failed admission retains the receipt. Normal suspension writes a new receipt.
 
+A per-turn sandbox turn can finish through recovery without a Runner
+suspension receipt or failover backup on the controller. For example, its
+post-turn checkpoint copy can fail on a full sandbox disk. The controller
+journal from the database then cannot prove that the old Runner suspended.
+When the next run starts in a new or replacement sandbox, admission archives
+that old authority as a continuity break, and the run starts a new Runner.
+All of these must be true:
+
+- The prior run is terminal and its coordinator committed a result.
+- The coordinator has no lease owner and no recovery state.
+- The prior controller is proven stopped.
+- The journal has a delivered terminal, only completed commands, and an
+  answer for each tool input.
+- No local Runner lifecycle evidence exists for the old owner.
+- Each quarantined root for the scope passes the same journal and local
+  Runner evidence checks for the same owner.
+
+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`
+rejection.
+
 A repository-free fresh plugin task directory remains the task's directory on
 ordinary follow-ups. Reuse requires its exact company, task, workspace, provider
 owner, fresh-attempt identity and path. Mode-only isolation settings do not turn

doc/testing-native-remote-continuity.md

@@ -49,8 +49,11 @@ They do not qualify distinct application releases or complete webserver disk los
 
 `native-controller-recovery.integration.test.ts` separately verifies encrypted
 database discovery, one-source import, idempotency, live-owner waiting, corrupt
-bindings and paginated inventory. `native-session-resume-projection.test.ts`
-retains the 32-candidate and newer-progress barrier checks. Snapshot transaction
+bindings and paginated inventory. `native-recovered-owner-followup.test.ts`
+admits a per-turn follow-up after recovery committed an owner that never saved
+a suspension receipt, and keeps unsettled or ambiguous owners held.
+`native-session-resume-projection.test.ts` retains the 32-candidate and
+newer-progress barrier checks. Snapshot transaction
 counts and stale-generation rejection are exercised by the control-plane port
 tests; incomplete-publication retention is exercised by the expiration tests.
 

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

@@ -0,0 +1,302 @@
+import { randomUUID } from "node:crypto";
+import { existsSync } from "node:fs";
+import { cp, mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { eq } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import {
+  agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults,
+} from "@paperclipai/db";
+import { DurablePrpControlPlane, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
+import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/helpers/recovery-postgres.js";
+import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js";
+import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
+import { buildNativeExecutionInput } from "./native-execution-input.js";
+import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
+import { createNativeControllerRecovery } from "./native-controller-recovery.js";
+import {
+  buildNativeExecutionWithCheckpoint, findNativeSessionResumeRun, nativeToolContractFingerprintForTarget,
+} from "./native-session-resume.js";
+import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
+
+const toolDefinitions = vi.hoisted(() => vi.fn());
+vi.mock("../../runtime-api.js", () => ({ resolveLocalRuntimeApiUrl: () => "http://127.0.0.1:43219" }));
+vi.mock("./paperclip-runner-tool-authority.js", () => ({
+  PaperclipRunnerToolAuthority: class {
+    readonly binding = {};
+    definitions() { return toolDefinitions(); }
+    close() {}
+  },
+}));
+vi.mock("./current-wake-comments.js", () => ({
+  resolveCurrentWakeCommentsBinding: async () => null,
+  assertCurrentWakeCommentsRead: async () => undefined,
+}));
+import { createRunnerdBackend } from "./native-session-executor.js";
+
+// Durable-root admission is the boundary under test. Reaching the run's tool
+// authority proves admission passed; no provider or runner process starts.
+const ADMITTED = "follow-up admission reached tool authority";
+const STRANDED = "runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate";
+
+describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered per-turn native owner", () => {
+  let database: Awaited<ReturnType<typeof startRecoveryPostgres>>;
+  let db: ReturnType<typeof createDb>;
+  let home: ReturnType<typeof createIsolatedTestHome>;
+  let stateDirectory: string;
+  const previousHome = process.env.PAPERCLIP_HOME;
+  const previousStateDirectory = process.env.PAPERCLIP_RUNNER_STATE_DIR;
+  const companyId = randomUUID(), agentId = randomUUID();
+
+  beforeAll(async () => {
+    home = createIsolatedTestHome();
+    process.env.PAPERCLIP_HOME = home.home;
+    database = await startRecoveryPostgres("native-recovered-owner-");
+    db = createDb(database.connectionString);
+    await db.insert(companies).values({ id: companyId, name: "Recovered owner", issuePrefix: "ROF" });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Runner", adapterType: "paperclip_runner",
+      adapterConfig: { provider: "acpx", acpxAgent: "claude" } });
+  }, 60_000);
+  afterAll(async () => {
+    await database?.cleanup();
+    home?.cleanup();
+    if (previousHome === undefined) delete process.env.PAPERCLIP_HOME;
+    else process.env.PAPERCLIP_HOME = previousHome;
+  });
+  beforeEach(async () => {
+    stateDirectory = await mkdtemp(join(tmpdir(), "native-recovered-owner-state-"));
+    process.env.PAPERCLIP_RUNNER_STATE_DIR = stateDirectory;
+    toolDefinitions.mockReset().mockRejectedValue(new Error(ADMITTED));
+  });
+  afterEach(async () => {
+    if (previousStateDirectory === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR;
+    else process.env.PAPERCLIP_RUNNER_STATE_DIR = previousStateDirectory;
+    await rm(stateDirectory, { recursive: true, force: true });
+  });
+
+  const issueFields = (issueId: string) => ({ id: issueId, title: "Continue after recovery", description: null, reviewPolicy: null });
+
+  /** A per-turn sandbox turn whose post-turn checkpoint copy failed. Recovery
+   * later committed its result and released the controller lease, but no
+   * suspended Runner receipt or failover backup ever reached the controller.
+   * Only the database controller checkpoint and recovery pointer survive. */
+  async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" = "settled") {
+    const issueId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID();
+    await db.insert(issues).values({ id: issueId, companyId, title: "Continue after recovery", assigneeAgentId: agentId, status: "in_progress" });
+    const [row] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native",
+      nativeIssueId: issueId, contextSnapshot: { issueId } }).returning();
+    const native = await prepareNativeHeartbeatRun({ db, run: row, issue: issueFields(issueId), environmentLeaseId: randomUUID() });
+    const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId));
+    const executionFor = (currentRunId: string) => buildNativeExecutionInput({
+      companyId, agentId, runId: currentRunId, provider: "acpx", acpxAgent: "claude", model: "claude-opus-5-5",
+      issue: { id: issueId, identifier: "ROF-1", title: "Continue after recovery", description: null, workMode: "standard" },
+      taskPrompt: "Continue the conversation.",
+      workspace: { id: workspaceId, cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null },
+      normalizedSessionId: native.normalizedSessionId,
+      lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
+      runtimeContext: nativeRuntimeContextFixture(),
+      completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: contract.schemaVersion,
+        contract: contract.contractJson as never },
+    });
+    const execution = executionFor(runId);
+    const identity = { runId, normalizedSessionId: native.normalizedSessionId, runnerInstanceId: native.runnerInstanceId,
+      environmentLeaseId: native.environmentLeaseId, turnId: native.turnId, itemId: native.itemId };
+    const owner = { companyId, issueId, runId, authorityRunId: runId, bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() };
+    await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "observed",
+      controllerBootId: owner.bootId, controllerGeneration: 1, controllerPid: 2_147_483_000,
+      controllerProcessStartedAt: new Date(Date.now() - 60_000), leaseOwner: owner.leaseOwner,
+      leaseExpiresAt: new Date(Date.now() + 60_000) });
+    const scratch = await mkdtemp(join(tmpdir(), "native-recovered-owner-controller-"));
+    const controller = new DurablePrpControlPlane({ stateDirectory: scratch, identity,
+      expectedRunnerVersion: "recovered-owner-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
+    try {
+      // Shape the file-backed journal first; the database checkpoint captures
+      // it when persistence moves to the controller's external writer.
+      const commands = ["run.prepare", "session.open", "turn.start", "runner.suspend"].map(type => controller.queueCommand(type));
+      for (const command of control === "pending-command" ? commands.slice(0, -1) : commands) {
+        Object.assign(command, { status: "completed", result: { status: "completed" } });
+      }
+      const event = (sourceSeq: number, sourceEventId: string, eventType: string) => ({ sourceSeq, sourceEventId, eventType,
+        priority: 0 as const, envelope: { ...identity }, deliveryCount: 1, logicalEffectCount: 1 });
+      controller.store.state.committedEvents.push(
+        ...(control === "unanswered-tool" ? [event(1, "unanswered-tool-input", "semantic_tool.input")] : []),
+        event(2, "recovered-terminal", "run.terminal"));
+      controller.store.state.ackedSourceSeq = 2;
+      controller.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, owner));
+      await controller.flushCheckpoint();
+    } finally {
+      await controller.stop();
+      await rm(scratch, { recursive: true, force: true });
+    }
+    const snapshot: PersistedNativeSession = { backendKind: "runner", sessionId: native.normalizedSessionId,
+      providerSessionId: "provider-session-before-recovery", activeTurnId: null, pendingRuntimeRequests: [],
+      identity: { companyId, agentId, issueId, runId, sessionId: native.normalizedSessionId } };
+    const [prepared] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
+    const [result] = await db.insert(nativeRunResults).values({ companyId, issueId, runId,
+      completionContractId: prepared.completionContractId!, serverFingerprint: "fixture", schemaStatus: "accepted",
+      resultJson: {}, canonicalSha256: "fixture" }).returning();
+    await db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null })
+      .where(eq(nativeRunFinalizations.runId, runId));
+    await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), runnerProfileJson: {
+      ...prepared.runnerProfileJson, nativeExecutionInput: execution, sessionCheckpoint: snapshot,
+      nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("remote", execution.provider.kind),
+      nativeControllerRecovery: createNativeControllerRecovery(snapshot,
+        { configDigest: `sha256:${"b".repeat(64)}`, processConfigDigest: null },
+        { runId, runnerInstanceId: native.runnerInstanceId, normalizedSessionId: native.normalizedSessionId }),
+    } }).where(eq(heartbeatRuns.id, runId));
+    return { issueId, runId, native, identity, executionFor };
+  }
+
+  /** Mirrors runtime selection: the follow-up inherits the selected source's
+   * provider checkpoint before the executor admits its durable root. */
+  async function admitFollowUp(prior: Awaited<ReturnType<typeof seedRecoveredPriorRun>>) {
+    const runId = randomUUID();
+    const [row] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native",
+      nativeIssueId: prior.issueId, nativeSessionId: prior.native.normalizedSessionId, contextSnapshot: { issueId: prior.issueId } }).returning();
+    const next = await prepareNativeHeartbeatRun({ db, run: row, issue: issueFields(prior.issueId), environmentLeaseId: randomUUID() });
+    await db.insert(nativeRunFinalizations).values({ companyId, issueId: prior.issueId, runId, phase: "observed" });
+    await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, prior.issueId));
+    const execution = prior.executionFor(runId);
+    const selected = await findNativeSessionResumeRun(db, { companyId, agentId, issueId: prior.issueId,
+      normalizedSessionId: prior.native.normalizedSessionId, currentRunId: runId, beforeCreatedAt: row.createdAt });
+    expect(selected?.id).toBe(prior.runId);
+    const admission = buildNativeExecutionWithCheckpoint({ previousRun: selected, normalizedSessionId: prior.native.normalizedSessionId,
+      executionTargetKind: "remote", buildExecution: () => execution });
+    expect(admission.checkpoint).not.toBeNull();
+    const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
+    await db.update(heartbeatRuns).set({ runnerProfileJson: { ...current.runnerProfileJson, nativeExecutionInput: execution,
+      nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("remote", execution.provider.kind),
+      sessionCheckpoint: admission.checkpoint } }).where(eq(heartbeatRuns.id, runId));
+    const logs: string[] = [];
+    const start = (outcome: "created" | "replacement" = "created") => createRunnerdBackend({
+      db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); },
+      runnerExecutionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", environmentId: randomUUID(),
+        leaseId: randomUUID(), remoteCwd: "/workspace", reusableLeaseConfigured: true,
+        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 };
+  }
+
+  async function quarantined() {
+    const path = join(stateDirectory, "quarantine");
+    return existsSync(path) ? readdir(path) : [];
+  }
+
+  async function sessionRoot() {
+    const roots = (await readdir(stateDirectory)).filter(name => /^[a-f0-9]{64}$/.test(name));
+    expect(roots).toHaveLength(1);
+    return join(stateDirectory, roots[0]!);
+  }
+
+  async function archivedOwners(root: string) {
+    const parent = join(root, "continuity-breaks");
+    if (!existsSync(parent)) return [];
+    return Promise.all((await readdir(parent)).map(async name => {
+      expect(name).toMatch(/^\d+-[0-9a-f-]{36}$/);
+      return JSON.parse(await readFile(join(parent, name, "control-plane", "control-plane-state.json"), "utf8")).identity.runId as string;
+    }));
+  }
+
+  it.each(["created", "replacement"] as const)(
+    "admits a fresh runner authority when the recovered owner is provably finished: %s sandbox", async outcome => {
+      const prior = await seedRecoveredPriorRun();
+      const followUp = await admitFollowUp(prior);
+      expect(await followUp.start(outcome)).toBe(ADMITTED);
+      expect(await quarantined()).toEqual([]);
+      const root = await sessionRoot();
+      // The prior authority is retained as continuity evidence, not active state.
+      expect(existsSync(join(root, "control-plane"))).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("admits again when a retry rebuilds the same projection from the database", async () => {
+    const prior = await seedRecoveredPriorRun();
+    const followUp = await admitFollowUp(prior);
+    expect(await followUp.start()).toBe(ADMITTED);
+    // Every admission restores the prior projection from the database. The
+    // same proof must hold each time instead of rejecting the rebuilt root.
+    expect(await followUp.start()).toBe(ADMITTED);
+    expect(await quarantined()).toEqual([]);
+    expect(await archivedOwners(await sessionRoot())).toEqual([prior.runId, prior.runId]);
+  }, 60_000);
+
+  it("admits a stranded task whose earlier attempts quarantined the same projection", async () => {
+    const prior = await seedRecoveredPriorRun();
+    const followUp = await admitFollowUp(prior);
+    await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId));
+    expect(await followUp.start()).toBe(STRANDED);
+    expect(await quarantined()).toHaveLength(1);
+    await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId));
+    expect(await followUp.start()).toBe(ADMITTED);
+    expect(await quarantined()).toHaveLength(1);
+  }, 60_000);
+
+  it("keeps an unsuspended Runner receipt decisive after a retry rebuilds the root", async () => {
+    const prior = await seedRecoveredPriorRun();
+    const followUp = await admitFollowUp(prior);
+    // Learn the scope key from a held attempt, then give the active root the
+    // local evidence of an owner whose Runner never suspended.
+    await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId));
+    expect(await followUp.start()).toBe(STRANDED);
+    const [copy] = await quarantined();
+    const root = join(stateDirectory, copy!.split(".")[0]!);
+    await rename(join(stateDirectory, "quarantine", copy!), root);
+    await mkdir(join(root, "runner"));
+    await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({
+      schema: "paperclip.runner.durable.state.v1", lifecycle: "ready", ...prior.identity }));
+    await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId));
+    expect(await followUp.start()).toBe(STRANDED);
+    expect(await quarantined()).toHaveLength(1);
+    // The retry restores only the database journal. The quarantined receipt
+    // must still hold it.
+    expect(await followUp.start()).toBe(STRANDED);
+    expect(await quarantined()).toHaveLength(2);
+    expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint");
+  }, 60_000);
+
+  it.each(["uncommitted 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"
+      : scenario === "unanswered tool input" ? "unanswered-tool" : "settled");
+    const followUp = await admitFollowUp(prior);
+    let expectedQuarantine = 1;
+    if (scenario === "uncommitted result") {
+      await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", resultId: null })
+        .where(eq(nativeRunFinalizations.runId, prior.runId));
+    }
+    if (scenario === "live controller lease") {
+      await db.update(nativeRunFinalizations).set({ leaseOwner: "another-controller", leaseExpiresAt: new Date(Date.now() + 60_000) })
+        .where(eq(nativeRunFinalizations.runId, prior.runId));
+    }
+    if (scenario === "unexpired controller lease") {
+      await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() + 60_000) })
+        .where(eq(nativeRunFinalizations.runId, prior.runId));
+    }
+    if (scenario === "recovery pending") {
+      await db.update(nativeRunFinalizations).set({ recoveryState: "resuming_session" })
+        .where(eq(nativeRunFinalizations.runId, prior.runId));
+    }
+    if (scenario === "foreign quarantine") {
+      // A copy of this projection is harmless; a different owner saved under
+      // the same scope is unresolved evidence that must not be overtaken.
+      await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId));
+      expect(await followUp.start()).toBe(STRANDED);
+      const [copy] = await quarantined();
+      const foreign = join(stateDirectory, "quarantine", copy!.replace(/\.[0-9a-f-]{36}$/, `.${randomUUID()}`));
+      await cp(join(stateDirectory, "quarantine", copy!), foreign, { recursive: true });
+      const path = join(foreign, "control-plane", "control-plane-state.json");
+      const state = JSON.parse(await readFile(path, "utf8"));
+      await writeFile(path, JSON.stringify({ ...state, identity: { ...state.identity, runId: randomUUID() } }));
+      await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId));
+      expectedQuarantine = 3;
+    }
+    const outcome = await followUp.start();
+    // A live owner is still waited for before any projection is rebuilt.
+    expect(outcome).toBe(scenario === "live controller lease" ? "native_controller_recovery_waiting_for_owner" : STRANDED);
+    expect(await quarantined()).toHaveLength(scenario === "live controller lease" ? 0 : expectedQuarantine);
+    expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint");
+  }, 60_000);
+});

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

@@ -200,7 +200,7 @@ import {
   nativeRunResults,
 } from "@paperclipai/db";
 import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js";
-import { assertNativeRetainedSandboxBinding, deferNativeControllerRecovery, NativeControllerRecoveryWaitingError, resolveNativeControllerRecovery } from "./native-controller-recovery.js";
+import { assertNativeRetainedSandboxBinding, deferNativeControllerRecovery, NativeControllerRecoveryWaitingError, readNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js";
 import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js";
 import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
 import { appendNativeCredentialUpdate } from "./native-credential-updates.js";
@@ -226,7 +226,7 @@ import {
   RemoteRunnerRecoveryError, attachRemoteRunnerRecoveryDiagnostic, readRemoteRunnerRecoveryError, remoteRecoveryProbeReason,
 } from "./remote-runner-recovery-diagnostics.js";
 import { readRemoteRunnerRecoveryProbe, REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS } from "./remote-runner-recovery-probe.js";
-import { inspectRemoteWarmSessionRecovery, type RemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js";
+import { inspectRemoteWarmSessionRecovery, isSettledRemoteWarmControlState, type RemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js";
 import { resolveNativeStagedAttachmentRecovery, type NativeStagedAttachmentRecovery } from "./native-staged-attachment-recovery.js";
 import { emitStartupReuse, environmentStartupReuse } from "@paperclipai/adapter-utils/startup-reuse";
 import { persistActivity, publishActivity } from "../activity-log.js";
@@ -266,6 +266,7 @@ import {
 } from "./native-runner-ownership.js";
 import {
   currentNativeControllerIdentity,
+  evaluateNativeControllerTakeover,
   nextNativeProviderAttempt,
   type NativeControllerIdentity,
   type NativeRestartRecoveryClaim,
@@ -5102,6 +5103,140 @@ async function recoverPluginRunnerCheckpoint(input: {
   }
 }
 
+// Session-root entries owned by one provider session's authority. A continuity
+// break moves them aside together. Backups belong to the retired session too:
+// left active, they make a fresh replacement look like lost harness state.
+const NATIVE_CONTINUITY_STATE_ENTRIES: readonly string[] = [
+  "control-plane",
+  "runner",
+  "codex-home",
+  "opencode",
+  "acpx",
+  "failover-backups",
+  "provider-pack-authority.json",
+];
+
+function moveNativeContinuityState(root: string, archiveRoot: string): void {
+  for (const name of NATIVE_CONTINUITY_STATE_ENTRIES) {
+    const source = resolve(root, name);
+    if (existsSync(source)) renameSync(source, resolve(archiveRoot, name));
+  }
+}
+
+/**
+ * Returns the journal bytes when a root holds only a settled controller
+ * journal for this exact owner. The controller acknowledged its terminal,
+ * completed every command and answered every tool input, so the unreadable
+ * 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.
+ */
+function readSettledRunnerdProjection(root: string, identity: Record<string, string>): string | null {
+  try {
+    if (!isSafeNativeStateDirectory(root) ||
+        runnerdAuthorityLifecycle(root, identity as RunnerdDurableIdentity) !== "absent" ||
+        readdirSync(root).some(name => name !== "continuity-breaks" && !NATIVE_CONTINUITY_STATE_ENTRIES.includes(name))) {
+      return null;
+    }
+    const bytes = readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"),
+      NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8");
+    const control = record(JSON.parse(bytes));
+    const answered = new Set((Array.isArray(control.commands) ? control.commands : []).map(record)
+      .filter(command => command.type === "semantic_tool.result" && command.status === "completed")
+      .map(command => record(command.payload).sourceEventId));
+    if (control.schema !== RUNNERD_CONTROL_PLANE_STATE_SCHEMA || control.warmTransition != null ||
+        control.stagedRunAttachment != null || !isSettledRemoteWarmControlState(control, identity) ||
+        (control.committedEvents as unknown[]).map(record).some(event =>
+          (event.eventType === "semantic_tool.input" || event.eventType === "mcp_app.tool_input") &&
+          !answered.has(event.sourceEventId))) return null;
+    return bytes;
+  } catch {
+    return null;
+  }
+}
+
+/**
+ * A per-turn sandbox turn can finish through recovery after its post-turn
+ * checkpoint copy failed: the result is committed and the controller lease is
+ * released, but no suspended Runner receipt or failover backup ever reached
+ * this root. Its database projection can never prove that suspension, so
+ * quarantine only lets the next run rebuild and reject it again. When the
+ * owner is provably finished and this run uses a different sandbox, archive
+ * 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.
+ */
+async function retireFinishedPerTurnRunnerdOwner(input: {
+  db: Db;
+  root: string;
+  identity: RunnerdDurableIdentity;
+  execution: NativeExecutionInput;
+  runnerExecutionTarget?: AdapterExecutionTarget | null;
+  onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
+}): Promise<boolean> {
+  const target = input.runnerExecutionTarget;
+  // A resumed sandbox still holds the prior Runner's own lifecycle, which its
+  // existing recovery path reads. Plugin tasks own a separate saved checkpoint.
+  if (input.execution.session.lifecyclePolicy.mode !== "per_turn" || target?.kind !== "remote" ||
+      target.transport !== "sandbox" || !target.runner || target.runnerTask ||
+      !["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) return false;
+  const identity: Record<string, string> = {};
+  for (const key of ["runId", "normalizedSessionId", "runnerInstanceId", "environmentLeaseId", "turnId", "itemId"]) {
+    const value = input.identity[key];
+    if (typeof value !== "string" || !value) return false;
+    identity[key] = value;
+  }
+  const controlBytes = readSettledRunnerdProjection(input.root, identity);
+  if (controlBytes === null) return false;
+  const { binding } = input.execution;
+  let prior: { run: typeof heartbeatRuns.$inferSelect; owner: typeof nativeRunFinalizations.$inferSelect } | undefined;
+  try {
+    [prior] = await input.db.select({ run: heartbeatRuns, owner: nativeRunFinalizations }).from(heartbeatRuns)
+      .innerJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id),
+        eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId)))
+      .where(and(eq(heartbeatRuns.id, identity.runId), eq(heartbeatRuns.companyId, binding.companyId),
+        eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId),
+        eq(nativeRunFinalizations.issueId, binding.issueId)))
+      .limit(1);
+  } catch {
+    return false;
+  }
+  if (!prior || prior.run.runtimeMode !== "native" || !TERMINAL_HEARTBEAT_RUN_STATUSES.has(prior.run.status) ||
+      prior.owner.phase !== "committed" || !prior.owner.resultId || prior.owner.leaseOwner || prior.owner.recoveryState) {
+    return false;
+  }
+  // A durable controller is fenced by its released database lease; a legacy
+  // controller still needs process evidence, exactly as for warm recovery.
+  const profile = record(prior.run.runnerProfileJson);
+  const stopped = await evaluateNativeControllerTakeover({ owner: prior.owner, now: new Date(),
+    requireStoppedController: true,
+    remoteLeaseFenced: readNativeControllerRecovery(profile.nativeControllerRecovery, profile.sessionCheckpoint) !== null });
+  if (!stopped.allowed) return false;
+  // 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.
+  const quarantine = resolve(runnerdStateBase(), "quarantine");
+  if (existsSync(quarantine)) {
+    if (!isSafeNativeStateDirectory(quarantine)) return false;
+    for (const name of readdirSync(quarantine)) {
+      if (name.startsWith(`${basename(input.root)}.`) &&
+          readSettledRunnerdProjection(resolve(quarantine, name), identity) === null) return false;
+    }
+  }
+  // Database reads can await. Retire only the exact evidence proven above.
+  if (readSettledRunnerdProjection(input.root, identity) !== controlBytes) return false;
+  const archivedAt = Date.now();
+  const archiveRoot = resolve(input.root, "continuity-breaks", `${archivedAt}-${randomUUID()}`);
+  mkdirSync(archiveRoot, { recursive: true, mode: 0o700 });
+  noteNativeRunnerEvidence(archivedAt);
+  moveNativeContinuityState(input.root, archiveRoot);
+  await input.onLog?.(
+    "stdout",
+    `[paperclip-runner] Run ${identity.runId} finished without a saved runner checkpoint; starting a new runner for this conversation.\n`,
+  );
+  return true;
+}
+
 async function migrateRunnerdStateRootForExecution(input: {
   db: Db;
   execution: NativeExecutionInput;
@@ -5231,6 +5366,14 @@ async function migrateRunnerdStateRootForExecution(input: {
         allowVerifiedBackup: input.allowVerifiedBackup,
         retainedWarmOwnerRunId: input.retainedWarmOwnerRunId,
       });
+      if (
+        verification === "terminal_state_indeterminate" &&
+        !input.restartRecovery &&
+        !input.retainedWarmOwnerRunId &&
+        (await retireFinishedPerTurnRunnerdOwner({ ...input, root: scoped, identity }))
+      ) {
+        return;
+      }
       if (
         verification !== "verified" &&
         verification !== "retained_warm_runner"
@@ -14175,21 +14318,7 @@ async function createRunnerdBackendWithinSessionClaim(
     mkdirSync(archiveRoot, { recursive: true, mode: 0o700 });
     await cursorCloudRuntime?.archiveCheckpoint(archiveRoot, previous);
     noteNativeRunnerEvidence(archivedAt);
-    for (const name of [
-      "control-plane",
-      "runner",
-      "codex-home",
-      "opencode",
-      "acpx",
-      // Backups belong to the retired provider session. Leaving them active
-      // makes the fresh replacement look like ambiguous lost harness state.
-      // Keep their evidence inside the same continuity-break archive.
-      "failover-backups",
-      "provider-pack-authority.json",
-    ]) {
-      const source = resolve(root, name);
-      if (existsSync(source)) renameSync(source, resolve(archiveRoot, name));
-    }
+    moveNativeContinuityState(root, archiveRoot);
     if (remoteCommandRunner && remoteSessionRoot) {
       const escapedSource = remoteSessionRoot.replaceAll("'", "'\\''");
       const archivedRemote = `${remoteSessionRoot}.continuity-break-${archiveToken}`;