fix(runner): recover pending Stop cleanup after controller restart (#18682)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - The server owns execution, Stop delivery, and usage settlement.
> - A Stop can remain pending when the controller restarts.
> - Recovery must reach the existing session without starting more paid
work.
> - Startup and budget fences currently block that cleanup, and
interrupted runs leave the restart selection.
> - This pull request gives exact audited Stop cleanup a shared
authority check and records its real acknowledgement.
> - The benefit is safe cleanup while usage debt remains honest.

## Linked Issues or Issue Description

**What happened?**

A pending native run-only Stop can retain an unfinished execution owner
after controller loss. Its startup cancellation fence prevents recovery
admission. If teardown marks the run interrupted and clears its task
lock, restart selection also omits it. A recovered session can receive
cancellation without the durable acknowledgement that the normal Stop
request would save.

**Expected behavior**

Recover only the existing remote session for the audited pending Stop.
Keep fresh provider work blocked, including during budget pause. Keep
incomplete usage pending until a real complete receipt arrives.

**Steps to reproduce**

1. Start a native run with a held reservation and an authenticated
remote checkpoint.
2. Persist a run-only Stop and its startup and execution cancellation
fences before acknowledgement.
3. Lose the controller. Leave the coordinator unfinished, mark the run
interrupted, and clear its task lock.
4. Attempt restart recovery and observe the missing claim or
cancellation admission rejection.

Related work: #14733 binds caller Stop requests. #18658 handles plugin
startup cleanup obligations. #18035 proposes broader accounting changes.
This change repairs existing-session Stop recovery and preserves current
accounting guards.

## What Changed

- Add one shared check for the exact run, company, issue, pending Stop,
intent audit, live recovery lease, generation, and remote reattachment
disposition.
- Include eligible interrupted Stop owners in restart selection. Keep
generic interruptions and another task execution owner fenced.
- Reuse only the original held reservation for cleanup. Keep budget
pause effective for fresh, bootstrap, dead-runner, and unproven recovery
work.
- Apply the same cleanup authority at heartbeat runtime selection and
native entry. Skip attachment staging and warm-owner retirement only for
that exact cleanup. Revalidate at session publication and fail closed if
the proof changed.
- Send the existing Stop through the audited cancellation producer
before turn submission. Close the coordinator only after
acknowledgement, including system-owned run-only Stops.
- Give verified cleanup a transport purpose that retains pending
semantic inputs without replay, dispatch, or a semantic reply. Add an
explicit Core retain-only intake mode; preserve authenticated identity,
schema, digest, sequence and replay checks. Ordinary execution still
requires its dispatcher.
- Add PostgreSQL, native protocol, and startup-fence regressions. Update
the execution contract and recovery maps.

## Verification

- The combined Stop regression fails on unchanged production at the
cancellation reservation guard. Interrupted restart selection also fails
before the repair while its running sibling passes.
- An assembled `heartbeatService.recoverNativeRunsAfterRestart`
regression exposes the earlier runtime-selection fence with the rest of
the repair present. The fixed path reaches selector/claim, runtime
selection, original held-budget reuse during pause, dispatch, real
native protocol, audited cancellation, and heartbeat settlement. It
starts no provider turn, acquires no new sandbox, and does not reclaim
the task lock.
- Fail-if-called attachment staging and turn boundaries remain
untouched. A retained warm-owner regression prevents retirement.
Invalidation at session publication fails closed. Actual backend/driver
constructor tests verify cleanup-purpose forwarding, absent default
options, and denial under foreign ownership.
- Missing usage retains accounting debt and the hold. A later complete
receipt records one 25-cent charge and settles that same hold.
Concurrent duplicate cleanup is denied. Both user-owned and system-owned
Stops pass.
- Real transport tests: 9 pass on Node 24.22.1. They preserve ordinary
saved-tool and completion recovery, prove the authority-denial race
before cancellation, and qualify cleanup with pending
ordinary/completion inputs and either handler ordering.
- Real authenticated Core intake: 11 pass. Callback-present and
callback-absent retain-only cases persist and acknowledge unanswered
input without dispatch, recovery or result. Default missing-dispatcher
rejection and digest-integrity/suspension guards remain effective.
Omitting the callback alone first failed by disconnecting; the explicit
retain-only mode fixes that defect.
- Repository `pnpm -r typecheck`, the server/runner-vendor build, map
checks, and module boundary checks passed locally before integrating the
latest retained-startup change. That integration passes 11 native
startup/cleanup compatibility cases and 3 assembled recovery/accounting
cases. The final broad affected-suite attempt was stopped without a
verdict; it is incomplete. Full repository tests and build are delegated
to current-head GitHub CI.
- Review exposed an additional real-process close failure with
unanswered input. The corrected path requires a fresh exact
provider-stop receipt confirming exit, followed by durable event drain
and exact runner suspension. All 11 focused transport cases pass,
including missing/false exit-proof negatives and ordinary recovery
siblings. Runner TypeScript and map validation pass after this
correction.
- The assembled test controls provider, workspace-transfer and GitHub
bootstrap I/O. It does not qualify hosted credentials, physical provider
termination, workspace preservation, or historical receipt recovery.
Local socket-permission setup failures are separate from semantic
regression results. No hosted replay has run on this repair.

- CI exposed an ordinary factory-call compatibility regression and an
obsolete authority-read test fixture. The repair preserves
single-argument ordinary calls and updates the race fixture without
changing its Stop assertions. The complete native executor suite now
passes all 1,049 tests; the ordinary factory boundary and all five
Stop-accounting cases pass. A separate local environment-default case
twice exceeded its five-second wait while other work ran; its prior-head
CI case passed. The same unchanged case passed an isolated diagnostic
without changing its assertion or timeout; the precise contention source
remains unproven.

- The final fixture edit initially used the wrong overloaded promise
type. CI rejected it; the annotation-only correction passes the exact
`pnpm run typecheck:build-gaps` entry point locally, including server
service tests and remaining workspaces. No runtime assertion changed.

- Final exact head `007fd4d04a2d6338c6dc073e97ab470a277fc3d0`: required
`ci / verify` and `ci / e2e`, build, typecheck and test groups pass.
Greptile is 5/5 with both findings resolved. Three expected checks are
skipped; the optional native check is neutral, while its final
current-head comment gives 5/5 from source inspection without a checkout
or test execution. The first server-shard attempt lost its self-hosted
runner connection; one scoped rerun passed. Its original annotation and
unavailable-log evidence are retained.

## Risks

- This changes execution authority. Cleanup requires an audited, exactly
bound pending Stop, an unfinished coordinator, a live remote
reattachment claim, and the original held reservation. It never
authorizes a replacement provider turn.
- Missing checkpoint, original held reservation, termination, or usage
evidence can keep cleanup or accounting pending. The repair does not
manufacture acknowledgement, replies, receipts, or zero cost.
- Retain-only intake is an internal trusted constructor option. It
changes no agent tool permission. Ordinary callback requirements and
semantic dispatch remain unchanged.
- No schema, public API, provider profile, or deployment setting changes
are required.

## Model Used

OpenAI Codex, `gpt-6.1-sol`, Extra High reasoning, with source
inspection, code execution, and regression testing. The context-window
capacity is not exposed in this session.

## 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 described the issue in-PR following the bug report template
- [x] I have not referenced internal/instance-local Paperclip issues or
links
- [x] My branch name describes the change and contains no internal
ticket ID
- [x] I have run affected 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 risks
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open findings
- [x] I will address all reviewer comments before merge

---------

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

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

@@ -38,6 +38,8 @@ import { tmpdir } from "node:os";
 import { inspect } from "node:util";
 import { dirname, join } from "node:path";
 import {
+  activityLog,
+  budgetReservations,
   heartbeatRuns,
   environmentLeases,
   heartbeatRunEvents,
@@ -591,7 +593,7 @@ describe("remote controller restart adoption", () => {
     ).rejects.toThrow("runner_remote_recovery_identity_mismatch");
     expect(execute).toHaveBeenCalledTimes(1);
   });
-  it("wires remote adoption when only controller state survived on the host", async () => {
+  it.each(["execution", "stop_cleanup", "invalid_cleanup"] as const)("wires remote adoption when only controller state survived on the host (%s)", async purpose => {
     const stateBase = await mkdtemp(
       join(tmpdir(), "paperclip-remote-reattach-"),
     );
@@ -619,14 +621,33 @@ describe("remote controller restart adoption", () => {
         join(root, "control-plane", "control-plane-state.json"),
         JSON.stringify(durableControlPlaneState(identity)),
       );
-      await createRunnerdBackend({
-        db: leaseDb(current),
+      const stopIntent = { startupCancellation: { beforeNativeSelection: false }, nativeCancellation: {
+        schema: "paperclip.native-cancellation.v1", scope: "run", reasonCode: "cancellation_run_only", ...current.binding,
+        intentId: `native-cancellation:${randomUUID()}`, intentAuditId: randomUUID(), dispatchState: "pending", dispatched: false,
+        effects: ["release_run_resources"],
+      } };
+      const stopDb = leaseDb(current, { leaseOwner: purpose === "invalid_cleanup" ? "another-owner" : claim.leaseOwner,
+        leaseExpiresAt: new Date(Date.now() + 60_000), controllerGeneration: 2, attempt: 1,
+        recoveryState: "awaiting_runner_reattach", recoveryHistory: [{ disposition: "reattach_remote_runner", controllerGeneration: 2, providerAttempt: 1 }],
+      }, stopIntent);
+      state.createTransport.mockClear();
+      const adopted = createRunnerdBackend({
+        db: purpose === "execution" ? leaseDb(current) : stopDb,
         execution: current,
         runnerInstanceId: "runner",
         runnerExecutionTarget: target as never,
         restartRecovery: claim,
+        ...(purpose === "execution" ? {} : { recoveryPurpose: "stop_cleanup" as const }),
       });
+      if (purpose === "invalid_cleanup") {
+        await expect(adopted).rejects.toThrow("native_cancellation_pending_recovery");
+        expect(state.createTransport).not.toHaveBeenCalled();
+        return;
+      }
+      await adopted;
       state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
+      if (purpose === "stop_cleanup") expect(state.createTransport.mock.calls.at(-1)![0]).toHaveProperty("recoveryPurpose", "stop_cleanup");
+      else expect(state.createTransport.mock.calls.at(-1)![0]).not.toHaveProperty("recoveryPurpose");
       expect(
         state.createTransport.mock.calls.at(-1)![0].adoptExistingRunner?.pid,
       ).toBe(123);
@@ -6437,6 +6458,8 @@ type LeaseCoordinator = {
   controllerBootId?: string;
   controllerPid?: number;
   controllerGeneration?: number;
+  recoveryState?: string;
+  recoveryHistory?: Array<Record<string, unknown>>;
   runId: string;
   companyId: string;
   issueId: string;
@@ -6517,6 +6540,10 @@ function leaseDb(
                     checkoutRunId: null,
                   },
                 ]
+              : table === budgetReservations
+                ? [{ id: "original-held-reservation" }]
+              : table === activityLog
+                ? [{ id: (runResultJson.nativeCancellation as Record<string, unknown> | undefined)?.intentAuditId }]
               : table === heartbeatRunEvents
                 ? priorAccountingEvents.map((event, seq) => ({ seq, eventType: event.eventType, sourceSeq: event.sourceSeq, payload: { prpEvent: event } }))
                 : [];
@@ -6750,12 +6777,22 @@ describe("native startup cancellation fence", () => {
     expect(cancel).toHaveBeenCalledOnce();
     expect(submit).not.toHaveBeenCalled();
   });
-  it.each(["current", "stale"] as const)("reattaches a %s remote recovery claim only to deliver an existing Stop", async mode => {
+  it.each([
+    { mode: "current", startupFence: false },
+    { mode: "stale", startupFence: false },
+    { mode: "current", startupFence: true },
+  ] as const)("reattaches a $mode remote recovery claim only to deliver an existing Stop (startup fence=$startupFence)", async ({ mode, startupFence }) => {
     const { currentNativeControllerIdentity } = await import("./native-restart-recovery.js");
     const controller = await currentNativeControllerIdentity();
-    const resultJson = { nativeCancellation: { schema: "paperclip.native-cancellation.v1", scope: "run",
+    const resultJson = { ...(startupFence ? { startupCancellation: { beforeNativeSelection: false,
+      requestedAt: new Date().toISOString(), errorCode: "cancelled" },
+      cancellation: { source: "operator", expected: true, initiator: { type: "user" },
+        reason: "Operator requested Stop", recordedAt: new Date().toISOString() },
+      executionCancellation: { state: "requested", requestedAt: new Date().toISOString() } } : {}),
+      nativeCancellation: { schema: "paperclip.native-cancellation.v1", scope: "run",
       companyId: execution.binding.companyId, runId: execution.binding.runId, issueId: execution.binding.issueId,
-      dispatchState: "pending", intentId: "native-cancellation:recovered-stop", effects: ["release_run_resources"] } };
+      dispatchState: "pending", dispatched: false, intentId: `native-cancellation:${randomUUID()}`,
+      intentAuditId: randomUUID(), reasonCode: "cancellation_run_only", effects: ["release_run_resources"] } };
     const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })), submit = vi.fn();
     state.execute.mockReset().mockImplementationOnce(async options => {
       try { await options.onSession({ cancel }); submit(); }
@@ -6763,7 +6800,9 @@ describe("native startup cancellation fence", () => {
       throw new Error("a stopped recovery must not submit a turn");
     });
     const db = leaseDb(execution, { leaseOwner: "stop-recovery", controllerBootId: controller.bootId,
-      controllerPid: controller.pid, controllerGeneration: 2, attempt: 1 }, resultJson);
+      controllerPid: controller.pid, controllerGeneration: 2, attempt: 1,
+      leaseExpiresAt: new Date(Date.now() + 60_000), recoveryState: "awaiting_runner_reattach",
+      recoveryHistory: [{ disposition: "reattach_remote_runner", controllerGeneration: 2, providerAttempt: 1 }] }, resultJson);
     await expect(executePaperclipNativeSession({ db, execution, runnerInstanceId: "recovered-stop",
       leaseOwner: "stop-recovery", restartRecovery: { kind: "reattach_remote_runner", runId: execution.binding.runId,
         leaseOwner: "stop-recovery", controllerGeneration: mode === "current" ? 2 : 1, providerAttempt: 1,
@@ -6895,9 +6934,9 @@ describe("native startup restart detachment", () => {
   it.each(["authority", "questions"])("keeps a post-publication Stop owner through %s attachment and restart", async (seam) => {
     const restarting = structuredClone(execution);
     restarting.binding.runId = `restart-stop-after-publication-${seam}`;
-    let published!: () => void, releaseRead!: () => void, acknowledgeStop!: () => void;
+    let releaseRead!: () => void, acknowledgeStop!: () => void;
     let cleaning!: () => void, releaseCleanup!: () => void;
-    const publication = new Promise<void>(resolve => { published = resolve; });
+    const published = vi.fn();
     const read = new Promise<void>(resolve => { releaseRead = resolve; });
     const receipt = new Promise<void>(resolve => { acknowledgeStop = resolve; });
     const cleanupStarted = new Promise<void>(resolve => { cleaning = resolve; });
@@ -6907,12 +6946,20 @@ describe("native startup restart detachment", () => {
     const submit = vi.fn();
     const db = leaseDb(restarting);
     const select = db.select.bind(db);
+    let publishingSession = false;
     db.select = ((fields?: Record<string, unknown>) => {
-      if (seam === "authority" && fields && Object.keys(fields).length === 2 && "status" in fields && "resultJson" in fields) {
-        return { from: () => ({ where: () => ({ limit: () => {
-          published();
-          return read.then(() => [{ status: "running", resultJson: {} }]);
-        } }) }) };
+      if (seam === "authority" && !fields && publishingSession) {
+        const query = select();
+        return { from: (table: unknown) => {
+          const source = query.from(table as never);
+          return table === heartbeatRuns ? { where: (condition: never) => {
+            const filtered = source.where(condition);
+            return { limit: (count: number) => {
+              published();
+              return read.then(() => filtered.limit(count));
+            } };
+          } } : source;
+        } };
       }
       if (seam === "questions" && !fields) {
         const query = select();
@@ -6924,6 +6971,7 @@ describe("native startup restart detachment", () => {
     }) as typeof db.select;
     state.execute.mockReset().mockImplementationOnce(async options => {
       try {
+        publishingSession = true;
         await options.onSession({ detachControllerForRestart: detach, cancel, resolveRuntimeRequest: vi.fn() });
         submit();
       } finally {
@@ -6933,11 +6981,13 @@ describe("native startup restart detachment", () => {
       }
     });
     const outcome = executePaperclipNativeSession({ db, execution: restarting, runnerInstanceId: "runner" }).catch(error => error);
-    await publication;
-    const stopping = cancelNativeSession(restarting.binding.runId, "operator Stop after publication");
     const completed = vi.fn();
-    const shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).then(completed);
+    let stopping: Promise<boolean> | undefined;
+    let shutdown: Promise<unknown> | undefined;
     try {
+      await vi.waitFor(() => expect(published).toHaveBeenCalled(), { timeout: 5_000 });
+      stopping = cancelNativeSession(restarting.binding.runId, "operator Stop after publication");
+      shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).then(completed);
       releaseRead();
       await new Promise(resolve => setImmediate(resolve));
       expect(cancel).toHaveBeenCalledOnce();
@@ -9772,6 +9822,66 @@ describe("native warm session supervision", () => {
       expect(state.execute).toHaveBeenCalledTimes(1);
     });
 
+    it.each([false, true])("keeps warm owners and attachment staging untouched during exact Stop cleanup (invalidated=%s)", async invalidated => {
+      const { base, next, run } = fixture("identity-native-stop-cleanup");
+      const close = vi.fn<() => Promise<void>>(async () => { throw new Error("Stop cleanup must not retire another warm owner"); });
+      const cancel = vi.fn(() => ({ cleanup: Promise.resolve() }));
+      const originalStaging = state.stageNativeRunnerWakeAttachments.getMockImplementation()!;
+      state.execute.mockReset().mockImplementationOnce(async options => { await options.onSession?.({ close }); return result; });
+      await run(base);
+      await createRunnerdBackend({ db: leaseDb(next), execution: next, runnerInstanceId: "stop-cleanup" });
+      state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
+      const root = state.createTransport.mock.calls.at(-1)![0].stateDirectory!;
+      const remoteIdentity = { runId: next.binding.runId, normalizedSessionId: next.session.normalizedSessionId!,
+        runnerInstanceId: "stop-cleanup", environmentLeaseId: next.binding.executionWorkspaceId, turnId: "stop-turn", itemId: "stop-item" };
+      const authority = new DurablePrpControlPlane({ stateDirectory: join(root, "control-plane"), identity: remoteIdentity,
+        expectedRunnerVersion: "stop-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
+      await authority.flushCheckpoint();
+      await authority.stop();
+      const controller = await import("./native-restart-recovery.js").then(module => module.currentNativeControllerIdentity());
+      const owner = "exact-stop-cleanup";
+      const resultJson = { cancelledByActorType: "user", cancelledByUserId: "test-board",
+        startupCancellation: { beforeNativeSelection: false }, nativeCancellation: {
+          schema: "paperclip.native-cancellation.v1", scope: "run", reasonCode: "cancellation_run_only", ...next.binding,
+          intentId: `native-cancellation:${randomUUID()}`, intentAuditId: randomUUID(), dispatchState: "pending", dispatched: false,
+          effects: ["release_run_resources"],
+        } };
+      const db = leaseDb(next, { leaseOwner: owner, leaseExpiresAt: new Date(Date.now() + 60_000),
+        controllerBootId: controller.bootId, controllerPid: controller.pid, controllerGeneration: 2, attempt: 1,
+        recoveryState: "awaiting_runner_reattach", recoveryHistory: [{ disposition: "reattach_remote_runner", controllerGeneration: 2, providerAttempt: 1 }],
+      }, resultJson);
+      state.stageNativeRunnerWakeAttachments.mockClear().mockImplementation(async () => { throw new Error("Stop cleanup must not stage attachments"); });
+      const submit = vi.fn(() => { throw new Error("Stop cleanup cannot submit a turn"); });
+      state.execute.mockImplementationOnce(async options => {
+        if (invalidated) {
+          Reflect.deleteProperty(resultJson, "startupCancellation");
+          Reflect.deleteProperty(resultJson, "nativeCancellation");
+        }
+        try { await options.onSession?.({ cancel }); submit(); throw new Error("Stop cleanup cannot return to execution"); }
+        finally { await options.onSession?.(null); }
+      });
+      try {
+        await expect(executePaperclipNativeSession({ db, execution: next, runnerInstanceId: "stop-cleanup", useRunnerd: true,
+          backend: {} as never, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace",
+            sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId: "existing-sandbox" }, runner: { execute: async request => ({
+              exitCode: 0, signal: null, pid: 123, startedAt: new Date().toISOString(), timedOut: false, stderr: "", stdout: request.args?.[1]?.includes("base64")
+                ? Buffer.from(JSON.stringify({ ...remoteIdentity, lifecycle: "running" })).toString("base64")
+                : request.args?.[1]?.includes("cat --") ? "nonce\n123\n2026-09-19T10:00:00.000Z\nstop-cleanup\nlinux:abcd-1234:100\n" : "",
+            }) } }, leaseOwner: owner, restartRecovery: { kind: "reattach_remote_runner", runId: next.binding.runId, leaseOwner: owner,
+            controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null,
+            remote: { providerLeaseId: "existing-sandbox", remoteCwd: "/workspace" } },
+        })).rejects.toThrow("native_cancellation_pending_recovery");
+        expect(close).not.toHaveBeenCalled();
+        expect(state.stageNativeRunnerWakeAttachments).not.toHaveBeenCalled();
+        expect(cancel).toHaveBeenCalledTimes(invalidated ? 0 : 1);
+        expect(submit).not.toHaveBeenCalled();
+      } finally {
+        state.stageNativeRunnerWakeAttachments.mockReset().mockImplementation(originalStaging);
+        close.mockImplementation(async () => undefined);
+        await closeIdleWarmNativeSessionsForRestart();
+      }
+    });
+
     it("does not retire an active turn when a new session identity arrives", async () => {
       const { base, next, run } = fixture("identity-active");
       let finish!: () => void;