Integrate runner recovery through work-folder PR 13006

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/services/acknowledged-native-stop.test.ts

@@ -0,0 +1,17 @@
+import { expect, it } from "vitest";
+import { isAcknowledgedNativeStop } from "./acknowledged-native-stop.js";
+const run = { id: "run", companyId: "company", nativeIssueId: "issue", status: "cancelled", resultJson: {
+  cancelledByActorType: "user", cancelledByUserId: "board", nativeCancellation: {
+    schema: "paperclip.native-cancellation.v1", runId: "run", companyId: "company", issueId: "issue",
+    scope: "run", reasonCode: "cancellation_run_only", dispatchState: "acknowledged", dispatched: true,
+    intentAuditId: "intent", acknowledgementAuditId: "ack",
+  },
+} };
+it("recognizes the native receipt used by the Stop button", () => {
+  expect(isAcknowledgedNativeStop(run)).toBe(true);
+});
+it.each([{ runId: "other" }, { companyId: "other" }, { issueId: "other" }, { dispatchState: "pending" },
+  { scope: "subtree" }, { acknowledgementAuditId: undefined }])("refuses unrelated or incomplete receipts %j", change => {
+  expect(isAcknowledgedNativeStop({ ...run, resultJson: { ...run.resultJson,
+    nativeCancellation: { ...run.resultJson.nativeCancellation, ...change } } })).toBe(false);
+});

server/src/services/acknowledged-native-stop.ts

@@ -0,0 +1,19 @@
+/** A server-recorded run-only Stop must not manufacture a recovery incident. */
+export function hasAcknowledgedNativeStopIntent(run: {
+  id: string; companyId: string; status: string; nativeIssueId: string | null;
+  resultJson: Record<string, unknown> | null;
+}): boolean {
+  const result = run.resultJson;
+  const intent = result?.nativeCancellation as Record<string, unknown> | undefined;
+  return result?.cancelledByActorType === "user" &&
+    typeof result.cancelledByUserId === "string" && Boolean(result.cancelledByUserId) &&
+    intent?.schema === "paperclip.native-cancellation.v1" && intent.runId === run.id &&
+    intent.companyId === run.companyId && intent.issueId === run.nativeIssueId &&
+    intent.scope === "run" && intent.reasonCode === "cancellation_run_only" &&
+    intent.dispatchState === "acknowledged" && intent.dispatched === true &&
+    typeof intent.intentAuditId === "string" && typeof intent.acknowledgementAuditId === "string";
+}
+
+export function isAcknowledgedNativeStop(run: Parameters<typeof hasAcknowledgedNativeStopIntent>[0]): boolean {
+  return run.status === "cancelled" && hasAcknowledgedNativeStopIntent(run);
+}

server/src/services/ai-connection-runtime.ts

@@ -164,15 +164,24 @@ done`,
 async function acquireCredentialLease(db: Db, grantId: string) {
   const client = await db.$client.reserve();
   try {
+    // A reserved client pins our connection to PgBouncer, not its backend.
+    // Keep the lease in one transaction so transaction-pooling deployments
+    // cannot acquire and release it on different PostgreSQL sessions.
+    await client`begin`;
+    await client`set local idle_in_transaction_session_timeout = 0`;
     const [result] =
-      await client`select pg_try_advisory_lock(hashtextextended(${`ai-runtime:${grantId}`}, 0)) as acquired`;
+      await client`select pg_try_advisory_xact_lock(hashtextextended(${`ai-runtime:${grantId}`}, 0)) as acquired`;
     if (!result.acquired)
       throw unprocessable(
         "This subscription is in use. Retry when its current execution finishes.",
         { code: "ai_connection_busy" },
       );
   } catch (error) {
-    client.release();
+    try {
+      await client`rollback`;
+    } finally {
+      client.release();
+    }
     throw error;
   }
   let released = false;
@@ -180,7 +189,7 @@ async function acquireCredentialLease(db: Db, grantId: string) {
     if (released) return;
     released = true;
     try {
-      await client`select pg_advisory_unlock(hashtextextended(${`ai-runtime:${grantId}`}, 0))`;
+      await client`rollback`;
     } finally {
       client.release();
     }

server/src/services/execution-recovery-resolution.ts

@@ -473,8 +473,9 @@ export async function settleUnrecoverableExecutions(
         const note = current
           ? "Automatic recovery stopped. Recorded work is preserved; actions with unverified outcomes will not be repeated."
           : "Recovery closed because the task's owner, execution, or status changed. No work was replayed.";
-        if (current)
-          await tx
+        let nativeFailureBlock = action.evidence.nativeFailureBlock;
+        if (current) {
+          const [projected] = await tx
             .update(issues)
             .set({
               status: "blocked",
@@ -482,7 +483,13 @@ export async function settleUnrecoverableExecutions(
               checkoutRunId: null,
               updatedAt: now,
             })
-            .where(eq(issues.id, task.id));
+            .where(eq(issues.id, task.id)).returning();
+          // Only a transition owned by this failure grants a recovery receipt.
+          // An already-blocked task may have a separate human/dependency hold.
+          if (task.status !== "blocked" && run.runtimeMode === "native") {
+            nativeFailureBlock = { runId: run.id, statusVersion: projected!.statusVersion };
+          }
+        }
         await tx
           .update(issueRecoveryActions)
           .set({
@@ -496,6 +503,7 @@ export async function settleUnrecoverableExecutions(
             monitorPolicy: null,
             evidence: {
               ...action.evidence,
+              ...(nativeFailureBlock ? { nativeFailureBlock } : {}),
               automaticRecovery: {
                 policy: "preserve_without_replay_v1",
                 runId: run.id,

server/src/services/heartbeat.ts

@@ -1,5 +1,6 @@
 import { AGENT_CHAT_DIRECTIVE, conversationReplay, isConversation, isConversationExecutionWake, isWaitingConversation, prepareConversationTurn, settleConversationTurn } from "./agent-conversations.js";
 import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js";
+import { hasAcknowledgedNativeStopIntent } from "./acknowledged-native-stop.js";
 import { legacyControllerBootId, legacyControllerClaim, renewLegacyControllerLease, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js";
 import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
 import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
@@ -195,6 +196,7 @@ import {
   cancelNativeSession,
   claimNativeRestartRecoveries,
   closeWarmNativeSessionsForEnvironment,
+  closeIdleWarmNativeSessionsForRestart,
   currentNativeControllerIdentity,
   dispatchNativeSessionResumptions,
   detachNativeSessionsForRestart,
@@ -210,6 +212,7 @@ import {
   materializeNativeInteractionResponses,
   nativeCompletionRequestsForComments,
   NativeCancellationPendingRecoveryError,
+  NativeControllerDetachedForRestartError,
   nativeToolContractFingerprintForTarget,
   prepareNativeSessionBootstrapPersistence,
   prepareNativeWorkspaceSync,
@@ -10147,6 +10150,7 @@ export function heartbeatService(
     if (!agent || agent.companyId !== companyId || agent.adapterType === "paperclip_runner") return;
     const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
       eq(heartbeatRuns.companyId, companyId),
+      eq(heartbeatRuns.agentId, wake.agentId),
       sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
       inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
     )).limit(1);
@@ -10168,6 +10172,7 @@ export function heartbeatService(
             !current || !queuedCommentIdsFromWakePayload(current.payload).length) return null;
         const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
           eq(heartbeatRuns.companyId, companyId),
+          eq(heartbeatRuns.agentId, wake.agentId),
           sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
           inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
         )).limit(1);
@@ -12699,6 +12704,10 @@ export function heartbeatService(
       else if (issueStatus === "cancelled") terminalStatus = "cancelled";
     }
 
+    // Teardown can beat the cancellation finalizer. Preserve the acknowledged
+    // user intent instead of reporting an infrastructure interruption.
+    if (hasAcknowledgedNativeStopIntent(run)) terminalStatus = "cancelled";
+
     const message = `run terminalized on environment lease release: heartbeat_runs.status was still ${run.status} at teardown`;
     // Match both "running" and "queued". A queued run has released its lease but
     // never reached "running", so a running-only update would miss it and leave
@@ -14175,6 +14184,10 @@ export function heartbeatService(
     now = new Date(),
   ) {
     shutdownInProgress = true;
+    const idleSessions = await closeIdleWarmNativeSessionsForRestart();
+    if (idleSessions.failed > 0) {
+      logger.warn({ idleSessions }, "idle native sessions could not checkpoint before controller shutdown");
+    }
     let intent: Awaited<ReturnType<typeof readHotRestartIntent>>;
     try {
       intent = await readHotRestartIntent();
@@ -14804,6 +14817,11 @@ export function heartbeatService(
         run.runtimeMode === "native" &&
         agent.adapterType === "paperclip_runner"
       ) {
+        // A graceful shutdown relinquishes controller authority just like a
+        // hot restart. Leaving the old event consumer attached lets its
+        // finalizer interrupt/suspend Claude while the next server is adopting
+        // the same turn.
+        await detachNativeSessionsForRestart([run.id]);
         const recoveryHistoryEntry = JSON.stringify({
           at: now.toISOString(),
           restartKind: "graceful",
@@ -21244,7 +21262,14 @@ export function heartbeatService(
               companyId: agent.companyId, issueId, runId: run.id, agentId: agent.id, workspaceId: workspace.id,
             });
           } else {
-            await issuesSvc.update(issueId, nextIssuePatch);
+            await issuesSvc.update(
+              issueId,
+              { ...nextIssuePatch, companyGuard: agent.companyId },
+              db,
+              undefined,
+              undefined,
+              { bindRuntimeSharedWorkspace: warmReusableExecutionWorkspace && workspace.mode === "shared_workspace" },
+            );
           }
           issueExecutionWorkspaceIdForRun = workspace.id;
           issueProjectWorkspaceIdForRun =
@@ -23845,6 +23870,12 @@ export function heartbeatService(
             }
           }
         } catch (adapterErr) {
+          if (adapterErr instanceof NativeControllerDetachedForRestartError) {
+            // Preserve the provider and its run for the new controller. This
+            // also keeps generic teardown from terminalizing/releasing its lease.
+            nativeSessionResumeScheduled = true;
+            throw adapterErr;
+          }
           if (adapterErr instanceof NativeRunnerOwnershipUnverifiedError) {
             nativeOwnershipHeld = true;
             throw adapterErr;
@@ -24701,6 +24732,10 @@ export function heartbeatService(
           wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run),
         });
       } catch (err) {
+        if (err instanceof NativeControllerDetachedForRestartError) {
+          nativeSessionResumeScheduled = true;
+          return;
+        }
         if (err instanceof NativeRunnerOwnershipUnverifiedError) {
           nativeOwnershipHeld = true;
           const heldRun = await getRun(run.id);

server/src/services/issues.ts

@@ -10473,6 +10473,7 @@ export function issueService(db: Db) {
       dbOrTx: any = db,
       postCommitActivityPublications?: ActivityPublication[],
       postCommitActions?: IssuePostCommitAction[],
+      options: { bindRuntimeSharedWorkspace?: boolean } = {},
     ) => {
       const ownedActivityPublications: ActivityPublication[] = [];
       const activityPublications =
@@ -10543,7 +10544,30 @@ export function issueService(db: Db) {
       const isolatedWorkspacesEnabled = (
         await instanceSettings.getExperimental()
       ).enableIsolatedWorkspaces;
-      if (!isolatedWorkspacesEnabled) {
+      if (options.bindRuntimeSharedWorkspace) {
+        const workspaceId = issueData.executionWorkspaceId ?? existing.executionWorkspaceId;
+        if (!workspaceId) {
+          throw unprocessable("Runtime workspace binding requires an existing shared workspace");
+        }
+        const [workspace] = await dbOrTx
+          .select({ mode: executionWorkspaces.mode })
+          .from(executionWorkspaces)
+          .where(and(
+            eq(executionWorkspaces.id, workspaceId),
+            eq(executionWorkspaces.companyId, existing.companyId),
+          ));
+        if (
+          workspace?.mode !== "shared_workspace" ||
+          (issueData.executionWorkspacePreference ?? existing.executionWorkspacePreference) !== "reuse_existing" ||
+          (issueData.executionWorkspaceSettings ?? existing.executionWorkspaceSettings)?.mode !== "shared_workspace"
+        ) {
+          throw unprocessable("Runtime workspace binding requires an existing shared workspace");
+        }
+      }
+      // Warm sandbox continuity is runtime bookkeeping, independent of the
+      // opt-in UI for creating isolated worktrees. Public updates still obey
+      // the feature gate; only the internal shared-workspace binding bypasses it.
+      if (!isolatedWorkspacesEnabled && !options.bindRuntimeSharedWorkspace) {
         delete issueData.executionWorkspaceId;
         delete issueData.executionWorkspacePreference;
         delete issueData.executionWorkspaceSettings;
@@ -10795,6 +10819,15 @@ export function issueService(db: Db) {
           projectGoalId: nextProjectGoalId,
           defaultGoalId: defaultCompanyGoal?.id ?? null,
         });
+        // Reasserting Blocked or changing its blockers is a fresh decision even
+        // when the status string stays the same. Invalidate recovery's prior
+        // status receipt without treating comment recency as blocking intent.
+        if (receiptExisting.status === "blocked" &&
+            (issueData.status === "blocked" ||
+              (issueData.status === undefined &&
+                (blockedByIssueIds !== undefined || issueData.unblockDescriptor !== undefined)))) {
+          patch.statusVersion = sql`${issues.statusVersion} + 1` as unknown as number;
+        }
         const updated = await tx
           .update(issues)
           .set(patch)

server/src/services/native-local-process-stop.ts

@@ -58,3 +58,18 @@ export async function hasNativeLocalProcessStop(db: Db, companyId: string, runId
     .limit(1);
   return event?.eventType === LOCAL_PROCESS_STOPPED;
 }
+
+/** Recover the exact stopped identity after the mutable run fields were cleared. */
+export async function readNativeLocalProcessStop(db: Db, companyId: string, runId: string) {
+  const [event] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload })
+    .from(heartbeatRunEvents)
+    .where(and(eq(heartbeatRunEvents.companyId, companyId), eq(heartbeatRunEvents.runId, runId),
+      isNull(heartbeatRunEvents.sourceEventId),
+      inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, LOCAL_PROCESS_STOPPED])))
+    .orderBy(desc(heartbeatRunEvents.seq)).limit(1);
+  const pid = event?.payload?.processPid;
+  const group = event?.payload?.processGroupId;
+  if (event?.eventType !== LOCAL_PROCESS_STOPPED || typeof pid !== "number" ||
+      !Number.isSafeInteger(pid) || pid <= 1 || group !== pid || !absent(pid) || !absent(-pid)) return null;
+  return { processPid: pid, processGroupId: pid };
+}

server/src/services/native-runtime/native-board-response-wait.ts

@@ -74,7 +74,7 @@ export async function readNativeBoardResponseWaitOrigin(
         eq(agentWakeupRequests.agentId, binding.agentId),
         eq(agentWakeupRequests.runId, binding.runId),
         eq(agentWakeupRequests.source, "automation"),
-        eq(agentWakeupRequests.reason, "issue_commented"),
+        inArray(agentWakeupRequests.reason, ["issue_commented", "issue_reopened_via_comment"]),
         eq(agentWakeupRequests.requestedByActorType, "user"),
       ),
     )
@@ -203,7 +203,7 @@ export async function readNativeBoardResponseWaitSource(
         eq(agentWakeupRequests.agentId, binding.agentId),
         eq(agentWakeupRequests.runId, binding.runId),
         eq(agentWakeupRequests.source, "automation"),
-        eq(agentWakeupRequests.reason, "issue_commented"),
+        inArray(agentWakeupRequests.reason, ["issue_commented", "issue_reopened_via_comment"]),
         eq(agentWakeupRequests.requestedByActorType, "user"),
       ),
     )

server/src/services/native-runtime/native-execution-input.test.ts

@@ -451,4 +451,26 @@ describe("native execution input external-chat framing", () => {
       }
     },
   );
+  it("gives ordinary Board tasks the same durable-question guidance as external chat", () => {
+    const input = buildNativeExecutionInput({
+      companyId: "10000000-0000-4000-8000-000000000001",
+      runId: "50000000-0000-4000-8000-000000000005",
+      agentId: "30000000-0000-4000-8000-000000000003",
+      issue: { id: "20000000-0000-4000-8000-000000000002", identifier: "QA-1", title: "Welcome", description: null, workMode: "standard" },
+      taskPrompt: "Ask whether the welcome should sound warm or formal before writing it.",
+      workspace: { id: "40000000-0000-4000-8000-000000000004", cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null },
+      normalizedSessionId: null,
+      provider: "codex",
+      completionContract: {
+        id: "70000000-0000-4000-8000-000000000007", sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1",
+        contract: { revision: "1", objective: "Write a welcome after the user's answer", criteria: [{ id: "objective", requirement: "Use the selected tone" }] },
+      },
+      runtimeContext: nativeRuntimeContextFixture(),
+    });
+    expect(input.task.prompt).toContain('interactionKind="questions"');
+    expect(input.task.prompt).toContain('continuationPolicy="wake_assignee"');
+    expect(input.task.prompt).toContain("Create the actual question before yielding");
+    expect(input.task.prompt).toContain("Wait for its real answer");
+  });
+
 });

server/src/services/native-runtime/native-execution-input.ts

@@ -19,13 +19,14 @@ import {
   renderPaperclipWakePrompt,
 } from "@paperclipai/adapter-utils/server-utils";
 
-const NATIVE_EXTERNAL_CHAT_QUESTION_GUIDANCE = [
-  "## Native external-chat questions",
+const NATIVE_QUESTION_GUIDANCE = [
+  "## Questions that need a user response",
   "A request for clickable choices, buttons, or a decision needed before continuing is not a self-contained text answer. The zero-API-call shortcut does not prohibit the structured question tool.",
   'Use the available request_human_input tool with interactionKind="questions", continuationPolicy="wake_assignee", a title, prompt, and a stable idempotencyKey. Put the actual requested choices in payload.questions: each question needs an id, prompt, selectionMode="single", and options with stable id and label fields. Reuse the same key if that creation call must be retried.',
   "Paperclip renders the supported question controls and authenticates the answer. Never fabricate answer URLs, query-string choice links, callback tokens, or fake Markdown buttons. Do not manually post a duplicate question card or use call_api as a substitute.",
   "For one question at a time, read the current request and authoritative prior answers, then create only the next unanswered question. Wait for its real answer before asking another; do not infer a selection or answer your own interaction. Keep completion and disposition truthful while waiting, and preserve existing review or approval gates.",
   "If the tool is unavailable or creation fails, report that actual limitation plainly; do not pretend interactive controls were created.",
+  "A completion summary saying that you asked a question does not create a question. Create the actual question before yielding; never claim to be waiting for a response to an interaction you have not created.",
 ].join("\n");
 
 const NATIVE_GITHUB_ATTACHMENT_RECOVERY_GUIDANCE = [
@@ -165,7 +166,7 @@ export function buildNativeExecutionInput(input: {
     isPaperclipExternalChatQuestionResponseTurn(wakePayload);
   const taskPrompt = [
     wakePrompt,
-    externalChatTurn ? NATIVE_EXTERNAL_CHAT_QUESTION_GUIDANCE : "",
+    NATIVE_QUESTION_GUIDANCE,
     externalChatTurn && wake?.externalChatProvider === "github"
       ? NATIVE_GITHUB_ATTACHMENT_RECOVERY_GUIDANCE
       : "",

server/src/services/native-runtime/native-finalization-reconciler.ts

@@ -465,7 +465,7 @@ export async function claimNativeSessionResumptions(input: {
         terminalRunToEmit = updatedRun ?? null;
         await issueService(tx as unknown as Db).update(
           row.coordinator.issueId,
-          { status: "in_review" },
+          { status: "blocked" },
           tx,
         );
         await issueRecoveryActionService(tx as unknown as Db).upsertSourceScoped({

server/src/services/native-runtime/native-provider-capacity.integration.test.ts

@@ -288,7 +288,7 @@ describe("native provider capacity failure persistence", () => {
         .select()
         .from(issues)
         .where(eq(issues.id, issueId));
-      expect(waitingIssue.status).toBe("in_review");
+      expect(waitingIssue.status).toBe("blocked");
       const events = await db
         .select()
         .from(heartbeatRunEvents)

server/src/services/native-runtime/native-safe-replacement.test.ts

@@ -1,5 +1,6 @@
 import { createRunDispatch, deriveCommentId } from "../../modules/run-dispatch/index.js";
 import { buildExecutionContinuation } from "../execution-continuation.js";
+import { issueService } from "../issues.js";
 import { activityService } from "../activity.js";
 import { buildPaperclipWakePayload, heartbeatService } from "../heartbeat.js";
 import { legacyExecutionNeedsReconciliation, terminalizeLegacyExecution } from "../legacy-execution-recovery.js";
@@ -12,9 +13,10 @@ import {
   deliverReconciledExecutions,
 } from "../execution-recovery-resolution.js";
 import { randomUUID } from "node:crypto";
+import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
 import { tmpdir } from "node:os";
 import { and, eq, inArray } from "drizzle-orm";
-import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
 import {
   agents,
   companies,
@@ -107,6 +109,77 @@ const support = externalDatabaseUrl
       });
       return { companyId, agentId, issueId, runId };
     }
+    it.each(["unproven", "changed", "verified"] as const)("requires stopped-session proof through commit (%s)", async (mode) => {
+      const source = await seed(2);
+      await db.update(nativeRunFinalizations).set({ failureCode: "native_session_cleanup_quarantined" }).where(eq(nativeRunFinalizations.runId, source.runId));
+      const [projected] = await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, source.issueId)).returning();
+      await db.insert(issueRecoveryActions).values({ companyId: source.companyId, sourceIssueId: source.issueId,
+        kind: "active_run_watchdog", cause: "native_session_cleanup_quarantined", fingerprint: source.runId,
+        ownerType: "board", returnOwnerAgentId: source.agentId, status: "resolved", outcome: "blocked",
+        evidence: { runId: source.runId, nativeFailureBlock: { runId: source.runId, statusVersion: projected!.statusVersion }, automaticRecovery: { replay: "blocked" } }, nextAction: "Automatic recovery stopped.",
+      });
+      const retire = vi.fn(() => mode === "verified");
+      const verifyStoppedSession = vi.fn(async (run: typeof heartbeatRuns.$inferSelect) =>
+        run.id === source.runId && mode !== "unproven" ? { evidence: { completedTaskControlCallIds: ["completion"] }, retire } : null);
+      await appendHeartbeatRunEvent(db, { companyId: source.companyId, runId: source.runId, agentId: source.agentId,
+        eventType: "tool.execution.started", stream: "system",
+        payload: { name: "paperclip_finish", executionId: "completion", transport: "process" },
+      });
+      await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession });
+      await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession });
+      const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId));
+      expect(successors).toHaveLength(mode === "verified" ? 1 : 0);
+      const [hold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, source.issueId));
+      if (mode === "verified") {
+        expect(retire).toHaveBeenCalledOnce();
+        expect(hold).toMatchObject({ status: "resolved", outcome: "handed_back", evidence: { automaticRecovery: { replay: "verified_safe_replacement" } } });
+        // Prove that the real dispatch gate accepts the successor, not only that a row exists.
+        const dispatch = createRunDispatch(db);
+        await db.update(heartbeatRuns).set({ status: "queued" }).where(eq(heartbeatRuns.id, successors[0]!.id));
+        const result = await dispatch.cancelStaleQueuedRun({ companyId: source.companyId, runId: successors[0]!.id, expectedStatus: "queued", now: new Date() });
+        expect(result.outcome).toBe("not_stale");
+      } else {
+        expect(hold!.evidence.automaticRecovery).toEqual({ replay: "blocked" });
+      }
+    });
+    it.each(["missing_receipt", "other_run", "other_cause", "manual_reblock", "dependency_edit", "queued_comment"] as const)("honors blocking intent before safe replacement (%s)", async (mode) => {
+      const source = await seed(2);
+      await db.update(nativeRunFinalizations).set({ failureCode: "native_session_cleanup_quarantined" }).where(eq(nativeRunFinalizations.runId, source.runId));
+      const projected = await issueService(db).update(source.issueId, { status: "blocked" });
+      const evidence = { runId: source.runId, ...(mode === "missing_receipt" ? {} : {
+        nativeFailureBlock: { runId: mode === "other_run" ? randomUUID() : source.runId, statusVersion: projected!.statusVersion },
+      }) };
+      const [hold] = await db.insert(issueRecoveryActions).values({ companyId: source.companyId, sourceIssueId: source.issueId,
+        kind: "active_run_watchdog", cause: mode === "other_cause" ? "native_event_replay_conflict" : "native_session_cleanup_quarantined",
+        fingerprint: source.runId, ownerType: "board", returnOwnerAgentId: source.agentId,
+        status: "resolved", outcome: "blocked", evidence, nextAction: "Preserve this hold.",
+      }).returning();
+      if (mode === "manual_reblock") await issueService(db).update(source.issueId, { status: "blocked", actorUserId: "operator" });
+      if (mode === "dependency_edit") {
+        const blockerId = randomUUID();
+        await db.insert(issues).values({ id: blockerId, companyId: source.companyId, title: "Human dependency", status: "todo" });
+        await issueService(db).update(source.issueId, { blockedByIssueIds: [blockerId], actorUserId: "operator" });
+      }
+      const comment = mode === "queued_comment" ? await issueService(db).addComment(source.issueId, "Also explain the result", { userId: "operator" }) : null;
+      const retire = vi.fn(() => true);
+      await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: async run =>
+        run.id === source.runId ? { evidence: {}, retire } : null });
+      const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId));
+      const [after] = await db.select().from(issues).where(eq(issues.id, source.issueId));
+      const [afterHold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold!.id));
+      if (mode === "queued_comment") {
+        expect(successors).toHaveLength(1);
+        expect(after!.status).toBe("in_progress");
+        expect(afterHold!.outcome).toBe("handed_back");
+        expect((await db.select().from(issueComments).where(eq(issueComments.id, comment!.id)))[0]!.body).toBe("Also explain the result");
+      } else {
+        expect(successors).toHaveLength(0);
+        expect(retire).not.toHaveBeenCalled();
+        expect(after!.status).toBe("blocked");
+        expect(afterHold!.evidence).toEqual(evidence);
+        expect(afterHold!.outcome).toBe("blocked");
+      }
+    });
     it("automatically closes an exhausted incident once, preserves ownership, and records no replay", async () => {
       const source = await seed(3);
       await reconcileSafeNativeReplacements(db);

server/src/services/native-runtime/native-safe-replacement.ts

@@ -8,13 +8,16 @@ import {
   heartbeatRunEvents,
   heartbeatRuns,
   issues,
+  issueRecoveryActions,
   nativeRunFinalizations,
   toolInvocations,
   type Db,
 } from "@paperclipai/db";
 import { decideNativeReplacement } from "./native-replacement-evidence.js";
+import { issueService } from "../issues.js";
 import { issueRecoveryActionService } from "../issue-recovery-actions.js";
 import { buildExecutionContinuation } from "../execution-continuation.js";
+import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
 
 export const NATIVE_SAFE_REPLACEMENT_REASON = "native_safe_replacement";
 const record = (v: unknown): Record<string, unknown> =>
@@ -36,6 +39,10 @@ export async function reconcileSafeNativeReplacements(
   db: Db,
   now = new Date(),
   options: {
+    verifyStoppedSession?: (run: typeof heartbeatRuns.$inferSelect) => Promise<{
+      evidence: Record<string, unknown>;
+      retire: () => boolean;
+    } | null>;
     /** Test fault injection at durability boundaries; never exposed by an API. */
     failpoint?: (phase: "successor_inserted" | "lineage_committed") => void;
   } = {},
@@ -51,9 +58,9 @@ export async function reconcileSafeNativeReplacements(
       and(
         eq(heartbeatRuns.status, "failed"),
         eq(nativeRunFinalizations.phase, "terminal_failure"),
-        eq(
+        inArray(
           nativeRunFinalizations.failureCode,
-          "native_provider_terminal_failed",
+          ["native_provider_terminal_failed", "native_session_cleanup_quarantined", "provider_transport_failed"],
         ),
         isNull(nativeRunFinalizations.resultId),
         sql`coalesce(${nativeRunFinalizations.failureDetail}->>'successorRunId', '') = ''`,
@@ -84,6 +91,11 @@ export async function reconcileSafeNativeReplacements(
         );
       // Teardown is still in progress. The next sweep rechecks its durable outcome.
       if (leases.some((lease) => lease.releasedAt === null)) continue;
+      const stoppedSession = coordinator.failureCode !== "native_provider_terminal_failed"
+        ? await options.verifyStoppedSession?.(run) ?? null : null;
+      // A transport label alone is not evidence. Keep inspecting these candidates
+      // as process cleanup and the final transcript become durable.
+      if (coordinator.failureCode !== "native_provider_terminal_failed" && !stoppedSession) continue;
       const invocations = await db
         .select()
         .from(toolInvocations)
@@ -144,6 +156,9 @@ export async function reconcileSafeNativeReplacements(
           return [event.eventType];
         if (event.eventType === "tool.execution.started") {
           const name = typeof p.name === "string" ? p.name : "unknown tool";
+          const completedTaskControlCallIds = stoppedSession?.evidence.completedTaskControlCallIds;
+          if (name === "paperclip_finish" && Array.isArray(completedTaskControlCallIds) &&
+              completedTaskControlCallIds.includes(p.executionId)) return [];
           const receiptedRead = invocations.some(
             (row) =>
               (row.id === p.executionId ||
@@ -190,12 +205,12 @@ export async function reconcileSafeNativeReplacements(
         // A facade transport failure can conceal an authoritative protocol
         // rejection. A stopped process and read receipts do not resolve that.
         failureMeaningKnown:
-          typeof coordinator.failureDetail?.originalFailureCode === "string" &&
+          Boolean(stoppedSession) || (typeof coordinator.failureDetail?.originalFailureCode === "string" &&
           ![
             "notification_transport_failed",
             "provider_turn_failed",
             "native_provider_terminal_failed",
-          ].includes(coordinator.failureDetail.originalFailureCode),
+          ].includes(coordinator.failureDetail.originalFailureCode)),
         predecessorFenced:
           coordinator.leaseOwner === null && run.status === "failed",
         providerStopped:
@@ -209,8 +224,8 @@ export async function reconcileSafeNativeReplacements(
           )),
         historyComplete,
         effectInventoryComplete:
-          record(run.runnerProfileJson).recoveryEventInventoryVersion === 1 &&
-          record(execution.provider).kind === "codex",
+          Boolean(stoppedSession) || (record(run.runnerProfileJson).recoveryEventInventoryVersion === 1 &&
+          record(execution.provider).kind === "codex"),
         attempts: coordinator.attempt,
         invocations,
         apiReceipts: record(record(run.resultJson).apiToolReceipts),
@@ -304,16 +319,53 @@ export async function reconcileSafeNativeReplacements(
         if (
           !task ||
           task.assigneeAgentId !== run.agentId ||
-          !["in_progress", "in_review"].includes(task.status) ||
+          !["todo", "in_progress", "in_review", "blocked"].includes(task.status) ||
           (task.executionRunId !== null && task.executionRunId !== run.id) ||
           (task.checkoutRunId !== null && task.checkoutRunId !== run.id) ||
           !current ||
           current.phase !== "terminal_failure" ||
+          current.failureCode !== coordinator.failureCode ||
           current.failureDetail?.successorRunId ||
           current.failureDetail?.replacementDenied ||
           current.attempt >= 3
         )
           return false;
+        if (task.status === "blocked") {
+          const failureHolds = await tx.select().from(issueRecoveryActions).where(and(
+            eq(issueRecoveryActions.companyId, run.companyId),
+            eq(issueRecoveryActions.sourceIssueId, task.id),
+            eq(issueRecoveryActions.kind, "active_run_watchdog"),
+            eq(issueRecoveryActions.cause, current.failureCode!),
+            sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
+          )).for("update");
+          const ownsBlock = failureHolds.some(hold => {
+            const receipt = record(hold.evidence.nativeFailureBlock);
+            return receipt.runId === run.id && receipt.statusVersion === task.statusVersion;
+          });
+          if (!ownsBlock) return false;
+        }
+        if (stoppedSession) {
+          const [currentRun] = await tx.select().from(heartbeatRuns).where(and(
+            eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
+          )).for("update");
+          if (!currentRun || currentRun.status !== "failed" || currentRun.runnerInstanceId !== run.runnerInstanceId ||
+              currentRun.nativeSessionId !== run.nativeSessionId || currentRun.processPid || currentRun.processGroupId) return false;
+        }
+        if (task.status === "blocked") {
+          // Restore only this failure's unchanged projection. The normal issue
+          // service still enforces dependency readiness and assignee eligibility.
+          await issueService(tx as unknown as Db).update(task.id, { status: "in_progress" }, tx);
+        }
+        if (stoppedSession) {
+          // If the last ownership proof changes, roll back the status restoration.
+          if (!stoppedSession.retire()) throw new Error("native_replacement_stopped_session_changed");
+          await appendHeartbeatRunEvent(tx as unknown as Db, {
+            companyId: run.companyId, runId: run.id, agentId: run.agentId,
+            eventType: "native.stopped_text_turn_verified", stream: "system", level: "info",
+            message: "The previous runner and provider stopped. The interrupted turn had no external actions; any completion bookkeeping has a verified receipt.",
+            payload: stoppedSession.evidence,
+          });
+        }
         const successorRunId = randomUUID();
         const dueAt = new Date(now.getTime() + 30_000);
         const predecessorContext = { ...record(run.contextSnapshot) };
@@ -374,6 +426,7 @@ export async function reconcileSafeNativeReplacements(
           .set({
             failureDetail: {
               ...current.failureDetail,
+              ...(stoppedSession ? { stoppedTextTurn: stoppedSession.evidence } : {}),
               successorRunId,
               nextAction:
                 "Continue in the linked fresh provider session after the retry delay.",
@@ -395,6 +448,29 @@ export async function reconcileSafeNativeReplacements(
           outcome: "handed_back",
           resolutionNote: `Safe continuation is scheduled in run ${successorRunId}.`,
         });
+        // A previous sweep can have resolved the UI bookkeeping while retaining
+        // an effective no-replay hold. Retire only this exact failure's hold in
+        // the same transaction as its verified successor; unrelated holds remain.
+        const holds = await tx.select().from(issueRecoveryActions).where(and(
+          eq(issueRecoveryActions.companyId, run.companyId),
+          eq(issueRecoveryActions.sourceIssueId, task.id),
+          eq(issueRecoveryActions.kind, "active_run_watchdog"),
+          eq(issueRecoveryActions.cause, current.failureCode!),
+          sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
+        )).for("update");
+        for (const hold of holds) {
+          await tx.update(issueRecoveryActions).set({
+            status: "resolved", outcome: "handed_back", resolvedAt: now, updatedAt: now,
+            wakePolicy: null, monitorPolicy: null,
+            nextAction: `Safe continuation is scheduled in run ${successorRunId}.`,
+            resolutionNote: "Verified process termination and action receipts allow a fresh session to continue.",
+            evidence: { ...hold.evidence, verifiedReplacement: { successorRunId, recordedAt: now.toISOString() },
+              ...(hold.evidence.automaticRecovery ? { automaticRecovery: {
+                ...record(hold.evidence.automaticRecovery), replay: "verified_safe_replacement", successorRunId,
+              } } : {}),
+            },
+          }).where(eq(issueRecoveryActions.id, hold.id));
+        }
         await tx
           .update(heartbeatRuns)
           .set({ executionStatusDeliveryId: randomUUID() })

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

@@ -130,6 +130,7 @@ const state = vi.hoisted(() => ({
     }),
   ),
   cancel: vi.fn(),
+  copyBackCodexAuth: vi.fn(async () => "kept-host"),
   toolAuthorityDefinitions: vi.fn(
     async (_binding: Record<string, unknown>) => [],
   ),
@@ -180,6 +181,13 @@ vi.mock("../../vendor/paperclip-runner/index.js", async (importOriginal) => ({
   parsePaperclipQuestionSet: (value: unknown) => value,
 }));
 
+vi.mock("@paperclipai/adapter-codex-local/server", async (importOriginal) => ({
+  ...(await importOriginal<
+    typeof import("@paperclipai/adapter-codex-local/server")
+  >()),
+  copyBackCodexAuth: state.copyBackCodexAuth,
+}));
+
 vi.mock("./paperclip-runner-tool-authority.js", () => ({
   PaperclipRunnerToolAuthority: class {
     readonly binding: Record<string, unknown>;
@@ -230,10 +238,13 @@ import {
   buildNativeHarnessBackupManifest,
   cancelNativeSession,
   closeWarmNativeSessionsForEnvironment,
+  closeIdleWarmNativeSessionsForRestart,
   createGovernedWaitEventObservation,
   createRemoteRunnerProcessLauncher,
   createRunnerdBackend,
   executePaperclipNativeSession,
+  detachNativeSessionsForRestart,
+  NativeControllerDetachedForRestartError,
   getNativeSessionSteeringState,
   NativeSessionSteeringError,
   assertRemoteRunnerBuildMetadata,
@@ -4188,6 +4199,7 @@ function leaseDb(
           : table === heartbeatRuns
             ? [
                 {
+                  id: boundExecution.binding.runId,
                   agentId: boundExecution.binding.agentId,
                   companyId: boundExecution.binding.companyId,
                   nativeIssueId: boundExecution.binding.issueId,
@@ -4348,6 +4360,22 @@ function cancellationDb(options?: {
   };
 }
 
+describe("native startup restart detachment", () => {
+  it("remembers shutdown while the session is still opening and detaches its late publication", async () => {
+    const restarting = structuredClone(execution);
+    restarting.binding.runId = "restart-during-session-open";
+    const detach = vi.fn(async () => undefined);
+    await expect(detachNativeSessionsForRestart([restarting.binding.runId])).resolves.toMatchObject({ inactiveRunIds: [restarting.binding.runId] });
+    state.execute.mockReset().mockImplementationOnce(async (options) => {
+      await options.onSession({ detachControllerForRestart: detach });
+      expect(detach).toHaveBeenCalledOnce();
+      await options.onSession(null);
+      throw new Error("detachment closed the old event stream");
+    });
+    await expect(executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner" })).rejects.toBeInstanceOf(NativeControllerDetachedForRestartError);
+  });
+});
+
 describe("native resumed preparation timing", () => {
   it("keeps answered-question ingress at the run root rather than charging it to preparation", async () => {
     const answeredAtMs = Date.now();
@@ -5128,7 +5156,7 @@ describe("native warm session supervision", () => {
     expect(onGoalCheckpoint).toHaveBeenCalledOnce();
   });
 
-  it("closes an idle warm session before its remote environment is destroyed", async () => {
+  it.each(["environment deletion", "controller restart"])("closes an idle warm session before %s", async (shutdownKind) => {
     const close = vi.fn(async () => undefined);
     const warmExecution = {
       ...execution,
@@ -5144,7 +5172,12 @@ describe("native warm session supervision", () => {
       },
     } as NativeExecutionInputV1;
     state.execute.mockReset().mockImplementationOnce(async (options) => {
-      options.onSession?.({ close });
+      await options.onSession?.({ close });
+      await expect(closeWarmNativeSessionsForEnvironment({
+        environmentId: "environment-warm-delete",
+        reason: "environment deleted",
+      })).resolves.toMatchObject({ busy: 1 });
+      expect(close).not.toHaveBeenCalled();
       return {
         result: { summary: "completed" },
         terminal: { runTerminalState: "succeeded" },
@@ -5177,17 +5210,73 @@ describe("native warm session supervision", () => {
         reason: "environment deleted",
       }),
     ).resolves.toEqual({ closed: 0, busy: 0, failed: 0 });
-    await expect(
-      closeWarmNativeSessionsForEnvironment({
+    const closeResult = shutdownKind === "controller restart"
+      ? closeIdleWarmNativeSessionsForRestart()
+      : closeWarmNativeSessionsForEnvironment({
         environmentId: "environment-warm-delete",
         reason: "environment deleted",
-      }),
-    ).resolves.toEqual({ closed: 1, busy: 0, failed: 0 });
+      });
+    await expect(closeResult).resolves.toMatchObject({ closed: 1, failed: 0 });
     expect(close).toHaveBeenCalledExactlyOnceWith({
-      reason: "environment deleted",
+      reason: shutdownKind === "controller restart" ? "controller restart" : "environment deleted",
     });
   });
 
+  it.each([false, true])("checkpoints a busy warm session on release after the restart sweep: checkpoint fails=%s", async (checkpointFails) => {
+    const checkpointError = new Error("restart checkpoint failed");
+    let finishCheckpoint!: () => void;
+    const checkpoint = new Promise<void>((resolve, reject) => {
+      finishCheckpoint = checkpointFails ? () => reject(checkpointError) : resolve;
+    });
+    const close = vi.fn(async () => checkpoint);
+    const warmExecution = {
+      ...execution,
+      binding: {
+        ...execution.binding,
+        runId: "run-warm-restart-release",
+        executionWorkspaceId: "workspace-warm-restart-release",
+      },
+      session: {
+        ...execution.session,
+        normalizedSessionId: "session-warm-restart-release",
+        lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 },
+      },
+    } as NativeExecutionInputV1;
+    state.execute.mockReset().mockImplementationOnce(async (options) => {
+      await options.onSession?.({ close });
+      await expect(closeIdleWarmNativeSessionsForRestart()).resolves.toMatchObject({ busy: 1 });
+      expect(close).not.toHaveBeenCalled();
+      // The active turn can finish after the shutdown sweep has passed it.
+      return {
+        result: { summary: "completed during shutdown" },
+        terminal: { runTerminalState: "succeeded" },
+        turnId: "turn-warm-restart-release",
+        normalizedSessionId: warmExecution.session.normalizedSessionId,
+        providerSessionId: "provider-warm-restart-release",
+        driverKind: "test",
+        driverVersion: "1",
+        nativeEventCount: 1,
+        highestContiguousSourceSeq: 1,
+        usage: null,
+      };
+    });
+    let settled = false;
+    const running = executePaperclipNativeSession({
+      db: leaseDb(warmExecution),
+      execution: warmExecution,
+      runnerInstanceId: "runner-warm-restart-release",
+    }).then((result) => { settled = true; return result; });
+    try {
+      await vi.waitFor(() => expect(close).toHaveBeenCalledExactlyOnceWith({ reason: "controller restart" }));
+      expect(settled).toBe(false);
+    } finally {
+      finishCheckpoint();
+      if (checkpointFails) await expect(running).rejects.toBe(checkpointError);
+      else await running;
+    }
+    await expect(closeIdleWarmNativeSessionsForRestart()).resolves.toEqual({ closed: 0, busy: 0, failed: 0 });
+  });
+
   it("preserves the active turn when a warm checkpoint resumes the same run", async () => {
     const stateBase = await mkdtemp(
       join(tmpdir(), "paperclip-warm-same-run-recovery-"),
@@ -5919,6 +6008,25 @@ describe("native warm session supervision", () => {
 });
 
 describe("native session bounded recovery", () => {
+  it("does not turn an acknowledged Stop before completion into a failure or a retry", async () => {
+    const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
+    const stop: Record<string, unknown> = {};
+    state.execute.mockReset().mockImplementationOnce(async () => {
+      Object.assign(stop, { cancelledByActorType: "user", cancelledByUserId: "board", nativeCancellation: {
+        schema: "paperclip.native-cancellation.v1", ...execution.binding, scope: "run", reasonCode: "cancellation_run_only",
+        dispatched: true, dispatchState: "acknowledged", intentAuditId: "intent", acknowledgementAuditId: "ack",
+      } });
+      throw new Error("native_finalization_missing: session returned no semantic result");
+    });
+    state.upsertRecoveryAction.mockClear();
+    await expect(executePaperclipNativeSession({
+      db: leaseDb(execution, {}, stop, updates), execution, runnerInstanceId: "stop-before-completion",
+    })).rejects.toThrow("native_cancellation_pending_recovery");
+    expect(updates.some(update => update.table === heartbeatRuns && update.values.status === "failed")).toBe(false);
+    expect(updates.some(update => update.table === nativeRunFinalizations && update.values.failureCode === "native_retry_cancelled")).toBe(true);
+    expect(state.upsertRecoveryAction).not.toHaveBeenCalled();
+  });
+
   it("keeps typed integrity failure permanent even if a wrapper changes its message", () => {
     const failure = new NativeSessionProtocolIntegrityError(
       "semantic_input_digest_mismatch",
@@ -6060,7 +6168,7 @@ describe("native session bounded recovery", () => {
         );
         expect(updateIssue).toHaveBeenCalledWith(
           execution.binding.issueId,
-          { status: "in_review" },
+          { status: "blocked" },
           expect.anything(),
         );
       } finally {
@@ -6114,7 +6222,7 @@ describe("native session bounded recovery", () => {
     const failure = new NativeSessionCleanupQuarantinedError();
     state.execute.mockReset().mockRejectedValueOnce(failure);
     state.upsertRecoveryAction.mockReset().mockResolvedValue({});
-    const updateIssue = vi.fn(async () => null);
+    const updateIssue = vi.fn(async () => ({ status: "blocked", statusVersion: 7 }));
     const service = vi
       .spyOn(issueServiceModule, "issueService")
       .mockReturnValue({ update: updateIssue } as unknown as ReturnType<
@@ -6146,6 +6254,7 @@ describe("native session bounded recovery", () => {
       expect(state.upsertRecoveryAction).toHaveBeenCalledWith(
         expect.objectContaining({
           cause: "native_session_cleanup_quarantined",
+          evidence: expect.objectContaining({ nativeFailureBlock: { runId: execution.binding.runId, statusVersion: 7 } }),
           ownerType: "board",
           wakePolicy: null,
           nextAction: expect.stringContaining(
@@ -6155,7 +6264,7 @@ describe("native session bounded recovery", () => {
       );
       expect(updateIssue).toHaveBeenCalledWith(
         execution.binding.issueId,
-        { status: "in_review" },
+        { status: "blocked" },
         expect.anything(),
       );
     } finally {
@@ -6461,7 +6570,7 @@ describe("native session bounded recovery", () => {
     });
   });
 
-  it("escalates exhausted result-less sessions to board review instead of leaving the provider as its own owner", () => {
+  it("blocks exhausted result-less sessions without manufacturing a human review", () => {
     expect(
       nativeSessionRecoveryProjection({
         phase: "retryable_failure",
@@ -6485,7 +6594,7 @@ describe("native session bounded recovery", () => {
       }),
     ).toEqual({
       exhausted: true,
-      issueStatus: "in_review",
+      issueStatus: "blocked",
       recoveryOwner: { kind: "board" },
       recoveryActionOwnerType: "board",
       recoveryActionOwnerAgentId: null,
@@ -6564,6 +6673,10 @@ describe("native process ownership", () => {
     const onSpawn = vi.fn(async () => undefined);
     state.createBackend.mockClear();
     state.execute.mockReset().mockImplementation(async (options) => {
+      await options.onSessionAdmission();
+      expect(updates).toContainEqual({ table: heartbeatRunEvents, values: expect.objectContaining({
+        eventType: "native.process_start_requested", runId: execution.binding.runId,
+      }) });
       await options.backend.onSpawn(processMetadata);
       return {
         result: { summary: "completed" },
@@ -6579,9 +6692,7 @@ describe("native process ownership", () => {
     });
     const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
     state.createBackend.mockImplementationOnce((_input, options) => {
-      expect(updates).toContainEqual({ table: heartbeatRunEvents, values: expect.objectContaining({
-        eventType: "native.process_start_requested", runId: execution.binding.runId,
-      }) });
+      expect(updates.some(update => update.values.eventType === "native.process_start_requested")).toBe(false);
       return { kind: "test", onSpawn: options.onSpawn };
     });
 
@@ -6740,6 +6851,89 @@ describe("runnerd provider runtime wiring", () => {
     expect(state.execute).toHaveBeenCalledOnce();
   });
 
+  it.each([
+    ["open", "before-close"],
+    ["open", "during-close"],
+    ["recover", "before-close"],
+    ["recover", "during-close"],
+  ] as const)("preserves managed Codex credentials after %s session detachment %s", async (mode, timing) => {
+    let finishClose!: () => void;
+    const closing = new Promise<void>((resolve) => { finishClose = resolve; });
+    const close = vi.fn(async () => {
+      if (timing === "during-close") await closing;
+    });
+    const detach = vi.fn(async () => undefined);
+    const rawSession = { close, detachControllerForRestart: detach };
+    state.copyBackCodexAuth.mockClear();
+    state.createBackend.mockReturnValueOnce({
+      kind: "test",
+      openSession: async () => rawSession,
+      recoverSession: async () => ({ recovered: true, session: rawSession }),
+    } as never);
+    const backend = await createRunnerdBackend({
+      db: leaseDb(execution),
+      execution,
+      runnerInstanceId: "runner-managed-credential-detach",
+      managedAiCredentialHome: join(isolatedStateDirectory, "managed-home"),
+    });
+    state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
+    const root = state.createTransport.mock.calls.at(-1)![0].stateDirectory!;
+    const authPath = join(root, "codex-home", "auth.json");
+    const auth = JSON.stringify({ OPENAI_API_KEY: "fixture-managed-codex-credential" });
+    await mkdir(join(root, "codex-home"), { recursive: true });
+    await writeFile(authPath, auth);
+    const session = mode === "open"
+      ? await backend.openSession({} as never)
+      : (await backend.recoverSession!({} as never, {
+          signal: new AbortController().signal,
+        })).session!;
+
+    if (timing === "before-close") {
+      await session.detachControllerForRestart!();
+      await session.close({ reason: "old controller finalizer" });
+    } else {
+      const closed = session.close({ reason: "old controller finalizer" });
+      await session.detachControllerForRestart!();
+      finishClose();
+      await closed;
+    }
+
+    expect(detach).toHaveBeenCalledOnce();
+    expect(close).toHaveBeenCalledOnce();
+    expect(state.copyBackCodexAuth).not.toHaveBeenCalled();
+    await expect(readFile(authPath, "utf8")).resolves.toBe(auth);
+  });
+
+  it("still cleans up managed Codex credentials after an owned session closes", async () => {
+    const close = vi.fn(async () => undefined);
+    state.copyBackCodexAuth.mockClear();
+    state.createBackend.mockReturnValueOnce({
+      kind: "test",
+      openSession: async () => ({ close }),
+    } as never);
+    const managedHome = join(isolatedStateDirectory, "managed-home");
+    const backend = await createRunnerdBackend({
+      db: leaseDb(execution),
+      execution,
+      runnerInstanceId: "runner-managed-credential-close",
+      managedAiCredentialHome: managedHome,
+    });
+    state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
+    const root = state.createTransport.mock.calls.at(-1)![0].stateDirectory!;
+    const authPath = join(root, "codex-home", "auth.json");
+    await mkdir(join(root, "codex-home"), { recursive: true });
+    await writeFile(authPath, "fixture-managed-codex-credential");
+    const session = await backend.openSession({} as never);
+
+    await session.close({ reason: "completed" });
+    await session.close({ reason: "repeated cleanup" });
+
+    expect(state.copyBackCodexAuth).toHaveBeenCalledExactlyOnceWith(
+      expect.objectContaining({ hostAuthPath: join(managedHome, "auth.json") }),
+    );
+    await expect(access(authPath)).rejects.toMatchObject({ code: "ENOENT" });
+  });
+
   it("stages from the authenticated run snapshot and cleans up after the provider turn", async () => {
     const cleanup = vi.fn(async () => undefined);
     state.stageNativeRunnerWakeAttachments.mockResolvedValueOnce({
@@ -9638,10 +9832,12 @@ describe("runnerd provider runtime wiring", () => {
       runtimeContext: nativeRuntimeContextFixture(),
     } as unknown as NativeExecutionInputV1;
     state.createBackend.mockClear();
+    const onSpawn = vi.fn(async () => undefined);
     await createRunnerdBackend({
       db: leaseDb(acpxExecution),
       execution: acpxExecution,
       runnerInstanceId: "runner",
+      onSpawn,
     });
 
     expect(state.createBackend).toHaveBeenCalledWith(
@@ -9660,6 +9856,7 @@ describe("runnerd provider runtime wiring", () => {
         provider: "acpx",
         acpxAgent: "codex",
         acpxPermissionMode: "approve-reads",
+        onSpawn,
       }),
     );
   });

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

@@ -1,6 +1,9 @@
 import { copyBackCodexAuth } from "@paperclipai/adapter-codex-local/server";
 import { nativeCompletionFeedback } from "./native-completion-feedback.js";
-import { PROCESS_START_REQUESTED } from "../native-local-process-stop.js";
+import { hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js";
+import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js";
+import { prepareVerifiedRemoteProviderPack } from "./remote-provider-pack.js";
+import { readNativeLocalProcessStop, PROCESS_START_REQUESTED } from "../native-local-process-stop.js";
 import { remoteLeaseCleanupScope } from "../remote-execution-termination.js";
 import { resolveConnectorAssignments, isConnectorSkill } from "../connector-runtime.js";
 import {
@@ -53,6 +56,7 @@ import {
   NativeSessionCleanupQuarantinedError,
   NativeSessionProtocolIntegrityError,
   completeRetainedNativeSessionCleanup,
+  completeTerminatedLocalNativeSessionCleanup,
   acpxRuntimeSessionDirectoryName,
   createNativeSessionBackend,
   createRunnerdCodexTransport,
@@ -86,6 +90,7 @@ import {
   eq,
   gt,
   inArray,
+  isNull,
   like,
   notInArray,
   or,
@@ -179,7 +184,18 @@ export class NativeCancellationPendingRecoveryError extends Error {
   }
 }
 
+export class NativeControllerDetachedForRestartError extends Error {
+  constructor() {
+    super("native_controller_detached_for_restart");
+    this.name = "NativeControllerDetachedForRestartError";
+  }
+}
+
 const activeNativeSessions = new Map<string, ActiveNativeSession>();
+// Shutdown can race provider startup before onSession publishes its handle.
+// Retain the request for the remainder of this controller's lifetime so that
+// the late publication detaches before it can dispatch another turn.
+const nativeRunsDetachingForRestart = new Set<string>();
 
 export async function detachNativeSessionsForRestart(
   runIds: readonly string[],
@@ -192,6 +208,7 @@ export async function detachNativeSessionsForRestart(
   const inactiveRunIds: string[] = [];
   const unsupportedRunIds: string[] = [];
   for (const runId of new Set(runIds)) {
+    nativeRunsDetachingForRestart.add(runId);
     const active = activeNativeSessions.get(runId);
     if (!active) {
       inactiveRunIds.push(runId);
@@ -325,6 +342,7 @@ type WarmNativeSession = {
   companyId: string;
   environmentId: string | null;
   busy: boolean;
+  closeOnReleaseReason?: string;
   idleTimer: ReturnType<typeof setTimeout> | null;
   lastActivityAt: string;
 };
@@ -341,15 +359,38 @@ const warmNativeSessions = new Map<string, WarmNativeSession>();
 export async function closeWarmNativeSessionsForEnvironment(input: {
   environmentId: string;
   reason: string;
+}): Promise<{ closed: number; busy: number; failed: number }> {
+  return closeIdleWarmNativeSessions(input);
+}
+
+/** Suspend idle owners and persist their remote backup before a controller
+ * exits. Active turns keep their separate authenticated restart handoff. */
+export async function closeIdleWarmNativeSessionsForRestart(): Promise<{
+  closed: number; busy: number; failed: number;
+}> {
+  return closeIdleWarmNativeSessions({
+    reason: "controller restart",
+    closeBusyOnRelease: true,
+  });
+}
+
+async function closeIdleWarmNativeSessions(input: {
+  environmentId?: string;
+  reason: string;
+  closeBusyOnRelease?: boolean;
 }): Promise<{ closed: number; busy: number; failed: number }> {
   let closed = 0;
   let busy = 0;
   let failed = 0;
   for (const [sessionId, entry] of [...warmNativeSessions]) {
-    if (entry.environmentId !== input.environmentId) {
+    if (input.environmentId !== undefined && entry.environmentId !== input.environmentId) {
       continue;
     }
     if (entry.busy) {
+      // A busy turn can complete while another idle session is checkpointing.
+      // Fence that entry now so its eventual release cannot leave a new idle
+      // owner behind after the shutdown sweep has already passed it.
+      if (input.closeBusyOnRelease) entry.closeOnReleaseReason = input.reason;
       busy += 1;
       continue;
     }
@@ -2251,6 +2292,129 @@ export function rebaseRetainedNativeCleanupProviderHome(
   }
 }
 
+/** Prove that a crashed local Codex turn ended without external effects. No provider
+ * is launched and no retained state is rewritten. The successor must use a fresh
+ * normalized session; the old directory remains available for investigation. */
+export async function verifyStoppedNativeSessionForReplacement(
+  db: Db,
+  run: typeof heartbeatRuns.$inferSelect,
+): Promise<{ evidence: Record<string, unknown>; retire: () => boolean } | null> {
+  try {
+    if (run.runtimeMode !== "native" || run.status !== "failed" || !run.finishedAt ||
+        !run.nativeIssueId || !run.nativeSessionId || !run.runnerInstanceId) return null;
+    const execution = parseNativeExecutionInput(record(run.runnerProfileJson).nativeExecutionInput);
+    if (execution.provider.kind !== "codex" || execution.session.driverKind !== "codex_app_server" ||
+        execution.binding.runId !== run.id || execution.binding.companyId !== run.companyId ||
+        execution.binding.agentId !== run.agentId || execution.binding.issueId !== run.nativeIssueId ||
+        nativeSessionKey(execution) !== run.nativeSessionId ||
+        record(run.runnerProfileJson).nativeToolContractFingerprint !== nativeToolContractFingerprintForTarget("local")) return null;
+    const scope = nativeSessionScopeKey(execution);
+    const idle = () => !activeNativeSessions.has(run.id) && !executingRunnerdSessionScopes.has(scope) &&
+      !initializingSessionToolAuthorities.has(scope) && !warmNativeSessions.has(scope);
+    if (!idle()) return null;
+    const leases = await db.select().from(environmentLeases).where(and(
+      eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id)));
+    if (leases.some(lease => lease.provider !== "local" || !lease.releasedAt)) return null;
+    const stopped = await readNativeLocalProcessStop(db, run.companyId, run.id);
+    if (!stopped) return null;
+    const root = scopedRunnerdStateRoot(execution);
+    const snapshot = cleanupStateSnapshot(root);
+    const identity = record(snapshot.control.identity);
+    if (!durableIdentityMatchesExecution(identity, execution) || identity.runnerInstanceId !== run.runnerInstanceId ||
+        !["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"].every(key =>
+          typeof identity[key] === "string" && identity[key] && snapshot.runner[key] === identity[key]) ||
+        snapshot.runner.lifecycle !== "ready" || snapshot.provider.lifecycle !== "turn_active" ||
+        record(snapshot.provider.config).provider !== "codex" || record(snapshot.provider.config).cwd !== execution.workspace.cwd ||
+        !Array.isArray(snapshot.control.committedEvents) || !Array.isArray(snapshot.control.commands)) return null;
+    const events = snapshot.control.committedEvents.map(entry => record(record(record(entry).envelope).payload));
+    const providerEvent = events.filter(event => ["session.started", "session.resumed"].includes(String(event.eventType))).at(-1);
+    const provider = record(providerEvent?.payload);
+    const bound = (event: Record<string, unknown>) => validatePrpEvent(event).ok && event.sourceKind === "runner" &&
+      event.sourceInstanceId === run.runnerInstanceId && event.runId === run.id &&
+      event.normalizedSessionId === run.nativeSessionId && event.turnId === identity.turnId && event.itemId === identity.itemId;
+    if (!providerEvent || !bound(providerEvent) || !cleanupProcessAbsent(provider.processId) ||
+        provider.processId === stopped.processPid || typeof provider.providerSessionId !== "string" ||
+        snapshot.provider.threadId !== provider.providerSessionId ||
+        typeof snapshot.provider.activeProviderTurnId !== "string") return null;
+    const receipts = await db.select().from(heartbeatRunEvents).where(and(
+      eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
+      eq(heartbeatRunEvents.sourceInstanceId, run.runnerInstanceId),
+      inArray(heartbeatRunEvents.eventType, ["session.started", "session.resumed"]))).limit(2);
+    const receipt = receipts.length === 1 ? receipts[0] : undefined;
+    const durableEvent = record(record(receipt?.payload).prpEvent);
+    // The persisted adapter enriches identity payloads with driverSessionId. Bind
+    // the normalized and provider identities explicitly instead of comparing raw JSON.
+    const durableProvider = record(durableEvent.payload);
+    if (!receipt || !validatePrpEvent(durableEvent).ok || durableEvent.runId !== run.id ||
+        durableEvent.sourceInstanceId !== run.runnerInstanceId || durableEvent.normalizedSessionId !== run.nativeSessionId ||
+        receipt.sourceEventId !== `${run.runnerInstanceId}:${run.id}:${durableEvent.sourceSeq}` ||
+        receipt.sourcePayloadSha256 !== nativeSha256(durableEvent) ||
+        (durableProvider.processId !== undefined && durableProvider.processId !== provider.processId) ||
+        (durableProvider.driverSessionId ?? durableProvider.providerSessionId) !== provider.providerSessionId) return null;
+    const turnId = snapshot.provider.activeProviderTurnId;
+    if (!snapshot.control.commands.map(record).some(command => command.type === "turn.start" && command.status === "completed" &&
+        record(record(command.result).result).providerTurnId === turnId) ||
+        !events.some(event => bound(event) && event.eventType === "turn.accepted" &&
+          record(event.payload).providerTurnId === turnId && record(event.payload).providerSessionId === provider.providerSessionId)) return null;
+    // Completion bookkeeping may precede the final answer. Its exact accepted
+    // receipt is safe to preserve; arbitrary provider tools still prevent replay.
+    const completedTaskControlCalls: Array<{ callId: string; input: unknown }> = [];
+    for (const event of events.filter(event => event.eventType === "semantic_tool.input")) {
+      const semantic = record(record(event.payload).semantic_tool);
+      const correlation = record(semantic.correlation);
+      if (!bound(event) || semantic.operationId !== "paperclip_finish" || semantic.phase !== "input" ||
+          typeof semantic.callId !== "string" || !validatePrpStructuredRunResult(semantic.input).ok ||
+          correlation.runId !== run.id || correlation.normalizedSessionId !== run.nativeSessionId ||
+          correlation.turnId !== identity.turnId || correlation.itemId !== identity.itemId ||
+          record(semantic.content).digest !== `sha256:${nativeSha256(semantic.input)}` ||
+          !events.some(resultEvent => {
+            const result = record(record(resultEvent.payload).semantic_tool);
+            return bound(resultEvent) && resultEvent.eventType === "semantic_tool.result" &&
+              result.operationId === semantic.operationId && result.callId === semantic.callId &&
+              result.outcome === "succeeded" && result.operationReceiptId === `operation_${semantic.callId}` &&
+              canonicalJson(result.correlation) === canonicalJson(semantic.correlation);
+          }) || !snapshot.control.commands.map(record).some(command => {
+            const payload = record(command.payload);
+            return command.type === "semantic_tool.result" && command.status === "completed" &&
+              payload.callId === semantic.callId && payload.operationId === semantic.operationId &&
+              payload.sourceEventId === event.sourceEventId && payload.isError === false &&
+              record(payload.result).success === true && canonicalJson(payload.input) === canonicalJson(semantic.input) &&
+              canonicalJson(payload.correlation) === canonicalJson(semantic.correlation);
+          })) return null;
+      completedTaskControlCalls.push({ callId: semantic.callId, input: semantic.input });
+    }
+    if (completedTaskControlCalls.length > 1) return null;
+    const pendingInventory = JSON.stringify([snapshot.runner.outbox, snapshot.provider.pendingEvents, snapshot.provider.queuedEvents]);
+    if (/semantic_tool\.input|mcp_app\.tool_input|runtime\.input\.requested|runtime_request\.created/.test(pendingInventory) ||
+        events.some(event => ["mcp_app.tool_input", "runtime.input.requested", "runtime_request.created"].includes(String(event.eventType)))) return null;
+    if (snapshot.control.commands.map(record).some(command => command.status === "pending" &&
+        !["turn.stop", "runner.drain", "runner.suspend"].includes(String(command.type)))) return null;
+    const home = cleanupProviderHomeSnapshot(resolve(root, "codex-home"), false);
+    const rollouts = home.entries.filter(entry => !entry.directory && entry.path.startsWith("sessions/") &&
+      basename(entry.path).endsWith(`-${provider.providerSessionId}.jsonl`));
+    if (rollouts.length !== 1 || rollouts[0]!.size > 32 * 1024 * 1024) return null;
+    const rolloutPath = resolve(root, "codex-home", rollouts[0]!.path);
+    const bytes = readBoundedNativeFile(rolloutPath, 32 * 1024 * 1024, "native_crash_inventory_unproven");
+    // A partial final write is not a closed transcript.
+    if (!bytes.toString("utf8").endsWith("\n")) return null;
+    const rows = bytes.toString("utf8").trimEnd().split("\n").map(line => JSON.parse(line));
+    if (!stoppedCodexTurnIsTextOnly({ rows, threadId: provider.providerSessionId, turnId, cwd: execution.workspace.cwd, completedTaskControlCalls })) return null;
+    const rolloutSha256 = nativeSha256(bytes.toString("utf8"));
+    const evidence = { schema: "paperclip.stopped_text_turn.v1", runId: run.id, nativeSessionId: run.nativeSessionId,
+      runnerInstanceId: run.runnerInstanceId, processPid: stopped.processPid, providerPid: provider.processId,
+      providerSessionId: provider.providerSessionId, providerTurnId: turnId,
+      stateFingerprint: snapshot.fingerprint, rolloutSha256,
+      completedTaskControlCallIds: completedTaskControlCalls.map(call => call.callId) };
+    return {
+      evidence,
+      retire: () => idle() && cleanupProcessAbsent(stopped.processPid) && cleanupProcessAbsent(provider.processId) &&
+        cleanupStateSnapshot(root).fingerprint === snapshot.fingerprint &&
+        nativeSha256(readBoundedNativeFile(rolloutPath, 32 * 1024 * 1024, "native_crash_inventory_unproven").toString("utf8")) === rolloutSha256 &&
+        completeTerminatedLocalNativeSessionCleanup({ companyId: run.companyId, runId: run.id, runnerInstanceId: run.runnerInstanceId! }),
+    };
+  } catch { return null; }
+}
+
 /** Exact local cleanup only: the accepted result and original quarantine are
  * never rewritten. A failed/interrupted maintenance attempt is retained for
  * inspection, not retried from an older snapshot with unknown process owners. */
@@ -5465,11 +5629,15 @@ async function releaseWarmNativeSession(
   entry.busy = false;
   entry.lastActivityAt = new Date().toISOString();
   if (entry.idleTimer !== null) clearTimeout(entry.idleTimer);
-  if (failed) {
+  if (failed || entry.closeOnReleaseReason !== undefined) {
     warmNativeSessions.delete(sessionId);
-    await entry.session
-      .close({ reason: "warm native session failed" })
-      .catch(() => undefined);
+    const closing = entry.session.close({
+      reason: entry.closeOnReleaseReason ?? "warm native session failed",
+    });
+    // Restart checkpointing is required to restore this successful session.
+    // Surface failure instead of reporting a clean release without authority.
+    if (entry.closeOnReleaseReason !== undefined) await closing;
+    else await closing.catch(() => undefined);
     return;
   }
   entry.idleTimer = setTimeout(() => {
@@ -5538,7 +5706,7 @@ export function nativeSessionRecoveryProjection(input: {
     issueStatus:
       exhausted &&
       input.failureCode !== NATIVE_ADOPTED_RUNNER_AUTHENTICATION_TIMEOUT
-        ? ("in_review" as const)
+        ? ("blocked" as const)
         : null,
     recoveryOwner: exhausted
       ? { kind: "board" as const }
@@ -7661,17 +7829,6 @@ async function executePaperclipNativeSessionWithinScope(
       input.db,
       input.execution.binding,
     );
-    // Invalidate prior stop evidence before a backend can spawn. A crash between
-    // spawn and the PID callback must not make an old receipt authorize a turn.
-    await appendHeartbeatRunEvent(input.db, {
-      companyId: input.execution.binding.companyId,
-      runId: input.execution.binding.runId,
-      agentId: input.execution.binding.agentId,
-      eventType: PROCESS_START_REQUESTED,
-      stream: "system",
-      level: "info",
-      message: "Native execution requested; prior local stop evidence no longer applies.",
-    });
     const runnerdBackend =
       input.useRunnerd && input.backend === undefined
         ? await createRunnerdBackend({
@@ -7704,6 +7861,18 @@ async function executePaperclipNativeSessionWithinScope(
         trace.activate(runnerSessionStartupScope);
         const result = await trace.run(runnerSessionStartupScope, () =>
           executeNativeSession({
+            onSessionAdmission: async () => {
+              // Invalidate prior stop evidence before a backend can spawn.
+              await appendHeartbeatRunEvent(input.db, {
+                companyId: input.execution.binding.companyId,
+                runId: input.execution.binding.runId,
+                agentId: input.execution.binding.agentId,
+                eventType: PROCESS_START_REQUESTED,
+                stream: "system",
+                level: "info",
+                message: "Native execution requested; prior local stop evidence no longer applies.",
+              });
+            },
             input: runnerExecution,
             remoteCleanupScope: remoteCleanupLease ? remoteLeaseCleanupScope(remoteCleanupLease) : undefined,
             turnTimeoutMs: input.turnTimeoutMs,
@@ -7817,7 +7986,7 @@ async function executePaperclipNativeSessionWithinScope(
                 `[paperclip-runner] provider session continuity break: exact resume failed (${continuity.reason}); old driver session=${continuity.previousDriverSessionId}, old provider session=${continuity.previousProviderSessionId ?? "unavailable"}, replacement driver session=${continuity.replacementDriverSessionId}, replacement provider session=${continuity.replacementProviderSessionId ?? "unavailable"}\n`,
               );
             },
-            onSession: (session) => {
+            onSession: async (session) => {
               releaseRegisteredGoalController();
               if (session?.goal) {
                 releaseGoalController = registerLiveRunnerGoalController(
@@ -7878,12 +8047,15 @@ async function executePaperclipNativeSessionWithinScope(
                   warmNativeSessions.delete(warmSessionId);
                 }
               }
-              if (session)
+              if (session) {
                 activeNativeSessions.set(input.execution.binding.runId, {
                   session,
                   cancelRequested: false,
                 });
-              else {
+                if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) {
+                  await session.detachControllerForRestart?.();
+                }
+              } else {
                 activeNativeSessions.delete(input.execution.binding.runId);
                 clearSteeringDeliveries(input.execution.binding.runId);
                 clearNativeRuntimeRequestResolutions(
@@ -7923,6 +8095,14 @@ async function executePaperclipNativeSessionWithinScope(
     clearSteeringDeliveries(input.execution.binding.runId);
     clearNativeRuntimeRequestResolutions(input.execution.binding.runId);
   } catch (error) {
+    if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) {
+      await leaseRenewal.stop().catch(() => undefined);
+      activeNativeSessions.delete(input.execution.binding.runId);
+      // Disconnecting deliberately ends the old event consumer. It is not a
+      // provider failure and must not overwrite the shutdown adoption record
+      // with a retry or release the still-live runner's lease.
+      throw new NativeControllerDetachedForRestartError();
+    }
     const protocolIntegrityFailure =
       error instanceof NativeSessionProtocolIntegrityError ? error : null;
     const ownershipUnverified =
@@ -7965,6 +8145,37 @@ async function executePaperclipNativeSessionWithinScope(
       activeNativeSessions.delete(input.execution.binding.runId);
       clearSteeringDeliveries(input.execution.binding.runId);
       clearNativeRuntimeRequestResolutions(input.execution.binding.runId);
+      // Stop before paperclip_finish is normal. Bounded provider teardown has
+      // finished; a missing result must not overwrite cancellation or trigger
+      // another turn to perform completion bookkeeping.
+      if (protocolIntegrityFailure === null && error instanceof Error && error.message === "native_finalization_missing: session returned no semantic result") {
+        const [stoppedRun] = await input.db.select().from(heartbeatRuns).where(and(
+          eq(heartbeatRuns.id, input.execution.binding.runId),
+          eq(heartbeatRuns.companyId, input.execution.binding.companyId),
+          eq(heartbeatRuns.agentId, input.execution.binding.agentId),
+          eq(heartbeatRuns.nativeIssueId, input.execution.binding.issueId),
+        )).limit(1);
+        if (stoppedRun && hasAcknowledgedNativeStopIntent(stoppedRun)) {
+          const [settled] = await input.db.update(nativeRunFinalizations).set({
+            phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null,
+            leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: null,
+            updatedAt: new Date(),
+          }).where(and(
+            eq(nativeRunFinalizations.runId, input.execution.binding.runId),
+            eq(nativeRunFinalizations.companyId, input.execution.binding.companyId),
+            eq(nativeRunFinalizations.leaseOwner, leaseOwner),
+            eq(nativeRunFinalizations.attempt, attempt),
+            isNull(nativeRunFinalizations.resultId),
+          )).returning({ runId: nativeRunFinalizations.runId });
+          if (settled) {
+            await stoppedLeaseRenewal;
+            if (warmSessionId !== null && lifecyclePolicy.mode === "warm") {
+              await releaseWarmNativeSession(warmSessionId, warmSessionOwnerToken, lifecyclePolicy.idleTimeoutMs, true);
+            }
+            error = new NativeCancellationPendingRecoveryError();
+          }
+        }
+      }
       if (
         error instanceof NativeResultPendingFinalizationError ||
         error instanceof NativeCancellationPendingRecoveryError
@@ -8178,12 +8389,14 @@ async function executePaperclipNativeSessionWithinScope(
             failureTask.executionRunId === input.execution.binding.runId) &&
           (!failureTask.checkoutRunId ||
             failureTask.checkoutRunId === input.execution.binding.runId);
+        let failureBlockStatusVersion: number | undefined;
         if (stillOwnsTask && recoveryProjection.issueStatus) {
-          await issueService(tx as unknown as Db).update(
+          const projected = await issueService(tx as unknown as Db).update(
             input.execution.binding.issueId,
             { status: recoveryProjection.issueStatus },
             tx,
           );
+          if (projected?.status === "blocked") failureBlockStatusVersion = projected.statusVersion;
         }
         if (
           !ownershipUnverified &&
@@ -8211,6 +8424,10 @@ async function executePaperclipNativeSessionWithinScope(
             .digest("hex"),
           evidence: {
             runId: input.execution.binding.runId,
+            ...(failureBlockStatusVersion !== undefined ? { nativeFailureBlock: {
+              runId: input.execution.binding.runId,
+              statusVersion: failureBlockStatusVersion,
+            } } : {}),
             coordinatorAttempt: attempt,
             sourceFailureCode,
             recoveryDisposition: failureCode,
@@ -10283,82 +10500,99 @@ async function createRunnerdBackendWithinSessionClaim(
       configuredProviderPackRoot &&
       stagedRemoteProviderPackRoot
     ) {
-      let preinstalledProviderPack = await discoverPreinstalledProviderPack();
-      if (preinstalledProviderPack) {
-        try {
-          await measureNativeRunnerSpan(
-            input.trace,
-            "provider_pack.verify_preinstalled",
-            () => verifyRemoteProviderPack(preinstalledProviderPack!),
-          );
-          const escapedSource = preinstalledProviderPack.replaceAll(
+      const packSource = await prepareVerifiedRemoteProviderPack({
+        verifyStaged: () => measureNativeRunnerSpan(
+          input.trace,
+          "provider_pack.verify",
+          () => verifyRemoteProviderPack(stagedRemoteProviderPackRoot),
+        ),
+        usePreinstalled: async () => {
+          let preinstalledProviderPack = await discoverPreinstalledProviderPack();
+          if (preinstalledProviderPack) {
+            try {
+              await measureNativeRunnerSpan(
+                input.trace,
+                "provider_pack.verify_preinstalled",
+                () => verifyRemoteProviderPack(preinstalledProviderPack!),
+              );
+              const escapedSource = preinstalledProviderPack.replaceAll(
+                "'",
+                "'\\''",
+              );
+              const escapedTarget = stagedRemoteProviderPackRoot.replaceAll(
+                "'",
+                "'\\''",
+              );
+              const escapedParent = posix
+                .dirname(stagedRemoteProviderPackRoot)
+                .replaceAll("'", "'\\''");
+              const linked = await remoteCommandRunner.execute({
+                command: "sh",
+                args: [
+                  "-c",
+                  `umask 077; mkdir -p '${escapedParent}' && rm -rf '${escapedTarget}' && ln -s '${escapedSource}' '${escapedTarget}'`,
+                ],
+                cwd: remoteTarget.remoteCwd,
+                bypassSession: true,
+                timeoutMs: 10_000,
+              });
+              if (linked.exitCode !== 0 || linked.timedOut) {
+                throw new Error(
+                  "runner_remote_provider_artifact_incompatible: preinstalled provider pack could not be linked",
+                );
+              }
+              activeRemoteProviderPackRoot = stagedRemoteProviderPackRoot;
+              await input.onLog?.(
+                "stderr",
+                "[paperclip-runner] using manifest-matched provider pack from the sandbox image\n",
+              );
+            } catch {
+              preinstalledProviderPack = null;
+            }
+          }
+          return preinstalledProviderPack !== null;
+        },
+        stageAndVerify: async () => {
+          if (!remoteCommandRunner.syncIn) {
+            throw new Error(
+              "runner_remote_provider_artifact_incompatible: this remote transport cannot stage a provider pack; preinstall the exact manifest-matched pack",
+            );
+          }
+          const escapedPackRoot = stagedRemoteProviderPackRoot.replaceAll(
             "'",
             "'\\''",
           );
-          const escapedTarget = stagedRemoteProviderPackRoot.replaceAll(
-            "'",
-            "'\\''",
-          );
-          const escapedParent = posix
-            .dirname(stagedRemoteProviderPackRoot)
-            .replaceAll("'", "'\\''");
-          const linked = await remoteCommandRunner.execute({
+          const cleared = await remoteCommandRunner.execute({
             command: "sh",
-            args: [
-              "-c",
-              `umask 077; mkdir -p '${escapedParent}' && rm -rf '${escapedTarget}' && ln -s '${escapedSource}' '${escapedTarget}'`,
-            ],
+            args: ["-c", `rm -rf '${escapedPackRoot}'`],
             cwd: remoteTarget.remoteCwd,
             bypassSession: true,
             timeoutMs: 10_000,
           });
-          if (linked.exitCode !== 0 || linked.timedOut) {
+          if (cleared.exitCode !== 0 || cleared.timedOut) {
             throw new Error(
-              "runner_remote_provider_artifact_incompatible: preinstalled provider pack could not be linked",
+              "runner_remote_provider_artifact_incompatible: stale provider pack could not be replaced",
             );
           }
+          await stageRemoteRunnerDirectory({
+            target: remoteTarget,
+            runner: remoteCommandRunner,
+            sourcePath: configuredProviderPackRoot,
+            targetPath: stagedRemoteProviderPackRoot,
+            mode: 0o700,
+          });
+          await measureNativeRunnerSpan(input.trace, "provider_pack.verify", () =>
+            verifyRemoteProviderPack(stagedRemoteProviderPackRoot),
+          );
           activeRemoteProviderPackRoot = stagedRemoteProviderPackRoot;
-          await input.onLog?.(
-            "stderr",
-            "[paperclip-runner] using manifest-matched provider pack from the sandbox image\n",
-          );
-        } catch {
-          preinstalledProviderPack = null;
-        }
-      }
-      if (!preinstalledProviderPack) {
-        if (!remoteCommandRunner.syncIn) {
-          throw new Error(
-            "runner_remote_provider_artifact_incompatible: this remote transport cannot stage a provider pack; preinstall the exact manifest-matched pack",
-          );
-        }
-        const escapedPackRoot = stagedRemoteProviderPackRoot.replaceAll(
-          "'",
-          "'\\''",
+        },
+      });
+      activeRemoteProviderPackRoot = stagedRemoteProviderPackRoot;
+      if (packSource === "staged") {
+        await input.onLog?.(
+          "stderr",
+          "[paperclip-runner] reusing manifest-matched provider pack from the workspace\n",
         );
-        const cleared = await remoteCommandRunner.execute({
-          command: "sh",
-          args: ["-c", `rm -rf '${escapedPackRoot}'`],
-          cwd: remoteTarget.remoteCwd,
-          bypassSession: true,
-          timeoutMs: 10_000,
-        });
-        if (cleared.exitCode !== 0 || cleared.timedOut) {
-          throw new Error(
-            "runner_remote_provider_artifact_incompatible: stale provider pack could not be replaced",
-          );
-        }
-        await stageRemoteRunnerDirectory({
-          target: remoteTarget,
-          runner: remoteCommandRunner,
-          sourcePath: configuredProviderPackRoot,
-          targetPath: stagedRemoteProviderPackRoot,
-          mode: 0o700,
-        });
-        await measureNativeRunnerSpan(input.trace, "provider_pack.verify", () =>
-          verifyRemoteProviderPack(stagedRemoteProviderPackRoot),
-        );
-        activeRemoteProviderPackRoot = stagedRemoteProviderPackRoot;
       }
     }
     remotePrepared = true;
@@ -11271,6 +11505,7 @@ async function createRunnerdBackendWithinSessionClaim(
     ),
     codexTransportFactory: (recoveryContext) =>
       createRunnerdCodexTransport({
+        onSpawn: input.onSpawn,
         provider:
           input.execution.provider.kind === "codex"
             ? "codex"
@@ -11726,10 +11961,20 @@ async function createRunnerdBackendWithinSessionClaim(
   const wrapManagedSession = (session: NativeSession): NativeSession => {
     if (!input.managedAiCredentialHome || input.execution.provider.kind !== "codex") return session;
     const close = session.close.bind(session);
+    const detach = session.detachControllerForRestart?.bind(session);
+    let detachedForRestart = false;
+    if (detach) {
+      session.detachControllerForRestart = async () => {
+        // Ownership is relinquished before the asynchronous detach completes.
+        // Late close finalizers must leave the live runner's credentials alone.
+        detachedForRestart = true;
+        await detach();
+      };
+    }
     let copied = false;
     session.close = async (closeInput) => {
       await close(closeInput);
-      if (copied) return;
+      if (detachedForRestart || copied) return;
       copied = true;
       const remoteAuth = remoteRunnerFilesystemRoot ? posix.join(remoteRunnerFilesystemRoot, "codex-home", "auth.json") : null;
       const localAuth = join(root, "codex-home", "auth.json");

server/src/services/native-runtime/remote-provider-pack.test.ts

@@ -0,0 +1,39 @@
+import { describe, expect, it, vi } from "vitest";
+import { prepareVerifiedRemoteProviderPack } from "./remote-provider-pack.js";
+
+describe("remote provider pack reuse", () => {
+  it("keeps a verified staged pack without consulting the image or uploading again", async () => {
+    const usePreinstalled = vi.fn();
+    const stageAndVerify = vi.fn();
+    expect(await prepareVerifiedRemoteProviderPack({
+      verifyStaged: async () => {}, usePreinstalled, stageAndVerify,
+    })).toBe("staged");
+    expect(usePreinstalled).not.toHaveBeenCalled();
+    expect(stageAndVerify).not.toHaveBeenCalled();
+  });
+
+  it.each(["missing pack", "manifest mismatch", "artifact hash mismatch"])(
+    "replaces a %s with a verified image pack",
+    async (reason) => {
+      const stageAndVerify = vi.fn();
+      expect(await prepareVerifiedRemoteProviderPack({
+        verifyStaged: async () => { throw new Error(reason); },
+        usePreinstalled: async () => true,
+        stageAndVerify,
+      })).toBe("preinstalled");
+      expect(stageAndVerify).not.toHaveBeenCalled();
+    },
+  );
+
+  it("uploads when neither existing pack is valid and fails closed if upload verification fails", async () => {
+    const stageAndVerify = vi.fn().mockRejectedValue(new Error("artifact hash mismatch"));
+    const input = {
+      verifyStaged: async () => { throw new Error("stale staged pack"); },
+      usePreinstalled: async () => false,
+      stageAndVerify,
+    };
+    await expect(prepareVerifiedRemoteProviderPack(input)).rejects.toThrow("artifact hash mismatch");
+    stageAndVerify.mockResolvedValue(undefined);
+    await expect(prepareVerifiedRemoteProviderPack(input)).resolves.toBe("uploaded");
+  });
+});

server/src/services/native-runtime/remote-provider-pack.ts

@@ -0,0 +1,16 @@
+/** Reuse only a pack that passes the same full verification as a new upload. */
+export async function prepareVerifiedRemoteProviderPack(input: {
+  verifyStaged: () => Promise<void>;
+  usePreinstalled: () => Promise<boolean>;
+  stageAndVerify: () => Promise<void>;
+}): Promise<"staged" | "preinstalled" | "uploaded"> {
+  try {
+    await input.verifyStaged();
+    return "staged";
+  } catch {
+    // Missing, stale, or modified packs are never executable cache hits.
+  }
+  if (await input.usePreinstalled()) return "preinstalled";
+  await input.stageAndVerify();
+  return "uploaded";
+}

server/src/services/native-runtime/stopped-codex-turn.test.ts

@@ -0,0 +1,54 @@
+import { describe, expect, it } from "vitest";
+import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js";
+const meta = { type: "session_meta", payload: { id: "thread", cwd: "/workspace" } };
+const start = { type: "event_msg", payload: { type: "task_started", turn_id: "turn" } };
+const context = { type: "turn_context", payload: { turn_id: "turn", cwd: "/workspace" } };
+const answer = { type: "response_item", payload: { type: "message", role: "assistant" } };
+const stop = { type: "event_msg", payload: { type: "turn_aborted", turn_id: "turn", reason: "interrupted" } };
+const check = (rows: unknown[]) => stoppedCodexTurnIsTextOnly({ rows, threadId: "thread", turnId: "turn", cwd: "/workspace" });
+describe("stopped Codex turn inventory", () => {
+  it("accepts an exactly bound, closed text-only turn", () => {
+    expect(check([meta, start, context, answer, stop])).toBe(true);
+  });
+  it("does not treat partial output or an unrelated abort as containment", () => {
+    expect(check([meta, start, context, answer])).toBe(false);
+    expect(check([meta, start, context, { ...stop, payload: { ...stop.payload, turn_id: "other" } }])).toBe(false);
+  });
+  it.each(["function_call", "custom_tool_call", "web_search_call", "unknown_future_action"])("refuses unverified %s outcomes", type => {
+    expect(check([meta, start, context, { type: "response_item", payload: { type } }, stop])).toBe(false);
+  });
+  it("refuses later work, duplicate starts, and changed session identity", () => {
+    expect(check([meta, start, context, stop, answer])).toBe(false);
+    expect(check([meta, start, context, start, stop])).toBe(false);
+    expect(check([{ ...meta, payload: { ...meta.payload, id: "other" } }, start, context, stop])).toBe(false);
+  });
+  it("does not replay completed actions in earlier turns", () => {
+    expect(check([meta, { type: "response_item", payload: { type: "custom_tool_call" } }, start, context, answer, stop])).toBe(true);
+  });
+  const completed = { callId: "finish-id", input: { summary: "ready" } };
+  const completionRows = (source: string) => [meta, start, context,
+    { type: "response_item", payload: { type: "custom_tool_call", name: "exec", call_id: "script", input: source } },
+    { type: "event_msg", payload: { type: "item_completed", thread_id: "thread", turn_id: "turn",
+      item: { type: "DynamicToolCall", tool: "paperclip_finish", id: completed.callId, arguments: completed.input, status: "completed", success: true } } },
+    { type: "response_item", payload: { type: "custom_tool_call_output", call_id: "script", output: [] } }, answer, stop];
+  const checkCompletion = (source: string, calls = [completed]) => stoppedCodexTurnIsTextOnly({
+    rows: completionRows(source), threadId: "thread", turnId: "turn", cwd: "/workspace", completedTaskControlCalls: calls,
+  });
+  it("preserves completion bookkeeping with an exact accepted receipt", () => {
+    expect(checkCompletion('const r = await tools.paperclip_finish({summary: "ready"}); text(r);')).toBe(true);
+    expect(checkCompletion('const r = await tools.paperclip_finish({summary: "ready"}); text(r);', [])).toBe(false);
+    expect(checkCompletion('const r = await tools.paperclip_finish({summary: "different"}); text(r);')).toBe(false);
+  });
+  it.each([
+    'await tools.send_email({}); const r = await tools.paperclip_finish({summary: "ready"}); text(r);',
+    'const r = await tools.paperclip_finish({summary: tools.write_file()}); text(r);',
+    'const r = await tools.paperclip_finish({get summary() { return "ready"; }}); text(r);',
+    'const r = await tools.paperclip_finish({...external, summary: "ready"}); text(r);',
+    'const r = await tools.paperclip_finish({summary: `ready`}); text(r);',
+    'const r = await tools["paperclip_finish"]({summary: "ready"}); text(r);',
+    'const r = await tools.paperclip_finish({summary: "ready"}); tools.send_email(r);',
+    'const r = await tools.paperclip_finish({__proto__: {summary: "ready"}}); text(r);',
+  ])("refuses unverified execution hidden in completion code %s", source => {
+    expect(checkCompletion(source)).toBe(false);
+  });
+});

server/src/services/native-runtime/stopped-codex-turn.ts

@@ -0,0 +1,141 @@
+import { parse } from "acorn";
+import { canonicalNativeJson } from "./canonical.js";
+
+const record = (value: unknown): Record<string, unknown> =>
+  value && typeof value === "object" && !Array.isArray(value)
+    ? value as Record<string, unknown> : {};
+
+export interface CompletedTaskControlCall { callId: string; input: unknown }
+
+// Interpret only JSON literals in one completion call. Never evaluate provider
+// JavaScript, and never infer safety merely from a script containing a tool name.
+function completionScriptInput(source: unknown): unknown {
+  if (typeof source !== "string" || source.length > 65536) throw new Error("script unavailable");
+  const program = parse(source, { ecmaVersion: 2022, sourceType: "module" }) as unknown as Record<string, unknown>;
+  const body = program.body as Record<string, unknown>[];
+  if (body.length !== 2) throw new Error("unknown script");
+  const first = body[0]!;
+  const declarations = first.declarations as Record<string, unknown>[] | undefined;
+  if (first.type !== "VariableDeclaration" || first.kind !== "const" || declarations?.length !== 1) throw new Error("unknown declaration");
+  const declaration = declarations[0]!;
+  const binding = record(declaration.id);
+  const awaited = record(declaration.init);
+  const call = record(awaited.argument);
+  const callee = record(call.callee);
+  const output = record(body[1]!.expression);
+  const args = call.arguments as unknown[] | undefined;
+  const outputArgs = output.arguments as unknown[] | undefined;
+  if (binding.type !== "Identifier" || awaited.type !== "AwaitExpression" || call.type !== "CallExpression" ||
+      callee.type !== "MemberExpression" || callee.computed || callee.optional || call.optional ||
+      record(callee.object).type !== "Identifier" || record(callee.object).name !== "tools" ||
+      record(callee.property).name !== "paperclip_finish" || args?.length !== 1 ||
+      body[1]!.type !== "ExpressionStatement" || output.type !== "CallExpression" || output.optional ||
+      record(output.callee).type !== "Identifier" || record(output.callee).name !== "text" ||
+      outputArgs?.length !== 1 || record(outputArgs[0]).type !== "Identifier" || record(outputArgs[0]).name !== binding.name)
+    throw new Error("unknown effect");
+  let nodes = 0;
+  const literal = (value: unknown, depth = 0): unknown => {
+    if (++nodes > 4096 || depth > 32) throw new Error("literal too large");
+    const node = record(value);
+    if (node.type === "Literal" && (node.value === null || ["string", "boolean", "number"].includes(typeof node.value)) && !node.regex && !node.bigint)
+      return node.value;
+    if (node.type === "ArrayExpression") return (node.elements as unknown[]).map(v => literal(v, depth + 1));
+    if (node.type !== "ObjectExpression") throw new Error("nonliteral input");
+    const result: Record<string, unknown> = Object.create(null);
+    for (const raw of node.properties as unknown[]) {
+      const property = record(raw), key = record(property.key);
+      const name = key.type === "Identifier" ? key.name : key.type === "Literal" ? key.value : null;
+      if (property.type !== "Property" || property.kind !== "init" || property.method || property.computed || property.shorthand ||
+          typeof name !== "string" || ["__proto__", "constructor", "prototype"].includes(name) || name in result)
+        throw new Error("unknown property");
+      result[name] = literal(property.value, depth + 1);
+    }
+    return result;
+  };
+  return literal(args[0]);
+}
+
+/** A closed, text-only turn can be continued without replaying an unknown action.
+ * This is deliberately a closed inventory, not a search for known bad tools.
+ * The caller must independently authenticate the session and contain its processes.
+ */
+export function stoppedCodexTurnIsTextOnly(input: {
+  rows: unknown[];
+  threadId: string;
+  turnId: string;
+  cwd: string;
+  completedTaskControlCalls?: CompletedTaskControlCall[];
+}): boolean {
+  const rows = input.rows.map(record);
+  const meta = rows[0];
+  if (!input.threadId || !input.turnId || meta?.type !== "session_meta" ||
+      record(meta.payload).id !== input.threadId || record(meta.payload).cwd !== input.cwd)
+    return false;
+  const starts = rows.flatMap((row, index) => row.type === "event_msg" &&
+    record(row.payload).type === "task_started" && record(row.payload).turn_id === input.turnId
+    ? [index] : []);
+  if (starts.length !== 1) return false;
+  const turn = rows.slice(starts[0]!);
+  let contextSeen = false;
+  let aborted = false;
+  const completedCalls = input.completedTaskControlCalls ?? [];
+  const scripts = new Set<string>();
+  const seenCalls = new Set<string>();
+  const outputs = new Set<string>();
+  for (let index = 0; index < turn.length; index++) {
+    const row = turn[index]!;
+    const payload = record(row.payload);
+    if (aborted) return false;
+    switch (row.type) {
+      case "turn_context":
+        if (contextSeen || payload.turn_id !== input.turnId || payload.cwd !== input.cwd) return false;
+        contextSeen = true;
+        break;
+      case "response_item":
+        if (payload.type === "custom_tool_call") {
+          try {
+            if (payload.name !== "exec" || typeof payload.call_id !== "string" || scripts.has(payload.call_id) ||
+                !completedCalls.some(call => canonicalNativeJson(call.input) === canonicalNativeJson(completionScriptInput(payload.input)))) return false;
+          } catch { return false; }
+          scripts.add(payload.call_id as string);
+          break;
+        }
+        if (payload.type === "custom_tool_call_output") {
+          if (typeof payload.call_id !== "string" || !scripts.has(payload.call_id) || outputs.has(payload.call_id)) return false;
+          outputs.add(payload.call_id);
+          break;
+        }
+        if (!["message", "reasoning"].includes(String(payload.type))) return false;
+        break;
+      case "world_state":
+      case "token_usage_record":
+        break;
+      case "event_msg":
+        switch (payload.type) {
+          case "task_started":
+            if (index !== 0 || payload.turn_id !== input.turnId) return false;
+            break;
+          case "turn_aborted":
+            if (payload.turn_id !== input.turnId || payload.reason !== "interrupted") return false;
+            aborted = true;
+            break;
+          case "token_count":
+            break;
+          case "item_completed":
+            if (payload.thread_id !== input.threadId || payload.turn_id !== input.turnId) return false;
+            if (record(payload.item).type === "DynamicToolCall") {
+              const item = record(payload.item);
+              if (item.tool !== "paperclip_finish" || item.status !== "completed" || item.success !== true ||
+                  typeof item.id !== "string" || seenCalls.has(item.id) ||
+                  !completedCalls.some(call => call.callId === item.id && canonicalNativeJson(call.input) === canonicalNativeJson(item.arguments))) return false;
+              seenCalls.add(item.id);
+            } else if (!["AgentMessage", "UserMessage", "Reasoning"].includes(String(record(payload.item).type))) return false;
+            break;
+          default: return false;
+        }
+        break;
+      default: return false;
+    }
+  }
+  return contextSeen && aborted && scripts.size === outputs.size && scripts.size === completedCalls.length && seenCalls.size === completedCalls.length;
+}

server/src/services/tool-access.ts

@@ -2776,10 +2776,18 @@ function healthFailureHttpStatus(failure: {
 }): number {
   if (failure.status === "missing_secret") return 422;
   if (failure.code === "composio_api_key_rejected") return 422;
+  if (failure.code === "tool_connection_transport_unsupported") return 422;
   if (failure.code.endsWith("_endpoint_rejected")) return 422;
   return 502;
 }
 
+function unsupportedToolConnectionTransport() {
+  return unprocessable(
+    "This connection has no supported tool integration. Add a supported account or MCP connection from Connectors.",
+    { code: "tool_connection_transport_unsupported" },
+  );
+}
+
 function sanitizeHttpFailure(error: unknown): {
   status: ToolConnectionHealthStatus;
   message: string;
@@ -2797,6 +2805,9 @@ function sanitizeHttpFailure(error: unknown): {
   }
   if (error instanceof HttpError) {
     const code = asRecord(error.details).code;
+    if (code === "tool_connection_transport_unsupported") {
+      return { status: "error", message: error.message, code };
+    }
     if (code === "composio_connected_account_inactive") {
       return { status: "degraded", message: error.message, code };
     }
@@ -7428,6 +7439,9 @@ export function toolAccessService(
       await validateComposioConnection(connection);
       return [];
     }
+    if (connection.transport !== "local_stdio") {
+      throw unsupportedToolConnectionTransport();
+    }
     await resolveCredentialHeaders(connection);
     return localTools(connection);
   }
@@ -7579,9 +7593,11 @@ export function toolAccessService(
         await remoteTools(connection, credentialHeaders, actor);
       } else if (isComposioConnection(connection)) {
         await validateComposioConnection(connection);
-      } else {
+      } else if (connection.transport === "local_stdio") {
         await resolveCredentialHeaders(connection);
         await stdioTemplateId(connection.companyId, connection.config);
+      } else {
+        throw unsupportedToolConnectionTransport();
       }
       const updated = await updateConnectionHealth(
         connection,