fix(recovery): own held interrupted preparations as an ownership kind (Phase 2 review)

Replace the startup-only re-check with the legacy_interrupted_preparation_hold
ownership kind: the shared reconciler settles it only after proveOwnerGone proves
the run's recorded controller gone (a missing stamp or one from another process
space stays held), retries it on every pass once its lease settles, and the settle
resolves the hold exactly once under the existing row locks. Only holds the system
recorded are listed (no owner-gone attestation, user continuation or assigned
owner), and the automatic settlement's activity entry names the owner proof.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

server/src/__tests__/interrupted-preparation-hold-settle.test.ts

@@ -1,5 +1,6 @@
+import { spawn } from "node:child_process";
 import { randomUUID } from "node:crypto";
-import { eq } from "drizzle-orm";
+import { and, eq } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
 import {
   activityLog,
@@ -14,23 +15,32 @@ import {
   issues,
 } from "@paperclipai/db";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { currentProcessSpaceHash } from "../services/controller-process-identity.js";
+import { settleHeldInterruptedPreparation } from "../services/interrupted-preparation-hold-settlement.js";
 import { legacyControllerBootId } from "../services/legacy-controller-lease.js";
-import { heartbeatService } from "../services/heartbeat.js";
-import { settleHeldInterruptedPreparations } from "../services/legacy-execution-recovery.js";
+import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js";
 
 const support = await getEmbeddedPostgresTestSupport();
 const suite = support.supported ? describe : describe.skip;
+const KIND = "legacy_interrupted_preparation_hold";
+
+async function exitedPid(): Promise<number> {
+  const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" });
+  const pid = child.pid!;
+  await new Promise<void>(resolve => { child.once("exit", () => resolve()); child.kill("SIGKILL"); });
+  return pid;
+}
 
 /**
  * The K2-during-preparation shape from Daytona warm wave w-20261108T142906-s9501
  * (t05, run c99b3a2e): a scheduled retry was 2 s into preparation when a
  * graceful shutdown interrupted it. The draining process recorded a
- * preserve-without-replay hold, because at that moment its own controller was
- * still live and the sandbox lease was still being cleaned up. The lease was
- * released with a confirmed stop moments later, and the next process booted,
- * but nothing re-judged the hold, so the task waited for a manual Retry.
+ * preserve-without-replay hold, because its own controller was still live and
+ * the sandbox lease was still cleaning up. The lease was released with a
+ * confirmed stop moments later and the next process booted, but nothing
+ * re-judged the hold, so the task waited for a manual Retry.
  */
-suite("held interrupted preparations after a restart", () => {
+suite("held interrupted preparations on the shared ownership path", () => {
   let db!: ReturnType<typeof createDb>;
   let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
   let companyId = "", agentId = "", environmentId = "";
@@ -58,15 +68,24 @@ suite("held interrupted preparations after a restart", () => {
 
   afterAll(async () => { await temporary?.cleanup(); });
 
-  async function interruptedPreparation(options: { controllerBootId?: string; leaseReleased?: boolean } = {}) {
+  /** The draining controller, now exited, in this PID space. */
+  async function deadController(overrides: Record<string, unknown> = {}) {
+    const bootId = randomUUID();
+    return { bootId, process: { bootId, pid: await exitedPid(), hostHash: currentProcessSpaceHash(),
+      processStartedAt: new Date(Date.now() - 60_000).toISOString(), ...overrides } as Record<string, unknown> | null };
+  }
+
+  async function interruptedPreparation(options: { controller?: { bootId: string; process: Record<string, unknown> | null };
+    leaseReleased?: boolean; holdEvidence?: Record<string, unknown> } = {}) {
     const issueId = randomUUID(), runId = randomUUID(), leaseId = randomUUID(), providerLeaseId = `sandbox-${leaseId}`;
     const finishedAt = new Date(Date.now() - 10_000);
+    const controller = options.controller ?? await deadController();
     await db.insert(issues).values({ id: issueId, companyId, title: "Warm follow-up", status: "in_progress",
       assigneeAgentId: agentId });
     await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "automation",
       status: "interrupted", errorCode: "server_shutdown_interrupted", runtimeMode: "legacy", executionStage: "preparing",
       startedAt: new Date(finishedAt.getTime() - 2_000), finishedAt, nextEventSeq: 1,
-      controllerBootId: options.controllerBootId ?? randomUUID(),
+      controllerBootId: controller.bootId, controllerProcess: controller.process,
       scheduledRetryAttempt: 1, scheduledRetryReason: "transient_failure",
       contextSnapshot: { issueId },
       runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
@@ -84,50 +103,63 @@ suite("held interrupted preparations after a restart", () => {
       fingerprint: `legacy-execution:${runId}`, status: "resolved", outcome: "blocked", ownerType: "board",
       nextAction: "Inspect the stopped provider and recorded actions.", resolvedAt: finishedAt,
       evidence: { runId, automaticRecovery: { policy: "preserve_without_replay_v1", runId, replay: "blocked",
-        actionOutcome: "unknown", recordedAt: finishedAt.toISOString() } } }).returning();
+        actionOutcome: "unknown", recordedAt: finishedAt.toISOString() }, ...options.holdEvidence } }).returning();
     return { issueId, runId, holdId: hold!.id };
   }
 
   const hold = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!;
-  const runRow = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!;
+  const pass = async () => (await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: true }))
+    .kinds.find(entry => entry.kind === KIND);
 
-  it("replays a preparation interrupted by a restart once durable proof shows no provider work started", async () => {
+  it("replays a preparation a restart interrupted once its controller is proven gone", async () => {
     const f = await interruptedPreparation();
-    await settleHeldInterruptedPreparations(db);
+    expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 });
     const settled = await hold(f.holdId);
     expect(settled).toMatchObject({ status: "resolved", outcome: "false_positive" });
     expect(settled.evidence).toMatchObject({ automaticRecovery: { replay: "allowed" },
       bootstrapPreparationContinuation: "pending", bootstrapPreparationSettledAt: expect.any(String) });
-    expect((await runRow(f.runId)).resultJson).toMatchObject({
-      executionRecovery: { kind: "bootstrap", providerWorkStarted: false } });
-  });
-
-  it("orphan recovery after a restart settles the hold without a user Retry", async () => {
-    const f = await interruptedPreparation();
-    await heartbeatService(db).reapOrphanedRuns();
-    expect(await hold(f.holdId)).toMatchObject({ status: "resolved", outcome: "false_positive" });
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]!.resultJson)
+      .toMatchObject({ executionRecovery: { kind: "bootstrap", providerWorkStarted: false } });
+    const [audit] = await db.select().from(activityLog).where(and(eq(activityLog.runId, f.runId),
+      eq(activityLog.action, "issue.recovery_action_resolved")));
+    expect(audit?.details).toMatchObject({ reason: "native_bootstrap_stopped_before_provider", ownerProof: "pid_dead" });
   });
 
   it.each([
-    ["provider work may have started", async (f: { runId: string }) => {
+    ["provider work may have started", async () => {
+      const f = await interruptedPreparation();
       await db.insert(heartbeatRunEvents).values({ companyId, runId: f.runId, agentId, seq: 1, eventType: "turn.started",
         stream: "system", level: "info", message: "turn started" });
+      return f;
     }],
-    ["the controller that owned it is still this process", async (f: { runId: string }) => {
-      await db.update(heartbeatRuns).set({ controllerBootId: legacyControllerBootId }).where(eq(heartbeatRuns.id, f.runId));
-    }],
+    ["its controller stamp is from another process space", async () =>
+      interruptedPreparation({ controller: await deadController({ hostHash: "other-container" }) })],
+    ["it has no controller stamp", async () =>
+      interruptedPreparation({ controller: { bootId: randomUUID(), process: null } })],
+    ["its controller is this process", async () => interruptedPreparation({ controller: { bootId: legacyControllerBootId,
+      process: { bootId: legacyControllerBootId, pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null } } })],
+    ["its sandbox lease is still cleaning up", async () => interruptedPreparation({ leaseReleased: false })],
+    ["an operator confirmed the owner gone on the hold", async () =>
+      interruptedPreparation({ holdEvidence: { ownerGoneAttestation: { actorId: "board-user" } } })],
+    ["the user already continued", async () =>
+      interruptedPreparation({ holdEvidence: { explicitUserContinuation: { actorId: "board-user" } } })],
   ] as const)("keeps the hold when %s", async (_label, arrange) => {
-    const f = await interruptedPreparation();
-    await arrange(f);
+    const f = await arrange();
     const before = await hold(f.holdId);
-    await settleHeldInterruptedPreparations(db);
+    expect((await pass())?.settled ?? 0).toBe(0);
     expect(await hold(f.holdId)).toEqual(before);
   });
 
-  it("keeps the hold while the sandbox lease is still being cleaned up", async () => {
-    const f = await interruptedPreparation({ leaseReleased: false });
-    const before = await hold(f.holdId);
-    await settleHeldInterruptedPreparations(db);
-    expect(await hold(f.holdId)).toEqual(before);
+  it("settles exactly once when two processes judge the same hold", async () => {
+    const f = await interruptedPreparation();
+    const record = { recordId: f.runId, companyId, runId: f.runId, issueId: f.issueId, owner: { kind: "none" as const },
+      updatedAt: null };
+    const context = { now: new Date(), proof: { verdict: "gone" as const, reason: "pid_dead" as const } };
+    const outcomes = await Promise.all([settleHeldInterruptedPreparation(db, record, context),
+      settleHeldInterruptedPreparation(db, record, context)]);
+    expect(outcomes.filter(outcome => outcome.settled)).toHaveLength(1);
+    const resolutions = await db.select().from(activityLog).where(and(eq(activityLog.runId, f.runId),
+      eq(activityLog.action, "issue.recovery_action_resolved")));
+    expect(resolutions).toHaveLength(1);
   });
 });

server/src/__tests__/ownership-reconciliation-report.test.ts

@@ -143,7 +143,7 @@ describe("report-only ownership reconciliation", () => {
     expect(new Set(names).size).toBe(names.length);
     expect(names.length).toBeGreaterThanOrEqual(40);
     expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id))
-      .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]);
+      .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "legacy_interrupted_preparation_hold", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]);
     expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true);
   });
 

server/src/services/heartbeat/recovery.ts

@@ -47,7 +47,7 @@ import {
 import { remoteTerminationReceipt } from "../remote-execution-termination.js";
 import { readEnvironmentStopFailure, type PluginEnvironmentStopFailure } from "@paperclipai/plugin-sdk";
 import { runUsedConversationAdapter } from "../conversation-continuation.js";
-import { legacyExecutionNeedsReconciliationWithEvidence, settleHeldInterruptedPreparations, terminalizeLegacyExecution } from "../legacy-execution-recovery.js";
+import { legacyExecutionNeedsReconciliationWithEvidence, terminalizeLegacyExecution } from "../legacy-execution-recovery.js";
 import { reconcileRetainedWorkspaces } from "../workspace-recovery.js";
 import { randomUUID } from "node:crypto";
 import os from "node:os";
@@ -2864,11 +2864,6 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
     });
     if (!opts?.skipRetainedNativeSessionCleanup) scheduleRetainedNativeSessionCleanup();
     await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments);
-    // A shutdown can hold a run it interrupted in preparation before the proof
-    // that no provider work started exists; once it does, replay that run.
-    await settleHeldInterruptedPreparations(db).catch(error => {
-      logger.warn({ err: error }, "failed to re-judge held interrupted preparations");
-    });
     if (!opts?.skipNativeStatusWakeIntents) await dispatchPendingNativeStatusWakeups().catch((error) => {
       logger.warn(
         { err: error },

server/src/services/interrupted-preparation-hold-settlement.ts

@@ -0,0 +1,62 @@
+import { and, eq, inArray, isNull, sql } from "drizzle-orm";
+import { heartbeatRuns, type Db } from "@paperclipai/db";
+import { isInterruptedNativePreparation } from "./cancelled-native-startup.js";
+import { LEGACY_RECOVERY_CAUSE, settleInterruptedNativeBootstrap } from "./legacy-execution-recovery.js";
+import type { OwnerProof } from "./owner-proof.js";
+import type { InProgressRecord, SettleOutcome } from "./recovery/ownership-registry.js";
+import { OWNER_GONE_ATTESTATION_KEY } from "./recovery/owner-gone-attestation.js";
+
+/**
+ * `legacy_interrupted_preparation_hold`: a run a graceful shutdown interrupted
+ * while it was still preparing, held with `preserve_without_replay_v1`.
+ *
+ * The draining process terminalizes the run before the no-provider-work proof
+ * can hold: its own controller is still live and its sandbox lease may still
+ * be cleaning up. Once the controller is proven gone and every lease is
+ * released with a confirmed stop, the existing proof decides, and a run with
+ * no provider execution is settled and replayed in its own uncharged lane.
+ *
+ * Only holds the system recorded itself are listed: an operator decision (an
+ * owner-gone attestation, an explicit user continuation, an assigned owner) is
+ * never reopened. A run with a lease still cleaning up is not listed yet; the
+ * next pass lists it once the lease settles.
+ */
+export function heldInterruptedPreparationWhere() {
+  return and(inArray(heartbeatRuns.status, ["interrupted", "failed"]), eq(heartbeatRuns.runtimeMode, "legacy"),
+    isNull(heartbeatRuns.runtimeModeResolvedAt), eq(heartbeatRuns.executionStage, "preparing"),
+    sql`exists (select 1 from issue_recovery_actions hold where hold.company_id = ${heartbeatRuns.companyId}
+      and hold.cause = ${LEGACY_RECOVERY_CAUSE} and hold.fingerprint = 'legacy-execution:' || ${heartbeatRuns.id}::text
+      and hold.evidence->>'runId' = ${heartbeatRuns.id}::text
+      and hold.status = 'resolved' and hold.outcome = 'blocked' and hold.owner_user_id is null
+      and hold.evidence->'automaticRecovery'->>'policy' = 'preserve_without_replay_v1'
+      and hold.evidence->'automaticRecovery'->>'replay' = 'blocked'
+      and not (hold.evidence ?| array['executionReconciliation', ${OWNER_GONE_ATTESTATION_KEY}, 'explicitUserContinuation']))`,
+    sql`not exists (select 1 from environment_leases lease where lease.company_id = ${heartbeatRuns.companyId}
+      and lease.heartbeat_run_id = ${heartbeatRuns.id}
+      and (lease.released_at is null or lease.status = 'pending_cleanup' or lease.cleanup_status = 'failed'))`);
+}
+
+/**
+ * Settle of the kind. The shared reconciler calls it only after
+ * `proveOwnerGone` proved the run's recorded controller gone; a run without a
+ * controller stamp, or one from another process space, is never settled. It
+ * re-checks the existing durable proof (no coordinator, result or execution
+ * events; leases released with stop receipts; workspace repair when needed)
+ * and resolves the hold through `settleInterruptedNativeBootstrap`, whose row
+ * locks make the transition, and the continuation it queues, happen once.
+ */
+export async function settleHeldInterruptedPreparation(db: Db, record: InProgressRecord,
+  context: { now: Date; proof: OwnerProof }): Promise<SettleOutcome> {
+  if (!record.companyId) return { settled: false, reason: "run_has_no_company" };
+  const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, record.recordId),
+    eq(heartbeatRuns.companyId, record.companyId), heldInterruptedPreparationWhere()));
+  if (!run) return { settled: false, reason: "no_longer_a_held_preparation" };
+  if (!await isInterruptedNativePreparation(db, run)) {
+    return { settled: false, reason: "provider_work_not_disproved", watch: false };
+  }
+  let resolved = 0;
+  await settleInterruptedNativeBootstrap(db, { run, providerDispatchStarted: false, proof: context.proof.reason,
+    onHoldsResolved: count => { resolved = count; } });
+  return resolved > 0 ? { settled: true, reason: context.proof.reason }
+    : { settled: false, reason: "hold_changed_since_proof" };
+}

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

@@ -328,7 +328,11 @@ export async function terminalizeLegacyExecution(input: {
 }
 
 export async function settleInterruptedNativeBootstrap(
-  db: Db, input: { run: Run; providerDispatchStarted: boolean },
+  db: Db, input: { run: Run; providerDispatchStarted: boolean;
+    /** Owner proof that authorized this settle, recorded in the activity log. */
+    proof?: string;
+    /** Called inside the transaction with the number of holds this call resolved. */
+    onHoldsResolved?: (count: number) => void },
 ): Promise<Run | null> {
   if (input.providerDispatchStarted) return null;
   const issueId = input.run.contextSnapshot?.issueId;
@@ -386,42 +390,14 @@ export async function settleInterruptedNativeBootstrap(
     }).where(inArray(issueRecoveryActions.id, holds.map(hold => hold.id))).returning({ id: issueRecoveryActions.id }) : [];
     if (resolved.length) await logActivity(tx as unknown as Db, { companyId: run.companyId, actorType: "system", actorId: "system",
       action: "issue.recovery_action_resolved", entityType: "issue", entityId: task.id, agentId: run.agentId, runId: run.id,
-      details: { reason: "native_bootstrap_stopped_before_provider", recoveryActionIds: resolved.map(r => r.id) },
+      details: { reason: "native_bootstrap_stopped_before_provider", recoveryActionIds: resolved.map(r => r.id),
+        ...(input.proof ? { ownerProof: input.proof } : {}) },
     });
+    input.onHoldsResolved?.(resolved.length);
     return settled;
   });
 }
 
-/**
- * Re-judges preparation holds once their proof can exist. A graceful shutdown
- * terminalizes a run still in preparation inside the draining process: its own
- * controller is live and its sandbox lease may still be cleaning up, so the
- * no-provider-work proof (`isInterruptedNativePreparation`) cannot hold yet and
- * the run is held for reconciliation. Once the owning process is gone and the
- * leases are released with confirmed stops, that proof decides: a run with no
- * provider execution is settled and its preparation is replayed in its own
- * uncharged lane. Anything with possible provider work, or without proof, stays
- * held. Recovery runs this after every restart.
- */
-export async function settleHeldInterruptedPreparations(db: Db, options: { limit?: number } = {}): Promise<number> {
-  const candidates = await db.select().from(heartbeatRuns).where(and(
-    inArray(heartbeatRuns.status, ["interrupted", "failed"]), eq(heartbeatRuns.runtimeMode, "legacy"),
-    isNull(heartbeatRuns.runtimeModeResolvedAt), eq(heartbeatRuns.executionStage, "preparing"),
-    sql`exists (select 1 from issue_recovery_actions hold where hold.company_id = ${heartbeatRuns.companyId}
-      and hold.cause = ${LEGACY_RECOVERY_CAUSE} and hold.fingerprint = 'legacy-execution:' || ${heartbeatRuns.id}::text
-      and hold.evidence->>'runId' = ${heartbeatRuns.id}::text and not (hold.evidence ? 'executionReconciliation')
-      and (hold.status in ('active', 'escalated') or (hold.status = 'resolved' and hold.outcome = 'blocked'
-        and hold.evidence->'automaticRecovery'->>'policy' = 'preserve_without_replay_v1'
-        and hold.evidence->'automaticRecovery'->>'replay' = 'blocked')))`,
-  )).orderBy(heartbeatRuns.finishedAt).limit(options.limit ?? 50);
-  let settled = 0;
-  for (const run of candidates) {
-    if (!await isInterruptedNativePreparation(db, run)) continue;
-    if (await settleInterruptedNativeBootstrap(db, { run, providerDispatchStarted: false })) settled += 1;
-  }
-  return settled;
-}
-
 /** The source stays discoverable until normal admission records a successor.
  * Known admission gates wait for eligibility events; unresolved writes retain
  * the pending delivery deadline. Concurrent successful delivery never regresses. */

server/src/services/recovery/ownership-registry.ts

@@ -46,6 +46,7 @@ import { nativeControllerOwnerStamp, runProcessOwnerStamp } from "../native-runt
 import { settleIssueLock } from "./issue-lock-settlement.js";
 import { settleAgentLifecycleLease } from "./agent-lifecycle-lease-settlement.js";
 import { legacyControllerOwnerStamp, settleLegacyControllerLease } from "../legacy-controller-lease.js";
+import { heldInterruptedPreparationWhere, settleHeldInterruptedPreparation } from "../interrupted-preparation-hold-settlement.js";
 import {
   nativeLeaseHolderStamp,
   readNativeLeaseHolder,
@@ -971,6 +972,19 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [
     },
     settle: settleLegacyControllerLease,
   },
+  {
+    // A preparation a shutdown interrupted, held before its no-provider-work
+    // proof could exist. Its owner is the recorded controller process: the
+    // shared prover must prove it gone (a missing stamp, or one from another
+    // host or PID space, is unprovable and stays held). The settle re-checks
+    // the durable proof and replays the preparation exactly once.
+    ...runKind("legacy_interrupted_preparation_hold", "Interrupted preparation held for replay", "ownership",
+      heldInterruptedPreparationWhere(), row => {
+        const stamp = legacyControllerOwnerStamp(row);
+        return stamp ? { kind: "process", stamp } : { kind: "none" };
+      }),
+    settle: settleHeldInterruptedPreparation,
+  },
   runKind("detached_process_marker", "Detached-process marker", "ownership",
     and(eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.errorCode, "process_detached")),
     row => row.processPid ? { kind: "process", stamp: runProcessStamp(row, "detached_process_marker") } : { kind: "none" },