fix(native): settle tool callbacks and holds that a server restart orphaned (#20628)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs send provider tool calls to the server as durable semantic inputs. The server runs the tool and saves the result in the controller journal > - Close must prove that every admitted tool call has a saved result. Otherwise it reports `native_session_settlement_incomplete` and the runner holds a cleanup gate (`native_session_cleanup_quarantined`) > - A server restart in the middle of a tool call leaves its input without a result. The new controller asks the tool authority for a receipt. Tools without a receipt (assigned app tools, `connections_search`) return nothing, so the input stays pending forever > - Runs that already failed this way kept no-replay holds for a turn that had completed. The task stayed blocked with no Retry and no continuation > - Restarts are normal. Work that a restart interrupts must settle by itself, not wait for a person > - This pull request settles such a callback as interrupted, with an unknown outcome, when its owner is provably gone, and settles the holds that earlier releases left once their controller is gone. It never runs the tool or the turn again > - The benefit is that close drains after a restart, and tasks that earlier releases blocked this way continue on their own ## Linked Issues or Issue Description Refs #20572 (that PR fixed the presentation of this quarantine. Its Risks section names this runner drain rule as a separate follow-up.) Refs #20641 (it settles a resolved no-replay hold when a sibling action saved the run's files. This PR covers a different case: the hold's own recovery saved the files, and its run's turn completed. Both use the same recovery receipt, `completedWorkspaceRecoveryCondition`.) **What happened?** A remote native run started an assigned app tool call. The server restarted 3 seconds later. The new server adopted the runner, and the provider turn completed about 9 minutes later. The tool call never got a saved result. Close reported `native_session_settlement_incomplete` with `runnerSuspended: true`, `providerDrained: false` and one pending callback. The session quarantined, and the same-run recovery was refused at admission. A second run showed the same shape with a `connections_search` callback after a later restart. The first run then had two resolved-but-blocked holds with `preserve_without_replay_v1` and an unknown action outcome: the cleanup-quarantine hold, and a continuation hold whose workspace recovery saved the files. The task view said "Saved work recovered. Send a message to continue." with no Retry and no continuation. The task waits on a child task and stays blocked. **Expected behavior** After a restart, a tool call without a saved result settles automatically. A durable receipt is replayed. Without a receipt, the agent is told that the outcome is unknown. Close proves the drain, and no cleanup gate is left. Holds that earlier releases left for a completed turn settle without a person, and the task continues when its own wait ends. **Steps to reproduce** 1. Start a native run and call an assigned app tool. 2. Restart the server while the tool call runs. 3. Let the adopted turn complete, then close the session. Close reports `native_session_settlement_incomplete` and quarantines. The task stays blocked by the no-replay holds. **Paperclip version or commit** master at fd6e71a7ff. ## What Changed - Root cause: `DurablePrpControlPlane.#recoverPendingSemanticInputsOnce` (`packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts`) keeps an input pending when `recoverSemanticToolInput` returns null. `PaperclipRunnerToolAuthority.recover` returns null for every tool without a durable receipt. Nothing else could settle the input, so `semanticToolResultsSettled()` stayed false and `#closeOnce` in `runnerd-codex-transport.ts` threw `NativeSessionCloseUnrecoverableError`. - The control plane now passes `admission` to `recoverSemanticToolInput`: `this_controller` when this instance dispatched the callback live, `earlier_controller` when it restored the input from an earlier controller's journal. The field is not written to the result command or the journal. A live dispatch leaves this record when its result is journaled. - New `packages/paperclip-runner/src/control-plane/orphaned-tool-callback.ts`: - `dispatchToolCallback` records the callbacks that this process runs, and the ones it saw fail. - `recoverToolCallback` replays a durable receipt first. Without one, it settles an `earlier_controller` callback that this process does not run as `tool_call_interrupted_by_restart` (error result, `outcome: "unknown"`, `retryable: false`). A `semantic_tool_outcome_unknown` rejection from the authority settles the same way, only under the same owner-gone proof. - `native-session-executor.ts`: `SessionAuthorityEpoch.execute` runs through `dispatchToolCallback`. `SessionAuthorityEpoch.recover` runs through `recoverToolCallback`. The three recovery call sites pass `admission`. - New ownership kind `native_completed_turn_cleanup_hold` (`server/src/services/native-runtime/completed-turn-cleanup-hold-settlement.ts`, registered in `recovery/ownership-registry.ts`). It lists a native run that failed with `native_session_cleanup_quarantined` after its saved checkpoint recorded a completed turn, and that has a system-recorded no-replay hold. After `proveOwnerGone` proves the closing controller gone (its in-process cleanup gate went with it), it settles each such hold as `restored`, removes its replay block, and logs `issue.recovery_action_resolved`. A hold that still protects files not verified as saved stays for workspace recovery. Operator decisions are never reopened. It starts no turn. `heartbeat.ts` registers the ordinary resume path (`resumeRemoteStopComments`) for saved user input. - Docs: `doc/execution-semantics.md` and `packages/paperclip-runner/docs/durable-recovery.md`. ## Verification - Red before the fix, green after: - `packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts` "host-tool callback orphaned by a server restart". It uses the real control plane and an authenticated runner. A callback starts 3 seconds before the restart, the old controller retires with no result, the new controller adopts the runner, the turn completes, and then recovery runs. `paperclip_call_assigned_tool` and `connections_search` settle as `tool_call_interrupted_by_restart`, the runner accepts the late result, `semanticToolResultsSettled()` becomes true (the condition close requires), and the tool is not dispatched again. Before the fix, 4 of 5 cases failed: recovery received no admission and the callback stayed pending. - `server/src/__tests__/completed-turn-cleanup-hold-settle.test.ts` uses the staging shape of both holds (real Postgres, rows as earlier releases saved them). Before the fix, the ownership reconciler did not list them (4 of 4 failed), and the execution blocker was the files hold with `canRetry: false` and `canContinue: false`. After the fix, both holds settle as `restored`, the execution blocker is null, no run is created, the task keeps `blocked`, and the next pass lists nothing. - Safety cases: - A callback that this process still runs stays pending. After it finishes, the next recovery pass settles it. - A callback that this controller dispatched and saw rejected stays pending, as before. - A durable receipt is replayed and not replaced. - Holds stay while the closing controller is alive (its gate may still exist). - A hold whose files are not saved stays; the run's other hold settles. - Holds of a run whose turn did not complete are never listed. - Focused suites pass: `durable-prp-control-plane.test.ts` and `runnerd-codex-transport.test.ts` (506 tests); runner adoption, release-resume, semantic readiness, ACPX reconciliation and external driver tests; `native-session-executor.test.ts` and `native-runner-restart-recovery.integration.test.ts` (1496 tests); `ownership-reconciliation-report.test.ts` and `interrupted-preparation-hold-settle.test.ts`. - `tsc --noEmit -p packages/paperclip-runner/tsconfig.test.json` passes. `pnpm --filter @paperclipai/server typecheck` passes. - `pnpm verify` returns exit 2 (a global doc input changed), so required CI supplies the broad coverage. ## Risks - Previous-release story: no persisted shape changes. `admission` is computed at recovery and never saved. A journal that an earlier release saved with a pending callback restores as `earlier_controller`, so that session settles at its next adoption. Holds that earlier releases saved settle on the reconciler's first pass after deploy. No migration. - No double execution: recovery never calls the tool again. It replays a durable receipt or reports an unknown outcome. The agent is told to check the current state before it repeats the call. Hold settlement starts no turn and replays nothing. - The owner-gone proof for a callback: the new controller holds the run's checkpoint fence, and the earlier controller retired its journal writer. If the earlier process is still alive during an overlapping deploy, it can still finish the tool, but it can never record a result in this session. The unknown outcome is accurate in that case. - Hold settlement requires `proveOwnerGone` on the closing controller's stamp, because the cleanup gate lives in that process. On a server that has not restarted since the quarantine, the holds stay. - The in-process failure memory is bounded to 4,096 entries. Eviction only lets a much older failed callback settle as interrupted later. It never replays it. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use and code 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>
2026-11-11 05:11:49 +01:00 · 2026-11-10 14:39:45 -06:00
1 parent d06f43b14f
commit 4398a723c0
@@ -1759,6 +1759,34 @@ it; a new message continues once the previous session has stopped. These rules
 read only `errorCode` and `runnerProfileJson.sessionCheckpoint`, so runs that
 earlier releases quarantined are shown the same way without a migration.

+A server restart in the middle of a host-tool callback (for example an assigned
+app tool or `connections_search`) must not cause that gate. The replacement
+controller restores the old controller's journal and asks the tool authority for
+the call's durable receipt. A receipt is replayed. Without one, the callback's
+owner is provably gone when an earlier controller dispatched it and this server
+process is not running it: the earlier controller's journal writer is retired
+and its checkpoint fence is superseded, so it can never record a result. The
+callback then settles as `tool_call_interrupted_by_restart` with an unknown
+outcome, and the agent is told to check the current state before repeating it.
+The tool is never executed again. Close can then prove the drain and settles
+normally. A callback that this server process still runs, or that this
+controller dispatched and saw fail, stays pending as before. The saved journal
+shape does not change, so a session that an earlier release left with such a
+callback settles at its next adoption without a migration.
+
+Runs that earlier releases already failed this way kept no-replay holds
+(`preserve_without_replay_v1`, replay blocked, action outcome unknown) for a
+turn that had completed: one for the cleanup quarantine, and often one whose
+workspace recovery saved the files ("Saved work recovered. Send a message to
+continue." with no Retry). Nothing is left to replay. The shared ownership
+reconciler lists them as `native_completed_turn_cleanup_hold`. Once it proves
+the controller that closed the session gone (its in-process cleanup gate went
+with it), it settles each such hold as `restored` and removes its replay block.
+A hold that still protects files that are not verified as saved stays for
+workspace recovery. Settling starts no turn: the task keeps its own state (for
+example, waiting on a child task) and continues through normal admission when
+that wait ends. Saved user input is resumed through the ordinary resume path.
+
 A continuation cancelled by the queued-run gate while waiting for review did not start a provider. Its matching dispatch-gate receipt and unclaimed execution fields keep this deliberate wait out of unknown-action reconciliation. Pending review and decision gates still apply. If an older server already created a recovery hold for this exact state, Retry or a new user message can request one fresh turn after the server verifies there is no provider, coordinator, or unfinished cleanup. Saved user input is reconsidered through the same admission path; the cancelled turn is not replayed.

 Other legacy queued-run gate cancellations also stay out of unknown-action reconciliation when the terminal run has no retained execution authority, session, workspace recovery, resources, or execution events. The same proof removes the effect of an old false recovery hold only for its exact company, task, and run; the audit row remains. Started or conflicting work retains its hold. Review waits retain their guarded Retry path. New work still passes the current approval, dependency, budget, pause, and ownership gates.
@@ -591,8 +591,18 @@ has a matching completed result-delivery command and the provider has drained.
 A callback rejected before business dispatch can report
 `semantic_tool_not_dispatched`. A callback rejected after dispatch has an
 unknown outcome; the controller must not invent a failure or execute it again.
-After controller restart, an already committed input without a result remains
-unsettled. It requires authoritative reconciliation, not an automatic retry.
+After controller restart, an already committed input without a result requires
+authoritative reconciliation, not an automatic retry. Recovery passes
+`admission` to `recoverSemanticToolInput`: `this_controller` when this
+controller instance dispatched the callback live, `earlier_controller` when it
+restored the input from an earlier controller's journal. The server replays a
+durable receipt when one exists. Otherwise `recoverToolCallback`
+(`control-plane/orphaned-tool-callback.ts`) settles an `earlier_controller`
+callback that this process is not running (`dispatchToolCallback`) as
+`tool_call_interrupted_by_restart`, an error result with an unknown outcome. The
+earlier controller can never record a result, so this is the callback's only
+settlement, and it never executes the operation again. Every other unresolved
+input stays unsettled.
 A different case is a runner crash after the controller saved the result but
 before its delivery command completed. Codex/OpenCode explicitly opt in to
 reconciling that exact `semantic_tool.result` command through their durable tool
@@ -48,6 +48,7 @@ import {
  waitForProcess,
  type RunnerProcessLaunchSpec,
 } from "./durable-prp-control-plane.js";
+import { dispatchToolCallback, recoverToolCallback, TOOL_CALLBACK_INTERRUPTED_CODE } from "./orphaned-tool-callback.js";
 import type { DurableRecoveryIdentity } from "./prp-transport-types.js";

 type WritableCoreStore = DurablePrpControlPlaneStore & {
@@ -5507,3 +5508,186 @@ it("retires late semantic receipts without overwriting a successor journal or re
    rmSync(root, { recursive: true, force: true });
  }
 });
+
+describe("host-tool callback orphaned by a server restart", () => {
+  // Incident shape: a callback started at 14:29:25, the server restarted at
+  // 14:29:28 before it returned, the replacement controller adopted the runner
+  // and the provider turn completed at 14:38:26. Close then needs every
+  // admitted callback's durable result (`semanticToolResultsSettled`).
+  const startedAt = "2026-11-10T14:29:25.000Z";
+  const turnCompletedAt = "2026-11-10T14:38:26.000Z";
+  type RecoveryCall = Parameters<NonNullable<ConstructorParameters<typeof DurablePrpControlPlane>[0]["recoverSemanticToolInput"]>>[0];
+
+  function callbackInput(operationId: string, callId: string): Record<string, unknown> {
+    const event = semanticInputEvent(1, { name: "submit_report", arguments: { draft: true } });
+    const payload = event.payload as Record<string, unknown>;
+    payload.emittedAt = startedAt;
+    const tool = (payload.payload as Record<string, unknown>).semantic_tool as Record<string, unknown>;
+    tool.operationId = operationId;
+    tool.callId = callId;
+    return event;
+  }
+
+  function turnCompleted(sourceSeq: number): Record<string, unknown> {
+    const event = semanticInputEvent(sourceSeq);
+    const payload = event.payload as Record<string, unknown>;
+    payload.sourceEventId = `turn-completed-${sourceSeq}`;
+    payload.eventType = "turn.completed";
+    payload.emittedAt = turnCompletedAt;
+    payload.payload = { status: "completed" };
+    return event;
+  }
+
+  // The server's recovery (SessionAuthorityEpoch.recover): replay a durable
+  // receipt, else settle a callback whose owner is gone. These tools keep no
+  // receipt, so the authority returns null, as PaperclipRunnerToolAuthority does.
+  const serverRecovery = (authorityRecover: () => Promise<{ result: unknown } | null> = async () => null) =>
+    vi.fn((call: RecoveryCall) => recoverToolCallback({ runId: call.correlation.runId, admission: call.admission,
+      call: { tool: call.operationId, callId: call.callId, arguments: call.input }, recover: authorityRecover }));
+
+  async function restartWithPendingCallback(operationId: string, callId: string, dispatch: () => Promise<{ result: unknown }>) {
+    const root = mkdtempSync(resolve(tmpdir(), "runner-orphaned-callback-"));
+    const options = { stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest };
+    const original = new DurablePrpControlPlane({ ...options, onSemanticToolInput: dispatch });
+    await original.start();
+    const client = (await authenticate(original, original.issueBootstrapTicket()))!;
+    sendSecure(client, callbackInput(operationId, callId));
+    expect((await receiveSecure(client))!.kind).toBe("ack");
+    // The server restarts 3s later: the controller detaches and retires its
+    // callback journal writer before the callback returns.
+    await original.stop();
+    await original.drainPendingConnectionProcessing();
+    original.retireSemanticToolCallbacks();
+    client.socket.destroy();
+    return { root, options, leaseToken: client.leaseToken! };
+  }
+
+  // The adopted runner resumes after the callback input the old controller ACKed.
+  const resumeRunner = (core: DurablePrpControlPlane, leaseToken: string) =>
+    authenticate(core, leaseToken, identity, expectedRunnerDigest, undefined, false, 1, 1);
+
+  async function settle(core: DurablePrpControlPlane, runner: AuthenticatedClient) {
+    const command = (await receiveSecure(runner))!;
+    expect(command).toMatchObject({ kind: "command", payload: { type: "semantic_tool.result" } });
+    const payload = command.payload as Record<string, unknown>;
+    // runnerd accepts a late result after its turn closed (`settled_after_turn`).
+    sendSecure(runner, { protocol: "paperclip.runner", version: 1, kind: "command_result", payload: {
+      commandId: payload.commandId, commandType: payload.type, controllerSeq: payload.controllerSeq,
+      status: "completed", result: { status: "settled_after_turn", callId: (payload.payload as Record<string, unknown>).callId },
+    } });
+    await vi.waitFor(() => expect(core.commandOutcome(String(payload.commandId))?.status).toBe("completed"));
+    await core.flushCheckpoint();
+    return payload.payload as Record<string, unknown>;
+  }
+
+  it.each([
+    // An assigned app tool call, adopted after the restart.
+    { operationId: "paperclip_call_assigned_tool", callId: "exec-d7e28482-f90d-48e6-abdd-c8a0a94dbfb0" },
+    // A connection search, adopted after a later restart.
+    { operationId: "connections_search", callId: "exec-0c3b0462-a1e1-488a-a4ee-e4748f30ce29" },
+  ])("settles $operationId as interrupted after the turn completed, so close drains", async ({ operationId, callId }) => {
+    // The earlier server process owned this dispatch; it never returns here.
+    const { root, options, leaseToken } = await restartWithPendingCallback(operationId, callId,
+      () => new Promise<{ result: unknown }>(() => undefined));
+    const authorityRecover = vi.fn(async () => null);
+    const recover = serverRecovery(authorityRecover);
+    const redispatch = vi.fn(async () => ({ result: "must not execute again" }));
+    const replacement = new DurablePrpControlPlane({ ...options, onSemanticToolInput: redispatch, recoverSemanticToolInput: recover });
+    try {
+      await replacement.start();
+      const runner = (await resumeRunner(replacement, leaseToken))!;
+      sendSecure(runner, turnCompleted(2));
+      expect((await receiveSecure(runner))!.kind).toBe("ack");
+      expect(replacement.semanticToolResultsSettled()).toBe(false);
+      await replacement.recoverPendingSemanticInputs();
+      expect(recover).toHaveBeenCalledWith(expect.objectContaining({ callId, operationId, admission: "earlier_controller" }));
+      expect(authorityRecover).toHaveBeenCalledOnce();
+      const delivered = await settle(replacement, runner);
+      expect(delivered).toMatchObject({ callId, operationId, isError: true,
+        result: { error: { code: TOOL_CALLBACK_INTERRUPTED_CODE, outcome: "unknown", retryable: false } } });
+      expect(delivered).not.toHaveProperty("admission");
+      expect(redispatch).not.toHaveBeenCalled();
+      // Close requires exactly this to prove the provider drained.
+      expect(replacement.semanticToolResultsSettled()).toBe(true);
+      runner.socket.destroy();
+    } finally {
+      await replacement.stop();
+      rmSync(root, { recursive: true, force: true });
+    }
+  });
+
+  it("replays a durable receipt instead of settling the callback as interrupted", async () => {
+    const callId = "exec-receipted";
+    const { root, options, leaseToken } = await restartWithPendingCallback("call_api", callId,
+      () => new Promise<{ result: unknown }>(() => undefined));
+    const recover = serverRecovery(async () => ({ result: { saved: "receipt" } }));
+    const replacement = new DurablePrpControlPlane({ ...options, recoverSemanticToolInput: recover });
+    try {
+      await replacement.start();
+      const runner = (await resumeRunner(replacement, leaseToken))!;
+      await replacement.recoverPendingSemanticInputs();
+      expect(await settle(replacement, runner)).toMatchObject({ callId, isError: false, result: { saved: "receipt" } });
+      expect(replacement.semanticToolResultsSettled()).toBe(true);
+      runner.socket.destroy();
+    } finally {
+      await replacement.stop();
+      rmSync(root, { recursive: true, force: true });
+    }
+  });
+
+  it("keeps a callback pending while this process still runs it", async () => {
+    const callId = "exec-still-running";
+    // This process still owns the dispatch (for example a session recreated in
+    // the same server). Its owner is alive, so it is never settled as abandoned.
+    let finish!: (value: { result: unknown }) => void;
+    const running = dispatchToolCallback(identity.runId, callId,
+      () => new Promise<{ result: unknown }>(resolveCall => { finish = resolveCall; }));
+    const { root, options, leaseToken } = await restartWithPendingCallback("paperclip_call_assigned_tool", callId, () => running);
+    const recover = serverRecovery();
+    const replacement = new DurablePrpControlPlane({ ...options, recoverSemanticToolInput: recover });
+    try {
+      await replacement.start();
+      const runner = (await resumeRunner(replacement, leaseToken))!;
+      await replacement.recoverPendingSemanticInputs();
+      expect(recover).toHaveBeenCalledWith(expect.objectContaining({ callId, admission: "earlier_controller" }));
+      expect(replacement.store.state.commands).toEqual([]);
+      expect(replacement.semanticToolResultsSettled()).toBe(false);
+      // Once that owner finishes, its retired journal cannot record the
+      // result, so the next recovery pass settles the callback.
+      finish({ result: "late" });
+      await running;
+      await replacement.recoverPendingSemanticInputs();
+      expect(await settle(replacement, runner)).toMatchObject({ callId,
+        result: { error: { code: TOOL_CALLBACK_INTERRUPTED_CODE } } });
+      expect(replacement.semanticToolResultsSettled()).toBe(true);
+      runner.socket.destroy();
+    } finally {
+      await replacement.stop();
+      rmSync(root, { recursive: true, force: true });
+    }
+  });
+
+  it("keeps a callback this controller dispatched pending when its dispatcher rejected", async () => {
+    const root = mkdtempSync(resolve(tmpdir(), "runner-own-callback-"));
+    const recover = serverRecovery();
+    const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest,
+      // A rejected dispatcher does not prove that its effect rolled back.
+      onSemanticToolInput: () => dispatchToolCallback(identity.runId, "call-1", async () => { throw new Error("uncertain"); }),
+      recoverSemanticToolInput: recover });
+    try {
+      await core.start();
+      const client = (await authenticate(core, core.issueBootstrapTicket()))!;
+      sendSecure(client, semanticInputEvent());
+      expect((await receiveSecure(client))!.kind).toBe("ack");
+      await vi.waitFor(() => expect(core.semanticToolSettlementDiagnostics()).toMatchObject({ failures: [expect.anything()] }));
+      await core.recoverPendingSemanticInputs();
+      expect(recover).toHaveBeenCalledWith(expect.objectContaining({ callId: "call-1", admission: "this_controller" }));
+      expect(core.store.state.commands).toEqual([]);
+      expect(core.semanticToolResultsSettled()).toBe(false);
+      client.socket.destroy();
+    } finally {
+      await core.stop();
+      rmSync(root, { recursive: true, force: true });
+    }
+  });
+});
@@ -302,8 +302,15 @@ export interface DurablePrpControlPlaneOptions {
  }) => Promise<{ readonly result: unknown; readonly isError?: boolean }>;
  /** Persist the canonical event before the runner receives its cumulative ACK. */
  onCommittedEvent?: (event: PrpEvent) => Promise<void>;
+  /**
+   * Resolve a saved input that has no durable result. Return null to keep it
+   * pending. `admission` says whether this controller instance dispatched the
+   * callback itself, or restored it from a journal that an earlier controller
+   * wrote (for example one that a server restart replaced).
+   */
  recoverSemanticToolInput?: (
-    input: Parameters<NonNullable<DurablePrpControlPlaneOptions["onSemanticToolInput"]>>[0],
+    input: Parameters<NonNullable<DurablePrpControlPlaneOptions["onSemanticToolInput"]>>[0]
+      & { readonly admission: SemanticInputAdmission },
  ) => Promise<{ readonly result: unknown; readonly isError?: boolean } | null>;
  /** Stop this exact owner after a proven, authenticated permanent integrity fault. */
  onProtocolIntegrityError?: (
@@ -314,6 +321,14 @@ export interface DurablePrpControlPlaneOptions {
  connectionLeaseTtlMs?: number;
 }

+/**
+ * Who dispatched a saved semantic input: this controller instance, or an
+ * earlier one whose journal this controller restored. This controller now
+ * owns the journal, so an earlier controller can never record a result for
+ * its callback. The caller decides whether that callback's owner is gone.
+ */
+export type SemanticInputAdmission = "this_controller" | "earlier_controller";
+
 export interface DurablePrpCheckpoint {
  /** Read-only protocol evidence. Does not grant attachment or execution authority. */
  readonly recoveryEvidence: {
@@ -2030,6 +2045,10 @@ export class DurablePrpControlPlane {
  #connectionProcessing = new Map<AuthorityConnection, Promise<void>>();
  #connectionLeaseFailure: DurablePrpConnectionLeaseFailure | null = null;
  #pendingSemanticCalls = new Set<string>();
+  /** Result command ids whose callback this instance dispatched live and has
+   * not journaled a result for. A queued result removes its id, so only
+   * in-flight and failed dispatches remain. */
+  #liveDispatchedSemanticCalls = new Set<string>();
  #semanticCallbacksRetired = false;
  #semanticResultPersistenceFailed = false;
  #semanticResultFailures: Array<{
@@ -2308,7 +2327,9 @@ export class DurablePrpControlPlane {
          sourceEventId: event.sourceEventId, sourceEventType: event.eventType,
          correlation: { runId: identity.runId, normalizedSessionId: identity.normalizedSessionId,
            turnId: identity.turnId, itemId: event.itemId ?? identity.itemId } };
-        const outcome = await this.#recoverSemanticToolInput(call);
+        const admission: SemanticInputAdmission = this.#liveDispatchedSemanticCalls.has(commandId)
+          ? "this_controller" : "earlier_controller";
+        const outcome = await this.#recoverSemanticToolInput({ ...call, admission });
        if (this.#identity !== identity || this.#semanticCallbacksRetired || this.#retainsSourceInputs()) return;
        if (!outcome || this.#getCommand(commandId)) continue;
        const { input, ...metadata } = call;
@@ -4403,7 +4424,10 @@ export class DurablePrpControlPlane {
      && !this.#getCommand(liveCommandId) && !this.#pendingSemanticCalls.has(liveCommandId);
    // Reserve the original callback before exposing its checkpoint wait to
    // recovery. A missing saved outcome must never suppress first dispatch.
-    if (ownsLiveDispatch) this.#pendingSemanticCalls.add(liveCommandId!);
+    if (ownsLiveDispatch) {
+      this.#pendingSemanticCalls.add(liveCommandId!);
+      this.#liveDispatchedSemanticCalls.add(liveCommandId!);
+    }
    let dispatched = false;
    try {
      try {
@@ -4443,6 +4467,8 @@ export class DurablePrpControlPlane {
              commandId,
              true,
            );
+            // Its journaled result settles it; recovery never asks again.
+            this.#liveDispatchedSemanticCalls.delete(commandId);
          } catch (error) {
            // A result that cannot fit the bounded durable journal cannot be
            // acknowledged as a usable tool response. Force a reconnect so
@@ -0,0 +1,116 @@
+/**
+ * Host-tool callbacks orphaned by a server restart.
+ *
+ * A provider tool call reaches Paperclip as a durable semantic input. The
+ * server process that hosts the run's controller executes it and journals the
+ * result. When that server restarts mid-call, the replacement controller
+ * restores the journal: the input is there, its result is not, and the old
+ * process can never record one (its journal writer is retired and its
+ * checkpoint fence is superseded). Without a decision the input stays pending
+ * forever, so close cannot prove the provider drained and the session
+ * quarantines.
+ *
+ * Recovery first asks the tool authority for a durable receipt and replays
+ * only that. When none exists and this callback's owner is provably gone, the
+ * callback settles as interrupted with an unknown outcome. It is never
+ * executed again. A callback that this process is still running, or that this
+ * process dispatched and saw fail, stays pending.
+ */
+
+import type { SemanticInputAdmission } from "./durable-prp-control-plane.js";
+
+export const TOOL_CALLBACK_INTERRUPTED_CODE = "tool_call_interrupted_by_restart";
+
+type CallbackOutcome = { readonly result: unknown; readonly isError?: boolean };
+type ToolCallbackCall = { tool: string; callId: string; arguments: unknown };
+
+const FAILED_DISPATCH_MEMORY = 4_096;
+const runningDispatches = new Map<string, number>();
+const failedDispatches = new Set<string>();
+
+function callbackKey(runId: string, callId: string): string {
+  return `${runId}\0${callId}`;
+}
+
+/** Run one host-tool callback and record that this process owns it while it
+ * runs. A rejected dispatch does not prove that its effect rolled back, so
+ * this process remembers it and never settles it as abandoned. */
+export async function dispatchToolCallback<T>(runId: string, callId: string, execute: () => Promise<T>): Promise<T> {
+  const key = callbackKey(runId, callId);
+  runningDispatches.set(key, (runningDispatches.get(key) ?? 0) + 1);
+  let completed = false;
+  try {
+    const value = await execute();
+    completed = true;
+    return value;
+  } finally {
+    const remaining = (runningDispatches.get(key) ?? 1) - 1;
+    if (remaining > 0) runningDispatches.set(key, remaining);
+    else runningDispatches.delete(key);
+    if (!completed) {
+      failedDispatches.delete(key);
+      failedDispatches.add(key);
+      // Bound process memory. Evicting the oldest failure only lets a much
+      // older callback settle as interrupted later; it never replays it.
+      if (failedDispatches.size > FAILED_DISPATCH_MEMORY) {
+        failedDispatches.delete(failedDispatches.values().next().value!);
+      }
+    }
+  }
+}
+
+/** True while this process runs the callback, or after it saw it fail. */
+export function toolCallbackOwnedByThisProcess(runId: string, callId: string): boolean {
+  const key = callbackKey(runId, callId);
+  return runningDispatches.has(key) || failedDispatches.has(key);
+}
+
+/** The result the provider receives for a callback that a restart cut off. */
+export function interruptedToolCallbackOutcome(): CallbackOutcome {
+  return {
+    isError: true,
+    result: {
+      error: {
+        code: TOOL_CALLBACK_INTERRUPTED_CODE,
+        outcome: "unknown",
+        retryable: false,
+        message: "Paperclip restarted while this tool call was running, so its result was lost. "
+          + "It may or may not have taken effect. Check the current state before you repeat it.",
+      },
+    },
+  };
+}
+
+/**
+ * Resolve a saved callback that has no journaled result.
+ *
+ * The owner is gone when an earlier controller dispatched the callback (this
+ * controller restored it from that controller's journal) and this process is
+ * not running it. A callback this controller dispatched keeps the earlier
+ * rule: without a durable receipt it stays pending.
+ */
+export async function recoverToolCallback(input: {
+  runId: string;
+  call: ToolCallbackCall;
+  admission: SemanticInputAdmission | undefined;
+  recover: () => Promise<CallbackOutcome | null>;
+}): Promise<CallbackOutcome | null> {
+  const ownerGone = () => input.admission === "earlier_controller"
+    && !toolCallbackOwnedByThisProcess(input.runId, input.call.callId);
+  let outcome: CallbackOutcome | null;
+  try {
+    outcome = await input.recover();
+  } catch (error) {
+    // A receipt that proves an unknown outcome settles the same way once its
+    // owner is gone. Otherwise keep the earlier, fail-closed rejection.
+    if (!isSemanticToolOutcomeUnknown(error) || !ownerGone()) throw error;
+    return interruptedToolCallbackOutcome();
+  }
+  if (outcome) return outcome;
+  return ownerGone() ? interruptedToolCallbackOutcome() : null;
+}
+
+function isSemanticToolOutcomeUnknown(error: unknown): boolean {
+  // Match across the server's vendored and source-mode runtime boundaries.
+  return error instanceof Error && "code" in error && error.code === "semantic_tool_outcome_unknown";
+}
@@ -58,7 +58,15 @@ export {
  type RunnerProcessConnection,
  type RunnerProcessLaunchSpec,
  type RunnerProcessHandle,
+  type SemanticInputAdmission,
 } from "./control-plane/durable-prp-control-plane.js";
+export {
+  TOOL_CALLBACK_INTERRUPTED_CODE,
+  dispatchToolCallback,
+  interruptedToolCallbackOutcome,
+  recoverToolCallback,
+  toolCallbackOwnedByThisProcess,
+} from "./control-plane/orphaned-tool-callback.js";
 export type { DurableRecoveryIdentity } from "./control-plane/prp-transport-types.js";
 export * from "./drivers/codex/app-server-transport.js";
 export * from "./drivers/codex/codex-app-server-driver.js";
@@ -1356,7 +1356,8 @@ it.each([false, true])("uses the same API result envelope and fences handler pub
  try {
    await bundle.transport.request("thread/start", { cwd: tmpdir() });
    const call = { callId: "recovered-api", operationId: "call_api", input: { operationId: "POST /api/example" },
-      sourceEventId: "saved-event", sourceEventType: "semantic_tool.input", correlation: core.store.state.identity };
+      sourceEventId: "saved-event", sourceEventType: "semantic_tool.input", correlation: core.store.state.identity,
+      admission: "earlier_controller" as const };
    if (changedOwner) {
      owner = core.store.state.identity.runId;
      const pending = callbacks.onSemanticToolInput!(call);
@@ -7168,7 +7168,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {

    }

-  async #recoverSemanticToolInput(call: Parameters<NonNullable<DurablePrpControlPlaneOptions["onSemanticToolInput"]>>[0]) {
+  async #recoverSemanticToolInput(call: Parameters<NonNullable<DurablePrpControlPlaneOptions["recoverSemanticToolInput"]>>[0]) {
    if (this.#retainsSourceInputs() || (this.#authorityRunId() !== undefined && call.correlation.runId !== this.#authorityRunId())) return null;
    if (this.#sourceRecoveryIdentity) throw new NativeRunnerSourceReconciliationRequiredError("semantic");
    if (call.operationId === "paperclip_finish" || call.operationId === "paperclip_block") {
@@ -0,0 +1,171 @@
+import { spawn } from "node:child_process";
+import { randomUUID } from "node:crypto";
+import { eq } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
+import {
+  activityLog,
+  agents,
+  companies,
+  createDb,
+  heartbeatRuns,
+  issueRecoveryActions,
+  issues,
+  nativeRunFinalizations,
+  workspaceOperations,
+} from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { currentControllerBootId, currentProcessSpaceHash } from "../services/controller-process-identity.js";
+import { getExecutionBlocker } from "../services/execution-blocker.js";
+import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js";
+
+const support = await getEmbeddedPostgresTestSupport();
+const suite = support.supported ? describe : describe.skip;
+const KIND = "native_completed_turn_cleanup_hold";
+
+async function exitedPid(): Promise<number> {
+  const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" });
+  const pid = child.pid!;
+  await new Promise<void>(resolve => { child.once("exit", () => resolve()); child.kill("SIGKILL"); });
+  return pid;
+}
+
+/**
+ * The staging shape of a run that completed its provider turn after a restart
+ * orphaned one tool callback. Close could not prove the drain, so the run
+ * failed with `native_session_cleanup_quarantined`, and the system recorded two
+ * resolved-but-blocked no-replay holds for it:
+ * - the cleanup-quarantine hold (board owner, no workspace evidence), and
+ * - a continuation hold whose workspace recovery saved the files ("Saved work
+ *   recovered. Send a message to continue.", no Retry, no continuation).
+ * The task waited on a child task and stayed blocked after deploys.
+ */
+suite("completed-turn cleanup holds on the shared ownership path", () => {
+  let db!: ReturnType<typeof createDb>;
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+  let companyId = "", agentId = "";
+
+  beforeAll(async () => {
+    temporary = await startEmbeddedPostgresTestDatabase("paperclip-completed-turn-cleanup-");
+    db = createDb(temporary.connectionString);
+    companyId = randomUUID(); agentId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "CTC",
+      requireBoardApprovalForNewAgents: false });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle",
+      adapterType: "paperclip_runner", adapterConfig: {}, runtimeConfig: {}, permissions: {} });
+  }, 30_000);
+
+  afterEach(async () => {
+    await db.delete(activityLog);
+    await db.delete(issueRecoveryActions);
+    await db.delete(workspaceOperations);
+    await db.delete(nativeRunFinalizations);
+    await db.delete(heartbeatRuns);
+    await db.delete(issues);
+  });
+
+  afterAll(async () => { await temporary?.cleanup(); });
+
+  async function quarantinedRun(options: { controllerAlive?: boolean; turnCompleted?: boolean; filesSaved?: boolean } = {}) {
+    const issueId = randomUUID(), runId = randomUUID(), leaseId = randomUUID();
+    const finishedAt = new Date(Date.now() - 60_000);
+    await db.insert(issues).values({ id: issueId, companyId, title: "Set up visual reviews", status: "blocked",
+      assigneeAgentId: agentId, statusVersion: 6 });
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "automation", issueId,
+      nativeIssueId: issueId,
+      status: "failed", errorCode: "native_session_cleanup_quarantined", runtimeMode: "native",
+      error: "native_session_cleanup_quarantined: prior session cleanup requires operator recovery",
+      startedAt: new Date(finishedAt.getTime() - 16 * 60_000), finishedAt, nextEventSeq: 1, contextSnapshot: { issueId },
+      runnerProfileJson: { sessionCheckpoint: { identity: { runId }, activeTurnId: null,
+        terminal: { turnTerminalState: options.turnCompleted === false ? "interrupted" : "completed" } } } });
+    // The controller that closed the session and held the in-process gate.
+    await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "terminal_failure", attempt: 2,
+      leaseOwner: null, failureCode: "native_session_cleanup_quarantined", recoveryState: "blocked", controllerGeneration: 2,
+      ...(options.controllerAlive
+        ? { controllerBootId: currentControllerBootId(), controllerPid: process.pid, controllerHostHash: currentProcessSpaceHash(),
+          controllerProcessStartedAt: new Date(Date.now() - process.uptime() * 1_000) }
+        : { controllerBootId: randomUUID(), controllerPid: await exitedPid(), controllerHostHash: currentProcessSpaceHash(),
+          controllerProcessStartedAt: new Date(finishedAt.getTime() - 20 * 60_000) }) });
+    const noReplay = { runId, policy: "preserve_without_replay_v1", replay: "blocked",
+      recordedAt: finishedAt.toISOString(), actionOutcome: "unknown" };
+    const [quarantineHold] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
+      kind: "active_run_watchdog", status: "resolved", outcome: "blocked", ownerType: "board", returnOwnerAgentId: agentId,
+      cause: "native_session_cleanup_quarantined", fingerprint: randomUUID().replaceAll("-", ""), attemptCount: 1, maxAttempts: 3,
+      nextAction: "Workspace repair required. Verify safe staging or repair before continuing.", resolvedAt: finishedAt,
+      evidence: { runId, recoveryMode: "exact_checkpoint_resume", automaticRecovery: noReplay,
+        sourceFailureCode: "native_session_cleanup_quarantined", coordinatorAttempt: 2,
+        nativeFailureBlock: { runId, statusVersion: 6 }, recoveryDisposition: "native_session_cleanup_quarantined",
+        providerSessionEstablished: true } }).returning();
+    // The workspace recovery receipt: the saved files and their finalize step.
+    const [saved] = await db.insert(workspaceOperations).values({ companyId, issueId, heartbeatRunId: runId,
+      phase: "workspace_recovery", status: options.filesSaved === false ? "running" : "succeeded", metadata: {} }).returning();
+    const preservationOperationIds = [saved!.id];
+    const [filesHold] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
+      kind: "active_run_watchdog", status: "resolved", outcome: "blocked", ownerType: "system",
+      cause: "native_continuation_requires_reconciliation", fingerprint: `native-workspace:${runId}`, attemptCount: 1,
+      nextAction: "The original sandbox is retained for workspace repair.", resolvedAt: finishedAt,
+      evidence: { runId, automaticRecovery: noReplay,
+        systemRecovery: { schema: "paperclip.system-recovery.v1", policyVersion: 1, phase: "reconcile_effects", state: "complete",
+          sourceRunId: runId, attempts: 1, retriesUsed: 0, revision: 4, failure: null, intentId: null, deadlineAt: null,
+          nextAttemptAt: null, firstFailureAt: null, freshAttemptRunId: null, providerCooldownAt: null,
+          attemptId: randomUUID(), generationId: randomUUID(), preservationOperationIds },
+        workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId, stage: "saving_receipts", status: "recovered",
+          attempt: 1, retryable: true, requestIds: [], attemptLimit: 6 },
+        originalFailureCode: "native_session_cleanup_quarantined", continuationDelivery: "invalidated",
+        workspaceRestoreFailure: "restore_failed",
+        workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] },
+        historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId, leaseIds: [leaseId] },
+        previousWorkspaceRecovery: null } }).returning();
+    await db.update(workspaceOperations).set({ metadata: { recoveryActionId: filesHold!.id } })
+      .where(eq(workspaceOperations.id, saved!.id));
+    await db.insert(workspaceOperations).values({ companyId, issueId, heartbeatRunId: runId, phase: "workspace_finalize",
+      status: "succeeded", metadata: { owningService: "workspace_recovery", preservationOperationIds } });
+    return { issueId, runId, quarantineHoldId: quarantineHold!.id, filesHoldId: filesHold!.id };
+  }
+
+  const hold = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!;
+  const pass = async () => (await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: true }))
+    .kinds.find(entry => entry.kind === KIND);
+
+  it("settles both no-replay holds once the controller that held the cleanup gate is gone", async () => {
+    const f = await quarantinedRun();
+    // As on staging: the files hold blocks the task with no Retry and no continuation.
+    expect(await getExecutionBlocker(db, companyId, f.issueId)).toMatchObject({ recoveryActionId: f.filesHoldId,
+      canRetry: false, canContinue: false });
+
+    expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 });
+    for (const id of [f.quarantineHoldId, f.filesHoldId]) {
+      const settled = await hold(id);
+      expect(settled).toMatchObject({ status: "resolved", outcome: "restored", ownerType: "system" });
+      expect(settled.evidence.automaticRecovery).toBeUndefined();
+      expect(settled.evidence.completedTurnCleanupSettlement).toMatchObject({ runId: f.runId });
+    }
+    expect(await getExecutionBlocker(db, companyId, f.issueId)).toBeNull();
+    // Settling replays nothing: no new run, and the task keeps its own wait.
+    expect((await db.select().from(heartbeatRuns)).map(run => run.id)).toEqual([f.runId]);
+    expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "blocked" });
+    expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.recovery_action_resolved"))).toHaveLength(1);
+    // Bounded: nothing is left to list.
+    expect(await pass()).toMatchObject({ inProgress: 0 });
+  }, 60_000);
+
+  it("keeps the holds while the controller that may still hold the gate is alive", async () => {
+    const f = await quarantinedRun({ controllerAlive: true });
+    expect(await pass()).toMatchObject({ inProgress: 1, settled: 0 });
+    expect(await hold(f.quarantineHoldId)).toMatchObject({ outcome: "blocked" });
+    expect((await hold(f.filesHoldId)).evidence.automaticRecovery).toMatchObject({ replay: "blocked" });
+  }, 60_000);
+
+  it("keeps a hold whose files are not saved yet, and settles the run's other hold", async () => {
+    const f = await quarantinedRun({ filesSaved: false });
+    expect(await pass()).toMatchObject({ settled: 1 });
+    expect(await hold(f.quarantineHoldId)).toMatchObject({ outcome: "restored" });
+    expect(await hold(f.filesHoldId)).toMatchObject({ outcome: "blocked" });
+    expect(await getExecutionBlocker(db, companyId, f.issueId)).toMatchObject({ recoveryActionId: f.filesHoldId });
+  }, 60_000);
+
+  it("never settles the holds of a run whose turn did not complete", async () => {
+    const f = await quarantinedRun({ turnCompleted: false });
+    expect(await pass()).toMatchObject({ inProgress: 0 });
+    expect(await hold(f.quarantineHoldId)).toMatchObject({ outcome: "blocked" });
+  }, 60_000);
+});
@@ -151,7 +151,7 @@ describe("report-only ownership reconciliation", () => {
    expect(new Set(names).size).toBe(names.length);
    expect(names.length).toBeGreaterThanOrEqual(40);
    expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id))
-      .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "workspace_command_operation", "remote_snapshot_save", "execution_workspace_runtime_lease", "remote_workspace_unsaved_files", "remote_snapshot_ledger", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "legacy_interrupted_preparation_hold", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]);
+      .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "workspace_command_operation", "remote_snapshot_save", "execution_workspace_runtime_lease", "remote_workspace_unsaved_files", "remote_snapshot_ledger", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "legacy_interrupted_preparation_hold", "native_completed_turn_cleanup_hold", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]);
    expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true);
  });

@@ -1,4 +1,5 @@
 import { registerHeldPreparationResumer } from "./interrupted-preparation-hold-settlement.js";
+import { registerCompletedTurnCleanupResumer } from "./native-runtime/completed-turn-cleanup-hold-settlement.js";
 import { endStorageOutageWaitOn, reportStorageOutageWaitsTo, runInOwnStorageScope } from "../storage/storage-unavailable.js";
 import { createStorageWaitNotice } from "./heartbeat/storage-wait-notice.js";
 import { resolveProviderApiPlacement } from "./native-runtime/provider-api-placement.js";
@@ -1539,10 +1540,12 @@ export function heartbeatService(
  }
  // The interrupted-preparation ownership kind settles a hold, then replays the
  // preparation through this same resume path (bounded, uncharged retry).
-  registerHeldPreparationResumer(async ({ companyId, runId }) => {
+  const resumeSettledRun = async ({ companyId, runId }: { companyId: string; runId: string }) => {
    const run = await getRun(runId);
    if (run && run.companyId === companyId) await resumeRemoteStopComments(run);
-  });
+  };
+  registerHeldPreparationResumer(resumeSettledRun);
+  registerCompletedTurnCleanupResumer(resumeSettledRun);

  function cancelRunInternal(...args: Parameters<typeof runControl.cancelRunInternal>) {
    return runControl.cancelRunInternal(...args);
@@ -0,0 +1,118 @@
+import { and, eq, isNull, sql, type SQL } from "drizzle-orm";
+import { heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db";
+import { logActivity } from "../activity-log.js";
+import { logger } from "../../middleware/logger.js";
+import type { InProgressRecord, SettleOutcome } from "../recovery/ownership-registry.js";
+import type { OwnerProof } from "../owner-proof.js";
+import { OWNER_GONE_ATTESTATION_KEY } from "../recovery/owner-gone-attestation.js";
+import { completedWorkspaceRecoveryCondition } from "../workspace-recovery-receipt.js";
+import { NATIVE_SESSION_CLEANUP_QUARANTINED } from "./completed-turn-cleanup.js";
+
+/**
+ * `native_completed_turn_cleanup_hold`: no-replay holds left by a native run
+ * that completed its provider turn and was then refused by the cleanup gate
+ * (`isCompletedTurnCleanupQuarantine`).
+ *
+ * A server restart that orphaned a tool callback made close unable to prove the
+ * drain (fixed for new runs in the runner's orphaned-tool-callback rule). The
+ * run failed with `native_session_cleanup_quarantined`, and the system recorded
+ * its holds resolved-but-blocked with `preserve_without_replay_v1`: replay
+ * blocked because one action outcome was unknown. Nothing is left to replay:
+ * the turn completed and its work is saved. The holds still blocked the task,
+ * with no Retry and no continuation, so the task could not continue even when
+ * the work it waited on finished.
+ *
+ * The cleanup gate lives in the controller process that closed the session.
+ * Once `proveOwnerGone` proves that controller gone, the gate is gone too, so
+ * this kind settles the run's no-replay holds: it removes their replay block
+ * and starts no turn. A hold that still protects files (a workspace restore not
+ * yet verified as recovered) is left for workspace recovery. Operator decisions
+ * (an assigned person, an owner-gone attestation, a reconciliation or explicit
+ * continuation) are never reopened. Every rule reads fields that earlier
+ * releases already saved, so their holds settle without a migration.
+ */
+
+const COMPLETED_TURN_CHECKPOINT = sql`run.runner_profile_json->'sessionCheckpoint'->'identity'->>'runId' = run.id::text
+  and run.runner_profile_json->'sessionCheckpoint'->>'activeTurnId' is null
+  and run.runner_profile_json->'sessionCheckpoint'->'terminal'->>'turnTerminalState' = 'completed'`;
+
+/** A no-replay hold of `runId` that this kind may settle. */
+function settleableHoldCondition(runId: SQL | string): SQL {
+  return and(
+    sql`${issueRecoveryActions.evidence}->>'runId' = ${runId}::text`,
+    eq(issueRecoveryActions.status, "resolved"),
+    eq(issueRecoveryActions.outcome, "blocked"),
+    isNull(issueRecoveryActions.ownerUserId),
+    isNull(issueRecoveryActions.ownerAgentId),
+    sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'policy' = 'preserve_without_replay_v1'`,
+    sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`,
+    sql`not (${issueRecoveryActions.evidence} ?| array['executionReconciliation', ${OWNER_GONE_ATTESTATION_KEY}, 'explicitUserContinuation'])`,
+    sql`(not (${issueRecoveryActions.evidence} ?| array['workspaceRestoreFailure', 'workspaceRestoreRecovery'])
+      or ${completedWorkspaceRecoveryCondition(runId)})`,
+  )!;
+}
+
+/** Coordinators of completed-turn cleanup quarantines that still have a
+ * settleable no-replay hold. */
+export function completedTurnCleanupHoldWhere(): SQL {
+  return and(
+    isNull(nativeRunFinalizations.leaseOwner),
+    sql`exists (select 1 from heartbeat_runs run where run.company_id = ${nativeRunFinalizations.companyId}
+      and run.id = ${nativeRunFinalizations.runId} and run.runtime_mode = 'native' and run.status = 'failed'
+      and run.error_code = ${NATIVE_SESSION_CLEANUP_QUARANTINED} and ${COMPLETED_TURN_CHECKPOINT})`,
+    sql`exists (select 1 from ${issueRecoveryActions} where ${issueRecoveryActions.companyId} = ${nativeRunFinalizations.companyId}
+      and ${settleableHoldCondition(sql`${nativeRunFinalizations.runId}`)})`,
+  )!;
+}
+
+/** Continues the task through the ordinary resume path
+ * (`resumeRemoteStopComments`), which still applies every task gate. The
+ * heartbeat service registers it. */
+type SettledHoldResumer = (input: { companyId: string; runId: string }) => Promise<void>;
+let settledHoldResumer: SettledHoldResumer | null = null;
+
+export function registerCompletedTurnCleanupResumer(resumer: SettledHoldResumer): void {
+  settledHoldResumer = resumer;
+}
+
+/** Called by the shared reconciler only after `proveOwnerGone` proved the
+ * run's recorded controller gone. Re-checks the whole condition under the
+ * coordinator row lock, so a coordinator taken over since the proof, or a hold
+ * changed since, is not settled. */
+export async function settleCompletedTurnCleanupHolds(db: Db, record: InProgressRecord,
+  context: { now: Date; proof: OwnerProof }): Promise<SettleOutcome> {
+  if (!record.companyId) return { settled: false, reason: "run_has_no_company" };
+  const companyId = record.companyId;
+  const now = context.now;
+  const settled = await db.transaction(async tx => {
+    const [coordinator] = await tx.select({ runId: nativeRunFinalizations.runId, issueId: nativeRunFinalizations.issueId })
+      .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId),
+        eq(nativeRunFinalizations.runId, record.recordId), completedTurnCleanupHoldWhere())).for("update");
+    if (!coordinator) return null;
+    const [run] = await tx.select({ agentId: heartbeatRuns.agentId }).from(heartbeatRuns)
+      .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, coordinator.runId)));
+    const holds = await tx.update(issueRecoveryActions).set({
+      outcome: "restored", ownerType: "system", updatedAt: now,
+      resolutionNote: "The run finished its turn and saved its work; its session cleanup no longer blocks the task.",
+      evidence: sql`(${issueRecoveryActions.evidence} - 'automaticRecovery') || ${JSON.stringify({
+        completedTurnCleanupSettlement: { runId: coordinator.runId, settledAt: now.toISOString(), ownerProof: context.proof.reason },
+      })}::jsonb`,
+    }).where(and(eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, coordinator.issueId),
+      settleableHoldCondition(coordinator.runId))).returning({ id: issueRecoveryActions.id });
+    if (!holds.length) return null;
+    await logActivity(tx as unknown as Db, { companyId, actorType: "system", actorId: "system",
+      action: "issue.recovery_action_resolved", entityType: "issue", entityId: coordinator.issueId,
+      agentId: run?.agentId ?? null, runId: coordinator.runId,
+      details: { reason: "native_completed_turn_cleanup_settled", recoveryActionIds: holds.map(hold => hold.id),
+        ownerProof: context.proof.reason },
+    });
+    return coordinator.runId;
+  });
+  if (!settled) return { settled: false, reason: "hold_changed_since_proof" };
+  // Settling starts no turn. Saved input and the task's own wait continue
+  // through normal admission, which still checks every other gate.
+  await settledHoldResumer?.({ companyId, runId: settled }).catch(error => {
+    logger.warn({ err: error, runId: settled }, "settled completed-turn cleanup hold could not resume saved input");
+  });
+  return { settled: true, reason: context.proof.reason };
+}
@@ -251,6 +251,7 @@ import { createNativeCredentialUpdateOwner, appendNativeCredentialUpdate } from
 import { canonicalNativeJson, nativeSha256 } from "./canonical.js";
 import { lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js";
 import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js";
+import { dispatchToolCallback, recoverToolCallback, type SemanticInputAdmission } from "../../vendor/paperclip-runner/index.js";
 import { createAssignedMcpTools, getAssignedMcpGateway } from "./assigned-mcp-tools.js";
 import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js";
 import { NativeChatAttachmentReadScope } from "./chat-attachment-read.js";
@@ -2014,18 +2015,21 @@ class SessionAuthorityEpoch {
    return this.#authority.definitions();
  }

-  async recover(call: Parameters<PaperclipRunnerToolAuthority["recover"]>[0]) {
+  /** Replay a durable receipt, or settle a callback a restart orphaned
+   * (orphaned-tool-callback.ts). Never executes the tool again. */
+  async recover(call: Parameters<PaperclipRunnerToolAuthority["recover"]>[0], admission?: SemanticInputAdmission) {
    this.#assertCurrent();
-    const result = await this.#authority.recover(call);
+    const result = await recoverToolCallback({ runId: this.runId, call, admission,
+      recover: () => this.#authority.recover(call) });
    this.#assertCurrent();
    return result;
  }

  async execute(call: Parameters<PaperclipRunnerToolAuthority["execute"]>[0]) {
    this.#assertCurrent();
-    return this.toolTrace
+    return dispatchToolCallback(this.runId, call.callId, async () => this.toolTrace
      ? await this.toolTrace.execute(call, () => this.#authority.execute(call))
-      : await this.#authority.execute(call);
+      : await this.#authority.execute(call));
  }
 }

@@ -15054,7 +15058,7 @@ async function createRunnerdBackendWithinSessionClaim(
        recoverSemanticToolInput: input.controllerPersistence ? call => {
          const current = sessionToolAuthorityEpochs.get(sessionScopeId);
          if (!current || current.runId !== call.correlation.runId) throw new Error("native_tool_authority_epoch_revoked");
-          return current.recover({ tool: call.operationId, callId: call.callId, arguments: call.input });
+          return current.recover({ tool: call.operationId, callId: call.callId, arguments: call.input }, call.admission);
        } : undefined,
        ...(provider.kind === "codex" ? { disableCodexWebSearch: "webSearch" in provider && provider.webSearch === "disabled" } : {}),
        ...(provider.kind === "acpx" ? { acpxAgent: provider.agent,
@@ -15103,8 +15107,8 @@ async function createRunnerdBackendWithinSessionClaim(
        stateDirectory: root, runnerBinary: remoteDot?.runnerBinary ?? resolvePaperclipRunnerBinary(),
        runnerArtifact: input.controllerPersistence?.runnerArtifact,
        runnerReconnectGraceMs: remoteDot ? REMOTE_RUNNER_RECONNECT_GRACE_MS : undefined,
-        recoverSemanticToolInput: (call: { operationId: string; callId: string; input: unknown }) =>
-          authorityEpoch.recover({ tool: call.operationId, callId: call.callId, arguments: call.input }),
+        recoverSemanticToolInput: (call: { operationId: string; callId: string; input: unknown; admission: SemanticInputAdmission }) =>
+          authorityEpoch.recover({ tool: call.operationId, callId: call.callId, arguments: call.input }, call.admission),
        ...(remoteDot ?? {}),
        identity: { runnerInstanceId: effectiveRunnerInstanceId, environmentLeaseId: effectiveEnvironmentLeaseId,
          runId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution),
@@ -16735,7 +16739,7 @@ async function createRunnerdBackendWithinSessionClaim(
        recoverSemanticToolInput: call => {
          const current = sessionToolAuthorityEpochs.get(sessionScopeId);
          if (!current || current.runId !== call.correlation.runId) throw new Error("native_tool_authority_epoch_revoked");
-          return current.recover({ tool: call.operationId, callId: call.callId, arguments: call.input });
+          return current.recover({ tool: call.operationId, callId: call.callId, arguments: call.input }, call.admission);
        },
        codexCommand: remoteCodexBinary ?? undefined,
        sourceCodexHome: remoteTarget
@@ -50,6 +50,7 @@ import { settleIssueLock } from "./issue-lock-settlement.js";
 import { settleAgentLifecycleLease } from "./agent-lifecycle-lease-settlement.js";
 import { legacyControllerOwnerStamp, settleLegacyControllerLease } from "../legacy-controller-lease.js";
 import { heldInterruptedPreparationWhere, settleHeldInterruptedPreparation } from "../interrupted-preparation-hold-settlement.js";
+import { completedTurnCleanupHoldWhere, settleCompletedTurnCleanupHolds } from "../native-runtime/completed-turn-cleanup-hold-settlement.js";
 import {
  nativeLeaseHolderStamp,
  readNativeLeaseHolder,
@@ -1077,6 +1078,15 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [
      }),
    settle: settleHeldInterruptedPreparation,
  },
+  {
+    // No-replay holds of a run that completed its turn and was then refused by
+    // the in-process cleanup gate. Its owner is the controller that closed the
+    // session (the coordinator's stamp); once that process is proven gone, so
+    // is its gate, and the holds settle without replaying anything.
+    ...nrfKind("native_completed_turn_cleanup_hold", "Completed-turn cleanup hold", "ownership",
+      completedTurnCleanupHoldWhere()),
+    settle: settleCompletedTurnCleanupHolds,
+  },
  runKind("detached_process_marker", "Detached-process marker", "ownership",
    and(eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.errorCode, "process_detached")),
    row => row.processPid ? { kind: "process", stamp: runProcessStamp(row, "detached_process_marker") } : { kind: "none" },
@@ -117,6 +117,9 @@ export const NativeSessionProtocolIntegrityError =
  runner.NativeSessionProtocolIntegrityError;
 export const NativeEnvironmentProbeCleanupError = runner.NativeEnvironmentProbeCleanupError;
 export const SemanticToolOutcomeUnknownError = runner.SemanticToolOutcomeUnknownError;
+export const dispatchToolCallback = runner.dispatchToolCallback;
+export const recoverToolCallback = runner.recoverToolCallback;
+export type SemanticInputAdmission = import("@paperclipai/paperclip-runner").SemanticInputAdmission;
 export const PaperclipSemanticDispatcher = runner.PaperclipSemanticDispatcher;
 export const CAPABILITY_SEMANTIC_TOOL_CATALOG =
  runner.CAPABILITY_SEMANTIC_TOOL_CATALOG;