fix(native): retire a superseded warm owner instead of failing the new run (#20844)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - Native runs keep a warm Runner between turns of the same task, so
the next run can start fast
> - When the next run needs a new provider session (for example after an
agent setting or skill changes), it first retires the previous run's
warm Runner
> - That retirement uses the previous owner's close. When the close
cannot prove checkpoint suspension, the new run failed before any
provider work, with the previous owner's error
`provider_transport_failed: runner did not durably suspend before
checkpoint`
> - A governed-wait park (a saved question or review request) always
leaves such an owner: its question callback stays unsettled by design. A
failed Launcher stop RPC leaves one too
> - Nothing retried the run, and the task rested as "recovery needed"
with no provider work done
> - This pull request lets the new run retire the old owner when that
owner is contained, and forces the containment when the close left it
unproven
> - The benefit is that a fresh run never fails on its predecessor's
close, and the chaos harness now covers this class

## Linked Issues or Issue Description

**What happened?**
A new run on a task whose previous run left a warm Runner retained
failed within two minutes, before any provider turn. The error was
`provider_transport_failed: runner did not durably suspend before
checkpoint`. The run row recorded `executionRecovery: { kind:
"bootstrap", providerWorkStarted: false }`. Two runs of the same agent
on two tasks failed this way 47 seconds apart, after a configuration
change made both start new provider sessions. One predecessor had parked
on a `request_human_input` question; its callback never settled.

**Expected behavior**
A run that has done no provider work does not fail because its
predecessor's Runner could not prove suspension. The new session never
resumes the old checkpoint. Core must only prove that the old owner is
contained, and then start.

**Steps to reproduce**
1. Run a warm native task that parks on a saved question (chaos workload
Q1).
2. After the asking run settles, change the agent's adapter
configuration, so the next run cannot resume the saved session.
3. Answer the question. The follow-up run retires the parked owner and
fails with the error above.

Related: #18373 added the stopped-owner reset path that this change
extends.

## What Changed

- `closeWarmNativeSession` (server native executor): during retirement
for a different normalized session, accept an unrecoverable close when
the old owner is contained. A plugin task owner must show its
`executionStopped: true` receipt and a completed handoff. Without a task
owner, the Runner's own durable suspension
(`settlement.runnerSuspended`) is the proof. Unsettled Core tool
callbacks of the discarded conversation no longer block the reset.
- With a plugin task owner, any close error qualifies once the Launcher
receipt exists. This includes a failed `stop` that
`releaseRunnerProcessOwnership` rethrows, and "Controller journal drain
timed out": the discarded session's journal lives in its own state root.
Owners without a task keep the narrower rule. `--matrix` now rejects
session faults.
- `createPluginRunnerTaskOwner`: add `retireSupersededTask()`. When
close left the receipt unproven, the new run sends one confirmed `stop`
(not resent after a confirmed stop), takes the credential snapshot, and
calls `complete`. It never reads the old state.
- Same-session continuation, a handoff that still fails, other close
errors and failed instruction collection still block replacement.
- Chaos harness: add session fault SR1
(`tests/restart-chaos/session-faults.ts`). With Q1 and `--lifecycle
warm` (or `--daytona-warm`), the board changes
`adapterConfig.idleTimeoutMs` after the asking run settles and before it
answers. `--tasks 2 --concurrency 2` is the two-concurrent-runs variant.
Q1's oracle applies, plus `session_rotated_before_answer` and
`follow_up_new_session`.
- Docs: `doc/plugins/PLUGIN_RUNNER_EXECUTION.md` and the chaos README
describe the new boundary and cell.

## Verification

- RED on master: `native-session-executor.test.ts -t "permanently
rejected plugin close during reset"` fails the `pending-semantic` and
`stop-unconfirmed-at-close` cases with
`NativeSessionCloseUnrecoverableError: provider_transport_failed: runner
did not durably suspend before checkpoint`. GREEN with this change: all
identity-transition cases pass (49).
- Chaos SR1 local, master server code (`--env local --lifecycle warm
--workloads Q1 --faults SR1`, seeds 301 with 2 tasks and 303 with 1
task): all tasks pass, and the follow-up started a new provider session.
The local parked owner closes cleanly, so the local cell does not
reproduce the failure. It needs a Daytona task owner.
- Chaos SR1 local with this change (seed 302, 2 concurrent tasks on one
agent): 2/2 pass, `follow_up_new_session` true for both.
- Not run: the Daytona SR1 variant. No runner image matches this
checkout's runner content ID, and this change does not touch runner
inputs.
- `pnpm exec vitest run --config tests/restart-chaos/vitest.config.ts
question-handoff scheduler config` passes. Server and harness `tsc
--noEmit` pass.

## Risks

- The reset no longer waits for unsettled Core tool callbacks of the
discarded session. Their run is already terminal and their provider is
stopped, so no result can reach a provider, and the new session never
replays them. Core-side tool executions keep their own receipts under
the old run.
- A local owner without a task needs the Runner's durable suspension. A
local owner that did not suspend still blocks replacement, as before.
- Previous-release story: this changes no persisted state. Warm owners
live only in controller memory, so a deploy drops them as before.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, agentic coding with
tool use (shell, file edits, test runs) in Claude Code.

## 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-session-executor.test.ts

@@ -11882,15 +11882,19 @@ describe("native warm session supervision", () => {
       await closeIdleWarmNativeSessionsForRestart();
     });
 
-    it.each(["retired", "unconfirmed", "other-error", "collection-failed", "same-session", "same-id-provider", "unbound", "pending-semantic"] as const)(
+    it.each(["retired", "unconfirmed", "other-error", "collection-failed", "same-session", "same-id-provider", "unbound", "pending-semantic", "stop-unconfirmed-at-close", "suspended-unbound", "journal-drain-timeout"] as const)(
       "handles a permanently rejected plugin close during reset (%s)", async mode => {
         const { base, next, run } = fixture(`identity-poisoned-close-${mode}`);
         const identity = { runnerInstanceId: "runner", environmentLeaseId: "lease", runId: base.binding.runId,
           normalizedSessionId: base.session.normalizedSessionId!, turnId: "turn", itemId: "item" };
         let stopped = false;
+        let stopFailures = mode === "stop-unconfirmed-at-close" ? 1 : 0;
         const task = vi.fn(async (operation: any) => {
           if (operation.kind === "describe") return { kind: "description", taskId: "old-task", launchPreparationVersion: 1,
             projectAuthCheckVersion: 1, runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } };
+          // The Launcher RPC can fail while the old owner closes; the successor's
+          // forced retirement must send its own confirmed stop.
+          if (operation.kind === "stop" && stopFailures-- > 0) throw new Error("launcher stop unavailable");
           if (operation.kind === "stop") stopped = true;
           if (operation.kind === "status") return { kind: "status", taskId: "old-task", phase: stopped ? "cancelled" : "running", executionStopped: stopped };
           if (operation.kind === "complete" && mode === "unconfirmed") throw new Error("handoff unconfirmed");
@@ -11910,15 +11914,29 @@ describe("native warm session supervision", () => {
           runnerInstanceId: "runner", runnerIngressAuthorized: true,
           runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: "lease", runnerTask: task } });
         const registration = await owner.registration({} as never, base.binding.runId);
-        const failure = mode === "other-error" ? new Error("controller retirement unconfirmed") : new NativeSessionCloseUnrecoverableError(undefined, mode !== "pending-semantic");
+        // A drain that outlives the close grace fails before the checkpoint's stop
+        // and completion, as releaseRunnerProcessOwnership does in the transport.
+        const drainTimedOut = mode === "journal-drain-timeout";
+        const failure = mode === "other-error" ? new Error("controller retirement unconfirmed")
+          : drainTimedOut ? new Error("Controller journal drain timed out; preserve the session before replacing its writer.")
+          // releaseRunnerProcessOwnership rethrows the checkpoint's failed stop
+          // before the transport can raise NativeSessionCloseUnrecoverableError.
+          : mode === "stop-unconfirmed-at-close" ? new Error("launcher stop unavailable")
+          // A local Runner that durably suspended is contained without a task owner.
+          : mode === "suspended-unbound" ? new NativeSessionCloseUnrecoverableError({ runnerSuspended: true, providerDrained: true }, false)
+          : new NativeSessionCloseUnrecoverableError(undefined, mode !== "pending-semantic");
         let closePromise: Promise<void> | undefined;
         const close = vi.fn(() => closePromise ??= (async () => {
           await registration.release();
-          await registration.checkpoint("unsettled").catch(() => undefined);
+          if (!drainTimedOut) await registration.checkpoint("unsettled").catch(() => undefined);
           throw failure;
         })());
         const session = { close };
-        if (mode !== "unbound") backend.bindManagedSession(session as never);
+        // Without a task owner, a close error other than proven suspension still blocks.
+        if (mode !== "unbound" && mode !== "suspended-unbound" && mode !== "other-error") backend.bindManagedSession(session as never);
+        // A governed-wait park leaves its Core tool callback unsettled, and a
+        // failed Launcher stop leaves no receipt: neither belongs to the new session.
+        const succeeds = mode === "retired" || mode === "pending-semantic" || mode === "stop-unconfirmed-at-close" || mode === "suspended-unbound" || drainTimedOut;
         const collectStopped = vi.fn(async () => {
           expect(owner.isRetired()).toBe(true);
           if (mode === "collection-failed") throw new Error("file collection failed");
@@ -11949,7 +11967,7 @@ describe("native warm session supervision", () => {
           const attempt = () => mode === "same-session"
             ? executePaperclipNativeSession({ db: leaseDb(replacement), execution: replacement, runnerInstanceId: "runner", refreshTools: true })
             : run(replacement);
-          if (mode === "retired") {
+          if (succeeds) {
             await attempt();
             expect(state.execute).toHaveBeenCalledTimes(2);
             expect(retirementFailed).not.toHaveBeenCalled();
@@ -11967,7 +11985,9 @@ describe("native warm session supervision", () => {
               expect(state.execute).toHaveBeenCalledTimes(2);
             }
           }
-          expect(task.mock.calls.filter(([operation]) => operation.kind === "stop")).toHaveLength(1);
+          // One confirmed stop per owner; a failed Launcher RPC is the only resend.
+          expect(task.mock.calls.filter(([operation]) => operation.kind === "stop"))
+            .toHaveLength(mode === "stop-unconfirmed-at-close" ? 2 : 1);
           expect(task.mock.calls.some(([operation]) => operation.kind === "read_state")).toBe(false);
         } finally {
           collectStopped.mockResolvedValue(undefined);
@@ -17142,7 +17162,7 @@ describe("runnerd provider runtime wiring", () => {
     state.toolAuthorityDefinitions.mockImplementation(async binding => binding.readRemoteTaskFile ? [{ name: "register_deliverable" }] as never : []);
     const readWorkspaceFile = vi.fn(async () => Buffer.from("proof"));
     const release = vi.fn(async () => {}), resume = vi.fn(async () => "resumed" as const);
-    const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, retiredBeforeClaim: () => false, credentialsUpdated: vi.fn(), rebind:vi.fn() });
+    const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, retireSupersededTask: async () => false, retiredBeforeClaim: () => false, credentialsUpdated: vi.fn(), rebind:vi.fn() });
     const task = vi.fn(async (operation: unknown) => {
       environmentTaskOperationSchema.parse(operation);
       return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, releaseResumeVersion: 1, ...(kind === "codex" ? { workspaceFileReadVersion: 1 } : {}), taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } };