Merge remote-tracking branch 'origin/fix/replay-preparation-interruption' into measure/remote-warm-2

* origin/fix/replay-preparation-interruption:
  fix(recovery): actually replay a settled interrupted preparation
  docs(recovery): keep the interrupted-preparation kind after the teardown kinds
  fix(recovery): own held interrupted preparations as an ownership kind (Phase 2 review)
  fix(recovery): replay a preparation a restart interrupted once proof shows no provider work started

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

@@ -0,0 +1,183 @@
+import { spawn } from "node:child_process";
+import { randomUUID } from "node:crypto";
+import { and, eq } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
+import {
+  activityLog,
+  agents,
+  authUsers,
+  companyMemberships,
+  companies,
+  createDb,
+  environmentLeases,
+  environments,
+  heartbeatRunEvents,
+  heartbeatRuns,
+  issueRecoveryActions,
+  issues,
+} from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { currentProcessSpaceHash } from "../services/controller-process-identity.js";
+import { heartbeatService } from "../services/heartbeat.js";
+import { settleHeldInterruptedPreparation } from "../services/interrupted-preparation-hold-settlement.js";
+import { legacyControllerBootId } from "../services/legacy-controller-lease.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 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 on the shared ownership path", () => {
+  let db!: ReturnType<typeof createDb>;
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+  let companyId = "", agentId = "", environmentId = "";
+
+  beforeAll(async () => {
+    temporary = await startEmbeddedPostgresTestDatabase("paperclip-interrupted-preparation-");
+    db = createDb(temporary.connectionString);
+    companyId = randomUUID(); agentId = randomUUID(); environmentId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "IPH",
+      requireBoardApprovalForNewAgents: false });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle",
+      adapterType: "paperclip_runner", adapterConfig: {}, runtimeConfig: {}, permissions: {} });
+    await db.insert(environments).values({ id: environmentId, name: "Sandbox", driver: "sandbox", status: "active",
+      config: { provider: "fake" } });
+    await db.insert(authUsers).values({ id: "local-board", name: "Local operator", email: "local-board@example.invalid",
+      createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing();
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "local-board", membershipRole: "owner" });
+  }, 30_000);
+
+  afterEach(async () => {
+    await db.delete(activityLog);
+    await db.delete(issueRecoveryActions);
+    await db.delete(environmentLeases);
+    await db.delete(heartbeatRunEvents);
+    await db.delete(heartbeatRuns);
+    await db.delete(issues);
+  });
+
+  afterAll(async () => { await temporary?.cleanup(); });
+
+  /** 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();
+    // Terminalization projected the task to blocked and kept a receipt so the
+    // settle can restore exactly that projection.
+    await db.insert(issues).values({ id: issueId, companyId, title: "Warm follow-up", status: "blocked",
+      assigneeAgentId: agentId, statusVersion: 2 });
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "automation",
+      responsibleUserId: "local-board", issueId,
+      status: "interrupted", errorCode: "server_shutdown_interrupted", runtimeMode: "legacy", executionStage: "preparing",
+      startedAt: new Date(finishedAt.getTime() - 2_000), finishedAt, nextEventSeq: 1,
+      controllerBootId: controller.bootId, controllerProcess: controller.process,
+      scheduledRetryAttempt: 1, scheduledRetryReason: "transient_failure",
+      contextSnapshot: { issueId },
+      runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
+      resultJson: { stopReason: "interrupted", cancellation: { reason: "Server shutdown interrupted execution",
+        source: "shutdown", expected: true, initiator: { type: "system" }, recordedAt: finishedAt.toISOString() } } });
+    const released = options.leaseReleased !== false;
+    await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, issueId, heartbeatRunId: runId,
+      status: released ? "released" : "pending_cleanup", cleanupStatus: released ? "success" : "failed",
+      leasePolicy: "reuse_by_environment", provider: "fake", providerLeaseId,
+      releasedAt: released ? new Date(finishedAt.getTime() + 2_000) : null,
+      metadata: released ? { remoteExecutionTermination: { schema: "paperclip.remote-termination.v1", companyId, runId,
+        leaseId, provider: "fake", providerLeaseId, state: "stopped", confirmedAt: finishedAt.toISOString() } } : {} });
+    const [hold] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
+      kind: "execution_reconciliation", cause: "legacy_execution_requires_reconciliation",
+      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() },
+        nativeBootstrapFailureBlock: { runId, previousStatus: "in_progress", statusVersion: 2 },
+        ...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 pass = async () => (await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: true }))
+    .kinds.find(entry => entry.kind === KIND);
+
+  it("replays a preparation a restart interrupted once its controller is proven gone", async () => {
+    // The heartbeat service registers the resume path that schedules the replay.
+    heartbeatService(db);
+    const f = await interruptedPreparation();
+    expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 });
+    // The task is unblocked and an uncharged interrupted-preparation successor exists.
+    expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "in_progress" });
+    const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.runId));
+    expect(successors).toHaveLength(1);
+    expect(successors[0]).toMatchObject({ scheduledRetryReason: "interrupted_preparation" });
+    const settled = await hold(f.holdId);
+    expect(settled).toMatchObject({ status: "resolved", outcome: "false_positive" });
+    expect(settled.evidence).toMatchObject({ automaticRecovery: { replay: "allowed" },
+      bootstrapPreparationContinuation: "delivered", bootstrapPreparationSettledAt: expect.any(String) });
+    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 () => {
+      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;
+    }],
+    ["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 arrange();
+    const before = await hold(f.holdId);
+    expect((await pass())?.settled ?? 0).toBe(0);
+    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

@@ -144,7 +144,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.ts

@@ -1,3 +1,4 @@
+import { registerHeldPreparationResumer } from "./interrupted-preparation-hold-settlement.js";
 import { resolveProviderApiPlacement } from "./native-runtime/provider-api-placement.js";
 import type { StopInvocationOptions } from "./heartbeat/run-control.js";
 import { isContextCompactionRun } from "./context-compaction.js";
@@ -1530,6 +1531,12 @@ export function heartbeatService(
   function resumeRemoteStopComments(...args: Parameters<typeof runControl.resumeRemoteStopComments>) {
     return runControl.resumeRemoteStopComments(...args);
   }
+  // The interrupted-preparation ownership kind settles a hold, then replays the
+  // preparation through this same resume path (bounded, uncharged retry).
+  registerHeldPreparationResumer(async ({ companyId, runId }) => {
+    const run = await getRun(runId);
+    if (run && run.companyId === companyId) await resumeRemoteStopComments(run);
+  });
 
   function cancelRunInternal(...args: Parameters<typeof runControl.cancelRunInternal>) {
     return runControl.cancelRunInternal(...args);

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

@@ -0,0 +1,84 @@
+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 { logger } from "../middleware/logger.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'))`);
+}
+
+/** Continues a settled preparation through the ordinary resume path
+ * (`resumeRemoteStopComments`): it schedules the uncharged
+ * `interrupted_preparation` successor and records the continuation. The
+ * heartbeat service registers it; it owns the scheduling dependencies. */
+type HeldPreparationResumer = (input: { companyId: string; runId: string }) => Promise<void>;
+let heldPreparationResumer: HeldPreparationResumer | null = null;
+
+export function registerHeldPreparationResumer(resumer: HeldPreparationResumer): void {
+  heldPreparationResumer = resumer;
+}
+
+/**
+ * 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; } });
+  if (resolved === 0) return { settled: false, reason: "hold_changed_since_proof" };
+  // Only the call that resolved the hold starts the replay, so it starts once.
+  // Without a registered resumer (no heartbeat service in this process), the
+  // pending continuation stays recorded for the next resume of this run.
+  if (heldPreparationResumer) {
+    await heldPreparationResumer({ companyId: record.companyId, runId: run.id }).catch(error => {
+      // The hold is settled either way; its pending continuation stays recorded
+      // for the next resume of this run (a wake, a cleanup completion).
+      logger.warn({ err: error, runId: run.id }, "settled interrupted preparation could not schedule its replay");
+    });
+  }
+  return { settled: true, reason: context.proof.reason };
+}

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,8 +390,10 @@ 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;
   });
 }

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,21 @@ 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.
+    // Keep this entry after the environment teardown kinds: a lease they settle
+    // in a pass is released before this kind lists in that same pass.
+    ...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" },