PaperClipAI/server/src/services/heartbeat/recovery.test.ts

import { randomUUID } from "node:crypto";
import { spawn } from "node:child_process";
import { once } from "node:events";
import type { Db } from "@paperclipai/db";
import { activityLog, agents, applyPendingMigrations, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environments, environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db";
import { eq } from "drizzle-orm";
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../environment-runtime.js";
import { environmentService } from "../environments.js";
import { createHeartbeatRecovery, type HeartbeatRecoveryDependencies } from "./recovery.js";
import { createHeartbeatRunState } from "./run-state.js";
import type { HotRestartIntent } from "../hot-restart.js";
import type { NativeRestartRecoveryClaim, NativeRestartRecoveryDisposition } from "../native-runtime/index.js";
import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "../native-runtime/native-runner-ownership.js";
import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_RESTART_RECOVERY_FAILED } from "../native-runtime/native-restart-recovery.js";
import { startNativeSessionExecutionLeaseRenewal } from "../native-runtime/native-execution-lease.js";
import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js";
import { adapterExecutionControls, createAdapterExecutionControl } from "../adapter-execution-control.js";
import { logger } from "../../middleware/logger.js";
import { listExpiredNativeRunnerAuthorities } from "../native-runtime/native-runner-checkpoint.js";
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../delivery-work-notifications.js";
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
import { hasRemoteTerminationReceipt, remoteTerminationReceipt } from "../remote-execution-termination.js";
import { persistHeartbeatRunProcessMetadata } from "./run-lifecycle.js";
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
import { hasNativeLocalProcessStop, PROCESS_START_REQUESTED, releaseSettledNativeRunProcess } from "../native-local-process-stop.js";

type Run = typeof heartbeatRuns.$inferSelect;

const native = vi.hoisted(() => ({
  closeIdle: vi.fn(async () => ({ closed: 0, failed: 0 })),
  detach: vi.fn(async () => 0),
  finalizations: vi.fn(async (_db: Db, _ids?: string[], _options?: unknown) => undefined),
  cleanups: vi.fn<() => Promise<void>>(async () => undefined),
  drainMaintenance: vi.fn<() => Promise<void>>(async () => undefined),
  claims: vi.fn<typeof claimNativeRestartRecoveries>(async () => []),
}));
const hotRestart = vi.hoisted(() => ({
  read: vi.fn<() => Promise<HotRestartIntent | null>>(async () => null),
}));
vi.mock("../hot-restart.js", async (importOriginal) => ({
  ...await importOriginal<typeof import("../hot-restart.js")>(),
  readHotRestartIntent: hotRestart.read,
}));
vi.mock("../native-runtime/index.js", async (importOriginal) => ({
  ...await importOriginal<typeof import("../native-runtime/index.js")>(),
  closeIdleWarmNativeSessionsForRestart: native.closeIdle,
  detachNativeSessionsForRestart: native.detach,
  reconcileNativeFinalizations: native.finalizations,
  claimNativeRestartRecoveries: native.claims,
}));

vi.mock("../native-runtime/retained-native-session-cleanup.js", async (importOriginal) => ({
  ...await importOriginal<typeof import("../native-runtime/retained-native-session-cleanup.js")>(),
  reconcileRetainedNativeSessionCleanups: native.cleanups,
}));

function callbacks(db: Db) {
  return {
    enterShutdown: vi.fn(),
    getRun: vi.fn<HeartbeatRecoveryDependencies["getRun"]>(async () => null),
    getRunWorkspaceSettlement: vi.fn(createHeartbeatRunState(db).getRunWorkspaceSettlement),
    getAgent: vi.fn<HeartbeatRecoveryDependencies["getAgent"]>(async () => null),
    appendRunEvent: vi.fn<HeartbeatRecoveryDependencies["appendRunEvent"]>(async () => undefined),
    setRunStatusIfRunning: vi.fn<HeartbeatRecoveryDependencies["setRunStatusIfRunning"]>(async () => ({ updated: false, run: null })),
    setRunStatusFromLive: vi.fn<HeartbeatRecoveryDependencies["setRunStatusFromLive"]>(async () => ({ updated: false, run: null })),
    setRunStatus: vi.fn<HeartbeatRecoveryDependencies["setRunStatus"]>(async () => null),
    setWakeupStatus: vi.fn(async () => undefined),
    releaseIssueExecutionAndPromote: vi.fn(async () => undefined),
    finalizeAgentStatus: vi.fn(async () => undefined),
    environmentRuntime: environmentRuntimeService(db),
    environmentsSvc: environmentService(db),
    instructionCopies: {
      recoverStopped: vi.fn(async () => 0),
      recoverCaptured: vi.fn(async () => 0),
    },
    runtimeEnv: {},
    activeRunExecutions: new Set<string>(),
    activeRunExecutionPromises: new Set<Promise<void>>(),
    scheduleBoundedRetryForRun: vi.fn<HeartbeatRecoveryDependencies["scheduleBoundedRetryForRun"]>(async () => ({ outcome: "retry_exhausted", retryReason: "transient_failure", attempt: 1, maxAttempts: 0 })),
    scheduleInteractionContinuationInfrastructureRetryIfEligible: vi.fn<HeartbeatRecoveryDependencies["scheduleInteractionContinuationInfrastructureRetryIfEligible"]>(async () => null),
    timerClaimWasFirstHeartbeat: vi.fn(() => undefined),
    executeRun: vi.fn<HeartbeatRecoveryDependencies["executeRun"]>(async () => undefined),
    scheduleNativeSessionResumeDispatch: vi.fn(),
    cancelHeartbeatNativeRun: vi.fn(async () => undefined),
    terminateHeartbeatRunProcess: vi.fn(async () => undefined),
    mergeRunStopMetadataForAgent: vi.fn<HeartbeatRecoveryDependencies["mergeRunStopMetadataForAgent"]>((_agent, _outcome, options) => options?.resultJson ?? null),
    classifyAndPersistRunLiveness: vi.fn<HeartbeatRecoveryDependencies["classifyAndPersistRunLiveness"]>(async run => run),
    releaseEnvironmentLeasesForRun: vi.fn<HeartbeatRecoveryDependencies["releaseEnvironmentLeasesForRun"]>(async () => undefined),
    acknowledgeRemoteStop: vi.fn(async () => undefined),
    resumeRemoteStopComments: vi.fn(async () => undefined),
    dispatchPendingNativeStatusWakeups: vi.fn(async () => undefined),
    cancelRunInternal: vi.fn(async () => undefined),
    startNextQueuedRunForAgent: vi.fn(async () => undefined),
  } satisfies HeartbeatRecoveryDependencies;
}

function guardedDatabase() {
  const access = vi.fn(() => { throw new Error("Unexpected recovery database access"); });
  return { db: new Proxy({}, { get: access }) as Db, access };
}

it("does not query checkpoint recovery without an expired active run", async () => {
  const { db, access } = guardedDatabase();
  expect(await listExpiredNativeRunnerAuthorities(db, new Date(), [])).toEqual([]);
  expect(access).not.toHaveBeenCalled();
});

beforeEach(() => {
  vi.clearAllMocks();
  hotRestart.read.mockResolvedValue(null);
  native.closeIdle.mockResolvedValue({ closed: 0, failed: 0 });
  native.finalizations.mockResolvedValue(undefined);
  native.cleanups.mockResolvedValue(undefined);
  native.claims.mockResolvedValue([]);
});

describe("heartbeat recovery module callbacks", () => {
  it("constructs without database access, shutdown, or background work", () => {
    const database = guardedDatabase();
    const deps = callbacks(database.db);
    createHeartbeatRecovery(database.db, deps);
    expect(database.access).not.toHaveBeenCalled();
    for (const callback of Object.values(deps)) if (typeof callback === "function") expect(callback).not.toHaveBeenCalled();
    for (const callback of Object.values(native)) expect(callback).not.toHaveBeenCalled();
  });

  it.each(["not_requested", "read_error"] as const)("enters shutdown before checkpointing for %s", async (mode) => {
    const database = guardedDatabase();
    const deps = callbacks(database.db);
    if (mode === "read_error") hotRestart.read.mockRejectedValueOnce(new Error("unreadable intent"));
    const recovery = createHeartbeatRecovery(database.db, deps);
    expect(await recovery.prepareHotRestartShutdown("SIGTERM")).toEqual({ mode, skipDrain: false, activeRunIds: [] });
    expect(native.closeIdle).toHaveBeenCalledAfter(deps.enterShutdown);
    expect(hotRestart.read).toHaveBeenCalledAfter(native.closeIdle);
    expect(database.access).not.toHaveBeenCalled();
  });

  it("does not adopt or query runs without a restart intent", async () => {
    const database = guardedDatabase();
    expect(await createHeartbeatRecovery(database.db, callbacks(database.db)).reconcileHotRestartAdoption()).toEqual({
      mode: "not_requested", adoptedRunIds: [], finalizedWhileDownRunIds: [], lostRunIds: [], skippedRunIds: [],
    });
    expect(database.access).not.toHaveBeenCalled();
  });

  it("does not broaden an empty selective shutdown drain to every run", async () => {
    const database = guardedDatabase();
    const deps = callbacks(database.db);
    expect(await createHeartbeatRecovery(database.db, deps).drainRunningRunsForShutdown("SIGINT", new Date(), [])).toEqual({
      interrupted: 0, interruptedRunIds: [], retryRunIds: [], restartSuspendedRunIds: [],
    });
    expect(database.access).not.toHaveBeenCalled();
    expect(deps.cancelHeartbeatNativeRun).not.toHaveBeenCalled();
    expect(deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
  });

  it.each(["running", "failed"] as const)("keeps a %s ownership hold when a status race wins", async (status) => {
    const database = guardedDatabase();
    const deps = callbacks(database.db);
    const run = { id: "run", runtimeMode: "native", status } as Run;
    const winner = { ...run, status: "cancelled" };
    deps.setRunStatusFromLive.mockResolvedValue({ updated: false, run: winner });
    expect(await createHeartbeatRecovery(database.db, deps).markNativeOwnershipUnverified(run, { reason: "live_process_identifier" })).toBe(winner);
    expect(deps.setRunStatusFromLive).toHaveBeenCalledWith("run", status, [status], expect.objectContaining({ errorCode: NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE }));
    expect(deps.appendRunEvent).not.toHaveBeenCalled();
    expect(database.access).not.toHaveBeenCalled();
  });

  it.each(["adopted_runner_authentication_timeout"] as const)("persists a %s ownership hold before its event without terminating or retrying", async reason => {
    const database = guardedDatabase();
    const deps = callbacks(database.db);
    const run = { id: "run", runtimeMode: "native", status: "running" } as Run;
    const blocked = { ...run, nativePhase: "terminal_failure", errorCode: NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE };
    deps.setRunStatusFromLive.mockResolvedValue({ updated: true, run: blocked });
    expect(await createHeartbeatRecovery(database.db, deps).markNativeOwnershipUnverified(run, { reason })).toBe(blocked);
    expect(deps.setRunStatusFromLive).toHaveBeenCalledWith("run", "running", ["running"], expect.objectContaining({ nativePhase: "terminal_failure" }));
    expect(deps.appendRunEvent).toHaveBeenCalledAfter(deps.setRunStatusFromLive);
    expect(deps.cancelHeartbeatNativeRun).not.toHaveBeenCalled();
    expect(deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
    expect(deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
    expect(database.access).not.toHaveBeenCalled();
  });
});

const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RECOVERY_TEST_DATABASE_URL");
const support = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
const describePostgres = support.supported ? describe : describe.skip;
if (!support.supported) console.warn(`Skipping recovery module database tests: ${support.reason}`);

describePostgres("heartbeat recovery module ownership", () => {
  let db: Db;
  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
  beforeAll(async () => {
    if (externalUrl) await applyPendingMigrations(externalUrl);
    else database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-recovery-module-");
    db = createDb(externalUrl ?? database.connectionString);
  }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
  afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });

  async function seedRun() {
    const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), issueId = randomUUID();
    await db.insert(companies).values({ id: companyId, name: "Recovery", issuePrefix: `T${companyId.slice(0, 6)}` });
    await db.insert(agents).values({ id: agentId, companyId, name: "Recovery", role: "engineer", adapterType: "process" });
    await db.insert(issues).values({ id: issueId, companyId, title: "Recover this run", status: "in_progress", assigneeAgentId: agentId });
    return (await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, invocationSource: "automation", status: "running", runtimeMode: "legacy" }).returning())[0];
  }

  it("routes an expired pre-capability startup to cleanup before generic orphan ownership holds", async () => {
    const run = await seedRun(), leaseId = randomUUID(), environmentId = randomUUID(), leaseOwner = randomUUID();
    const deadlineAtMs = Date.now() - 1_000, contextId = randomUUID();
    await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, run.agentId));
    await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, run.issueId!));
    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId, nativePhase: "observed",
      runnerInstanceId: randomUUID(), nativeSessionId: randomUUID(), runnerProfileJson: { nativeStartupContext: {
        schema: "paperclip.native-startup-context.v1", companyId: run.companyId, agentId: run.agentId, runId: run.id,
        contextId, deadlineAtMs, leaseId, leaseOwner, controllerGeneration: 1, attempt: 1 } } }).where(eq(heartbeatRuns.id, run.id));
    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!,
      phase: "observed", attempt: 1, leaseOwner, controllerGeneration: 1, leaseExpiresAt: new Date(Date.now() - 1_000),
      controllerBootId: randomUUID(), controllerPid: process.pid, controllerProcessStartedAt: new Date() });
    await db.insert(environments).values({ id: environmentId, name: environmentId, driver: "plugin", config: {} });
    await db.insert(environmentLeases).values({ id: leaseId, companyId: run.companyId, environmentId, heartbeatRunId: run.id,
      issueId: run.issueId, provider: "plugin", providerLeaseId: "early-task", status: "active", metadata: { runnerTaskLease: true } });
    const deps = callbacks(db);
    deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null);
    native.claims.mockImplementationOnce(claimNativeRestartRecoveries);
    try {
      await createHeartbeatRecovery(db, deps).recoverActiveExecutions({ runId: run.id });
      expect(native.claims).toHaveBeenCalledWith(expect.objectContaining({ runIds: [run.id] }));
      const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
      expect(current).toMatchObject({ status: "running", errorCode: "native_startup_cleanup_pending",
        resultJson: { nativeStartupCleanup: { contextId, state: "pending" } } });
      expect(deps.executeRun).not.toHaveBeenCalled();
      expect(deps.setRunStatusFromLive).not.toHaveBeenCalled();
    } finally {
      await db.delete(environmentLeases).where(eq(environmentLeases.id, leaseId));
      await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
      await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, run.id));
      await db.delete(activityLog).where(eq(activityLog.runId, run.id));
      await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, run.issueId!));
      await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
    }
  });

  async function ownedProcess() {
    const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)", `paperclip-recovery-${randomUUID()}`], {
      stdio: "ignore", detached: true, env: { PATH: process.env.PATH },
    });
    await once(child, "spawn");
    return {
      pid: child.pid!,
      async stop() {
        if (child.exitCode !== null || child.signalCode !== null) return;
        const exited = once(child, "exit");
        child.kill("SIGTERM");
        await exited;
      },
      alive() { return process.kill(child.pid!, 0); },
    };
  }

  async function nativeProcessFixture(pid: number, targetKind: "local" | "remote", status: "running" | "failed") {
    const run = await seedRun();
    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId, status,
      nativePhase: status === "failed" ? "retryable_failure" : "running",
      errorCode: status === "failed" ? "native_session_interrupted" : null,
      error: status === "failed" ? "native session timed out after 120000ms" : null,
      finishedAt: status === "failed" ? new Date() : null,
    }).where(eq(heartbeatRuns.id, run.id));
    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!,
      phase: status === "failed" ? "retryable_failure" : "running", nextAttemptAt: new Date(Date.now() - 1) });
    await persistHeartbeatRunProcessMetadata(db, run.id, { pid, processGroupId: pid,
      startedAt: new Date().toISOString(), targetKind });
    const deps = callbacks(db);
    deps.setRunStatusFromLive.mockImplementation(async (id, nextStatus, fromStatuses, patch) => {
      const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id));
      if (!current || !fromStatuses.includes(current.status)) return { updated: false, run: current ?? null };
      const [updated] = await db.update(heartbeatRuns).set({ ...patch, status: nextStatus })
        .where(eq(heartbeatRuns.id, id)).returning();
      return { updated: true, run: updated };
    });
    return {
      run, deps,
      current: async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0],
      async cleanup() {
        await Promise.all([...deps.activeRunExecutionPromises]);
        await db.delete(environmentLeases).where(eq(environmentLeases.heartbeatRunId, run.id));
        await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
        await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, run.id));
        await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
      },
    };
  }

  it.each(["failed", "running"] as const)("does not attribute a remote %s run's PID to a live host process", async status => {
    const child = await ownedProcess();
    const f = await nativeProcessFixture(child.pid, "remote", status);
    try {
      await db.update(agents).set({ adapterType: "process" }).where(eq(agents.id, f.run.agentId));
      const before = await f.current();
      expect(await createHeartbeatRecovery(db, f.deps).reapOrphanedRuns()).toEqual({ reaped: 0, runIds: [] });
      expect(await f.current()).toMatchObject({ status, errorCode: before.errorCode, error: before.error,
        processPid: child.pid, processGroupId: child.pid, processStartedAt: before.processStartedAt });
      expect(f.deps.setRunStatusFromLive).not.toHaveBeenCalled();
      expect(f.deps.executeRun).not.toHaveBeenCalled();
      expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
      expect(await hasNativeLocalProcessStop(db, f.run.companyId, f.run.id)).toBe(false);
      expect(child.alive()).toBe(true);
    } finally { try { await f.cleanup(); } finally { await child.stop(); } }
  });

  async function remoteRetryFixture(pid: number, target: "workspace" | "remote_workspace" | "plugin_task" = "workspace") {
    const fixture = await nativeProcessFixture(pid, "remote", "failed");
    const sessionId = randomUUID();
    const [lease] = await db.insert(environmentLeases).values({ companyId: fixture.run.companyId,
      heartbeatRunId: fixture.run.id, issueId: fixture.run.issueId, status: "active", provider: "daytona",
      providerLeaseId: randomUUID(), metadata: target === "plugin_task" ? { runnerTaskLease: true } : {},
    }).returning();
    await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, fixture.run.agentId));
    await db.update(issues).set({ executionRunId: fixture.run.id }).where(eq(issues.id, fixture.run.issueId!));
    await db.update(heartbeatRuns).set({ nativeSessionId: sessionId,
      ...(target === "remote_workspace" ? { processGroupId: null } : {}), runnerProfileJson: {
      sessionCheckpoint: { identity: { runId: fixture.run.id, companyId: fixture.run.companyId,
        issueId: fixture.run.issueId, agentId: fixture.run.agentId, sessionId }, providerSessionId: "existing-provider" },
      ...(target === "workspace" ? { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared",
        workspaceId: randomUUID(), leaseId: lease.id, providerLeaseId: lease.providerLeaseId, remoteCwd: "/work",
        descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null,
        resourceDisposition: "keep_running" } } : {}),
      ...(target === "remote_workspace" ? { remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1",
        workspaceId: randomUUID(), epoch: 3, leaseId: lease.id, providerLeaseId: lease.providerLeaseId,
        remoteCwd: "/work", state: "prepared", snapshotId: randomUUID() } } : {}),
    } }).where(eq(heartbeatRuns.id, fixture.run.id));
    await db.insert(nativeRunnerCheckpoints).values({ runId: fixture.run.id, companyId: fixture.run.companyId,
      issueId: fixture.run.issueId!, part: "opaque-checkpoint-record", material: { fixture: true } });
    fixture.deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns)
      .where(eq(heartbeatRuns.id, id)))[0] ?? null);
    native.claims.mockImplementation(claimNativeRestartRecoveries);
    return { ...fixture, lease };
  }

  it.each((["workspace", "remote_workspace"] as const).flatMap(target =>
    (["failed", "running"] as const).map(status => ({ target, status })),
  ))("routes a due $target $status retry through verified recovery without local PID inspection", async ({ target, status }) => {
    const child = await ownedProcess();
    const f = await remoteRetryFixture(child.pid, target);
    const processGroupId = target === "remote_workspace" ? null : child.pid;
    const originalKill = process.kill.bind(process);
    const kill = vi.spyOn(process, "kill").mockImplementation(originalKill);
    try {
      await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, f.run.id));
      await db.update(nativeRunFinalizations).set({ nextAttemptAt: null }).where(eq(nativeRunFinalizations.runId, f.run.id));
      const recovery = createHeartbeatRecovery(db, f.deps);
      await recovery.reapOrphanedRuns();
      await Promise.all([...f.deps.activeRunExecutionPromises]);
      expect(f.deps.executeRun).toHaveBeenCalledExactlyOnceWith(f.run.id, expect.objectContaining({
        nativeRestartRecovery: expect.objectContaining({ kind: "reattach_remote_runner", runId: f.run.id,
          remote: { providerLeaseId: f.lease.providerLeaseId, remoteCwd: "/work" } }),
      }));
      expect(native.claims).toHaveBeenCalledWith(expect.objectContaining({ runIds: [f.run.id] }));
      expect(await f.current()).toMatchObject({ status: "running", processPid: child.pid, processGroupId });
      expect(kill.mock.calls.filter(([pid]) => Math.abs(pid as number) === child.pid)).toEqual([]);
      expect(await hasNativeLocalProcessStop(db, f.run.companyId, f.run.id)).toBe(false);
      await recovery.reapOrphanedRuns();
      expect(f.deps.executeRun).toHaveBeenCalledOnce();
      expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
      kill.mockRestore();
      expect(child.alive()).toBe(true);
    } finally { kill.mockRestore(); try { await f.cleanup(); } finally { await child.stop(); } }
  });

  it("routes a due plugin-task retry through its existing remote lease", async () => {
    const child = await ownedProcess();
    const f = await remoteRetryFixture(child.pid, "plugin_task");
    try {
      await createHeartbeatRecovery(db, f.deps).reapOrphanedRuns();
      await Promise.all([...f.deps.activeRunExecutionPromises]);
      expect(f.deps.executeRun).toHaveBeenCalledExactlyOnceWith(f.run.id, expect.objectContaining({
        nativeRestartRecovery: expect.objectContaining({ kind: "reattach_remote_runner", runId: f.run.id,
          remote: { kind: "plugin_task", leaseId: f.lease.id } }),
      }));
      expect(await f.current()).toMatchObject({ processPid: child.pid, processGroupId: child.pid });
      expect(await hasNativeLocalProcessStop(db, f.run.companyId, f.run.id)).toBe(false);
      expect(child.alive()).toBe(true);
    } finally { try { await f.cleanup(); } finally { await child.stop(); } }
  });

  it.each((["workspace", "remote_workspace"] as const).flatMap(target =>
    (["future_retry", "live_owner", "missing_checkpoint", "active_execution"] as const).map(condition => ({ target, condition })),
  ))(
    "does not execute a $target retry without current authority: $condition", async ({ target, condition }) => {
      const child = await ownedProcess();
      const f = await remoteRetryFixture(child.pid, target);
      const processGroupId = target === "remote_workspace" ? null : child.pid;
      try {
        if (condition === "future_retry") await db.update(nativeRunFinalizations)
          .set({ nextAttemptAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.run.id));
        if (condition === "live_owner") await db.update(nativeRunFinalizations)
          .set({ leaseOwner: "other-controller", leaseExpiresAt: new Date(Date.now() + 60_000) })
          .where(eq(nativeRunFinalizations.runId, f.run.id));
        if (condition === "missing_checkpoint") {
          const run = await f.current();
          await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, sessionCheckpoint: null } })
            .where(eq(heartbeatRuns.id, f.run.id));
        }
        if (condition === "active_execution") f.deps.activeRunExecutions.add(f.run.id);
        await createHeartbeatRecovery(db, f.deps).reapOrphanedRuns();
        expect(f.deps.executeRun).not.toHaveBeenCalled();
        expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
        expect(await f.current()).toMatchObject({ processPid: child.pid, processGroupId });
        if (condition === "future_retry") expect(native.claims).not.toHaveBeenCalled();
        if (condition === "missing_checkpoint") {
          const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id));
          expect(coordinator).toMatchObject({ phase: "terminal_failure", recoveryState: "blocked",
            failureCode: "native_restart_recovery_blocked" });
        }
        expect(child.alive()).toBe(true);
      } finally { f.deps.activeRunExecutions.delete(f.run.id); try { await f.cleanup(); } finally { await child.stop(); } }
    },
  );

  it.each(["unknown", "remote"] as const)("exposes expired observed %s ownership without attributing host liveness", async target => {
    const child = await ownedProcess();
    const f = await nativeProcessFixture(child.pid, "remote", "running");
    const originalKill = process.kill.bind(process);
    let restoreKill: (() => void) | undefined;
    try {
      if (target === "unknown") await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id));
      await db.update(heartbeatRuns).set({ nativePhase: null })
        .where(eq(heartbeatRuns.id, f.run.id));
      await db.update(nativeRunFinalizations).set({ phase: "observed", attempt: 2,
        leaseOwner: "prior-native-owner", leaseExpiresAt: new Date(0), nextAttemptAt: null })
        .where(eq(nativeRunFinalizations.runId, f.run.id));
      const before = await f.current();
      const coordinatorBefore = await db.select().from(nativeRunFinalizations)
        .where(eq(nativeRunFinalizations.runId, f.run.id));
      const kill = vi.spyOn(process, "kill").mockImplementation(originalKill);
      restoreKill = () => kill.mockRestore();
      expect(await createHeartbeatRecovery(db, f.deps).reapOrphanedRuns()).toEqual({ reaped: 0, runIds: [] });
      expect(kill.mock.calls.filter(([pid]) => Math.abs(pid as number) === child.pid)).toEqual([]);
      expect(await f.current()).toMatchObject({ status: "running", nativePhase: null,
        processPid: child.pid, processGroupId: child.pid, processStartedAt: before.processStartedAt,
        errorCode: NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE });
      expect(f.deps.appendRunEvent).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({
        payload: { reason: "observed_owner_unverified" },
      }));
      expect(await db.select().from(nativeRunFinalizations)
        .where(eq(nativeRunFinalizations.runId, f.run.id))).toEqual(coordinatorBefore);
      expect(f.deps.executeRun).not.toHaveBeenCalled();
      expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
      kill.mockRestore(); restoreKill = undefined;
      expect(child.alive()).toBe(true);
    } finally { restoreKill?.(); try { await f.cleanup(); } finally { await child.stop(); } }
  });

  it.each(["failed", "running"] as const)("protects a verified local %s run's live process", async status => {
    const child = await ownedProcess();
    const f = await nativeProcessFixture(child.pid, "local", status);
    try {
      await db.update(agents).set({ adapterType: "paperclip_runner", adapterConfig: { driver: "daytona" } })
        .where(eq(agents.id, f.run.agentId));
      await createHeartbeatRecovery(db, f.deps).reapOrphanedRuns();
      expect(await f.current()).toMatchObject({ status, processPid: child.pid, processGroupId: child.pid,
        errorCode: NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE });
      expect(f.deps.appendRunEvent).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({
        payload: { reason: "live_process_identifier", processPidAlive: true, processGroupAlive: true },
      }));
      expect(f.deps.executeRun).not.toHaveBeenCalled();
      expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
      expect(child.alive()).toBe(true);
    } finally { try { await f.cleanup(); } finally { await child.stop(); } }
  });

  it.each(["missing", "new_launch", "wrong_birth", "foreign_host", "wrong_agent", "wrong_pid", "wrong_group", "native_source", "remote_workspace", "remote_lease"] as const)(
    "holds a native PID without local authority: %s", async evidence => {
      const child = await ownedProcess();
      await child.stop();
      const f = await nativeProcessFixture(child.pid, "local", "failed");
      try {
        if (evidence === "missing") await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id));
        if (evidence === "new_launch") await appendHeartbeatRunEvent(db, { companyId: f.run.companyId,
          runId: f.run.id, agentId: f.run.agentId, eventType: PROCESS_START_REQUESTED });
        if (["wrong_birth", "foreign_host", "wrong_agent", "wrong_pid", "wrong_group", "native_source"].includes(evidence)) {
          const [event] = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id));
          const foreignAgentId = randomUUID();
          if (evidence === "wrong_agent") await db.insert(agents).values({ id: foreignAgentId,
            companyId: f.run.companyId, name: "Other process owner", role: "engineer", adapterType: "process" });
          await db.update(heartbeatRunEvents).set({
            ...(evidence === "wrong_agent" ? { agentId: foreignAgentId } : {}),
            ...(evidence === "native_source" ? { sourceEventId: "provider-event" } : {}),
            payload: { ...event.payload, ...(evidence === "wrong_birth"
              ? { processStartedAt: new Date(0).toISOString() } : evidence === "foreign_host" ? { hostname: "foreign-host" }
              : evidence === "wrong_pid" ? { processPid: child.pid + 1 }
              : evidence === "wrong_group" ? { processGroupId: child.pid + 1 } : {}) },
          }).where(eq(heartbeatRunEvents.id, event.id));
        }
        if (evidence === "remote_workspace") await db.update(heartbeatRuns).set({
          runnerProfileJson: { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared",
            workspaceId: randomUUID(), leaseId: randomUUID(), providerLeaseId: randomUUID(), remoteCwd: "/work",
            descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, resourceDisposition: "destroy" } },
        }).where(eq(heartbeatRuns.id, f.run.id));
        if (evidence === "remote_lease") await db.insert(environmentLeases).values({ companyId: f.run.companyId,
          heartbeatRunId: f.run.id, issueId: f.run.issueId, status: "active", provider: "daytona", providerLeaseId: randomUUID() });
        const before = await f.current();
        await createHeartbeatRecovery(db, f.deps).reapOrphanedRuns();
        expect(await f.current()).toMatchObject({ processPid: child.pid, processGroupId: child.pid,
          processStartedAt: before.processStartedAt, errorCode: "native_session_interrupted" });
        expect(f.deps.executeRun).not.toHaveBeenCalled();
        expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
        expect(await hasNativeLocalProcessStop(db, f.run.companyId, f.run.id)).toBe(false);
      } finally { await f.cleanup(); }
    },
  );

  it("clears a verified stopped local retry identity and records its stop", async () => {
    const child = await ownedProcess();
    await child.stop();
    const f = await nativeProcessFixture(child.pid, "local", "failed");
    try {
      await createHeartbeatRecovery(db, f.deps).reapOrphanedRuns();
      expect(await f.current()).toMatchObject({ processPid: null, processGroupId: null, processStartedAt: null });
      expect(await hasNativeLocalProcessStop(db, f.run.companyId, f.run.id)).toBe(true);
      expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled();
    } finally { await f.cleanup(); }
  });

  it("retires a remote identity only with an exact authenticated termination receipt", async () => {
    const child = await ownedProcess();
    const f = await nativeProcessFixture(child.pid, "remote", "failed");
    try {
      await db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { fixture: true } } })
        .where(eq(heartbeatRuns.id, f.run.id));
      await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" })
        .where(eq(nativeRunFinalizations.runId, f.run.id));
      const [lease] = await db.insert(environmentLeases).values({ companyId: f.run.companyId,
        heartbeatRunId: f.run.id, issueId: f.run.issueId, status: "active", provider: "daytona", providerLeaseId: randomUUID() }).returning();
      expect(await releaseSettledNativeRunProcess(db, f.run.id)).toBe("pending");
      const receipt = remoteTerminationReceipt(lease, { state: "destroyed", providerLeaseId: lease.providerLeaseId })!;
      await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date(),
        metadata: { remoteExecutionTermination: { ...receipt, runId: randomUUID() } } }).where(eq(environmentLeases.id, lease.id));
      expect(await releaseSettledNativeRunProcess(db, f.run.id)).toBe("pending");
      expect(await f.current()).toMatchObject({ processPid: child.pid, processGroupId: child.pid });
      await db.update(environmentLeases).set({ metadata: { remoteExecutionTermination: receipt } })
        .where(eq(environmentLeases.id, lease.id));
      expect(await releaseSettledNativeRunProcess(db, f.run.id)).toBe("released");
      expect(await f.current()).toMatchObject({ processPid: null, processGroupId: null, processStartedAt: null });
      expect(await hasNativeLocalProcessStop(db, f.run.companyId, f.run.id)).toBe(false);
      expect(child.alive()).toBe(true);
    } finally { try { await f.cleanup(); } finally { await child.stop(); } }
  });

  it.each(["ephemeral", "reuse_by_environment", "live_owner", "live_executor", "mismatched_source", "finalizing"] as const)(
    "reconstructs missing native preservation during orphan cleanup without replay: %s", async state => {
      const run = await seedRun();
      const [environment] = await db.insert(environments).values({ name: run.id, driver: "sandbox" }).returning();
      const [lease] = await db.insert(environmentLeases).values({ companyId: run.companyId, issueId: run.issueId,
        heartbeatRunId: run.id, environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(),
        status: "active", leasePolicy: state === "reuse_by_environment" ? state : "ephemeral", updatedAt: new Date(0),
        metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: randomUUID(), remoteCwd: "/work" },
      }).returning();
      await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "native", nativeIssueId: run.issueId,
        finishedAt: new Date(), errorCode: "execution_finalization_deadline_exceeded", contextSnapshot: { issueId: run.issueId },
        runnerProfileJson: { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared",
          descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null,
          workspaceId: run.id, leaseId: lease.id, providerLeaseId: lease.providerLeaseId,
          remoteCwd: state === "mismatched_source" ? "/other" : "/work", resourceDisposition: "keep_running" } },
      }).where(eq(heartbeatRuns.id, run.id));
      await db.insert(nativeRunFinalizations).values({ companyId: run.companyId, issueId: run.issueId!, runId: run.id,
        phase: state === "finalizing" ? "workspace_finalizing" : "terminal_failure",
        leaseOwner: state === "live_owner" ? "another-controller" : null });
      const deps = callbacks(db);
      if (state === "live_executor") deps.activeRunExecutions.add(run.id);
      const recovery = createHeartbeatRecovery(db, deps);
      const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.workspaceRecovery, wake);
      const shouldRecover = state === "ephemeral" || state === "reuse_by_environment";
      try {
        expect(await recovery.sweepOrphanedActiveLeases({ backoffMs: 0 })).toEqual({ recovered: shouldRecover ? 1 : 0 });
        const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
        const [retained] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id));
        const records = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!));
        expect(source.status).toBe("failed");
        if (shouldRecover) {
          expect(source.resultJson).toMatchObject({ workspaceRestoreRecovery: { leaseIds: [lease.id] } });
          expect(retained).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure",
            metadata: { sandboxStopAndRetain: { leaseId: lease.id, runId: run.id } } });
          expect(records).toHaveLength(1);
          expect(records[0].evidence).toMatchObject({ automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } });
          expect(wake).toHaveBeenCalledOnce();
          const registered = source.resultJson;
          expect(await recovery.sweepOrphanedActiveLeases({ backoffMs: 0 })).toEqual({ recovered: 0 });
          expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].resultJson).toEqual(registered);
          expect(wake).toHaveBeenCalledOnce();
        } else {
          expect(source.resultJson).toBeNull();
          expect(retained.status).toBe("active");
          expect(records).toHaveLength(0);
          expect(wake).not.toHaveBeenCalled();
        }
        expect(hasRemoteTerminationReceipt(retained)).toBe(false);
        expect(retained.metadata).not.toHaveProperty("remoteExecutionTermination");
        expect(deps.executeRun).not.toHaveBeenCalled();
        expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, run.companyId))).toHaveLength(1);
      } finally {
        unsubscribe();
        await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id));
        await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!));
        await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
        await db.delete(activityLog).where(eq(activityLog.runId, run.id));
        await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
      }
    },
  );

  it.each(["restored", "restore_failed"])("keeps the sandbox available until adapter workspace settlement: %s", async outcome => {
    const run = await seedRun();
    const deps = callbacks(db);
    deps.setRunStatusIfRunning.mockImplementation(async (id, status, patch) => {
      const [updated] = await db.update(heartbeatRuns).set({ ...patch, status }).where(eq(heartbeatRuns.id, id)).returning();
      return { updated: true, run: updated };
    });
    deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null);
    const control = createAdapterExecutionControl();
    adapterExecutionControls.set(run.id, control);
    const drain = createHeartbeatRecovery(db, deps).drainRunningRunsForShutdown("SIGTERM", new Date(), [run.id]);
    try {
      await vi.waitFor(() => expect(control.controller.signal.aborted).toBe(true));
      expect(deps.releaseEnvironmentLeasesForRun).not.toHaveBeenCalled();
      expect(deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
      expect((await deps.getRun(run.id))?.status).toBe("interrupted");
      const resultJson = outcome === "restore_failed" ? { workspaceRestoreFailure: "restore_failed",
        workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [randomUUID()] } }
        : { workspaceRestored: true };
      await db.update(heartbeatRuns).set({ resultJson }).where(eq(heartbeatRuns.id, run.id));
      control.finish();
      await drain;
      expect(deps.releaseEnvironmentLeasesForRun).toHaveBeenCalledOnce();
      expect(deps.classifyAndPersistRunLiveness).toHaveBeenCalledWith(expect.objectContaining({
        resultJson: expect.objectContaining(resultJson) }), expect.anything());
    } finally {
      control.finish();
      adapterExecutionControls.delete(run.id);
      await drain;
      await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
    }
  });

  it("starts all owned shutdown cancellations within one settlement window", async () => {
    const runs = await Promise.all([seedRun(), seedRun()]);
    const controls = runs.map(run => {
      const control = createAdapterExecutionControl(); adapterExecutionControls.set(run.id, control); return control;
    });
    const deps = callbacks(db);
    deps.setRunStatusIfRunning.mockImplementation(async (id, status, patch) => {
      const [updated] = await db.update(heartbeatRuns).set({ ...patch, status }).where(eq(heartbeatRuns.id, id)).returning();
      return { updated: true, run: updated };
    });
    const drain = createHeartbeatRecovery(db, deps).drainRunningRunsForShutdown("SIGTERM", new Date(), runs.map(run => run.id));
    try {
      await vi.waitFor(() => expect(controls.every(control => control.controller.signal.aborted)).toBe(true));
      expect(deps.releaseEnvironmentLeasesForRun).not.toHaveBeenCalled();
    } finally {
      controls.forEach(control => control.finish());
      await drain;
      for (const run of runs) { adapterExecutionControls.delete(run.id); await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); }
    }
  });

  it.each(["ephemeral", "reuse_by_environment"] as const)("protects a %s export from another server and pins it after controller loss", async leasePolicy => {
    const run = await seedRun();
    await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: run.issueId } }).where(eq(heartbeatRuns.id, run.id));
    const [lease] = await db.insert(environmentLeases).values({ companyId: run.companyId, heartbeatRunId: run.id,
      issueId: run.issueId, status: "active", provider: "daytona", providerLeaseId: randomUUID(), leasePolicy,
      metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: "fixture-plugin", remoteCwd: "/home/daytona/workspace" } }).returning();
    try {
      const deadline = new Date(Date.now() + 30_000);
      await beginLegacyShutdownWorkspaceSettlement(db, run, deadline);
      await db.update(heartbeatRuns).set({ status: "interrupted", errorCode: "server_shutdown_interrupted", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run.id));
      const foreignDeps = callbacks(db);
      const foreign = createHeartbeatRecovery(db, foreignDeps);
      expect(await foreign.sweepOrphanedActiveLeases({ backoffMs: 0 })).toMatchObject({ recovered: 0 });
      expect(await environmentRuntimeService(db).releaseRunLeases(run.id)).toEqual([]);
      await db.update(environmentLeases).set({ status: "pending_cleanup" }).where(eq(environmentLeases.id, lease.id));
      expect(await foreign.sweepPendingCleanupLeases({ backoffMs: 0 })).toMatchObject({ destroyed: 0 });
      await db.update(environmentLeases).set({ status: "active" }).where(eq(environmentLeases.id, lease.id));
      let source = (await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0];
      expect(source.status).toBe("active");
      expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))).toHaveLength(0);
      // No local adapter map or timeout handler survives a container SIGKILL.
      expect(await allowLegacyShutdownWorkspaceCleanup(db, source, new Date(deadline.getTime() + 1))).toBe(true);
      source = (await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0];
      expect(source).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure",
        metadata: { legacyShutdownWorkspaceSettlement: { state: "expired" }, sandboxStopAndRetain: { leaseId: lease.id } } });
      expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].resultJson)
        .toMatchObject({ workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { leaseIds: [lease.id] } });
      expect(await foreign.sweepOrphanedActiveLeases({ backoffMs: 0 })).toMatchObject({ recovered: 0 });
    } finally {
      await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id));
      await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!));
      await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
    }
  });

  async function seedShutdownSource() {
    const run = await seedRun();
    const [lease] = await db.insert(environmentLeases).values({ companyId: run.companyId, heartbeatRunId: run.id,
      issueId: run.issueId, status: "active", provider: "daytona", providerLeaseId: randomUUID(), leasePolicy: "ephemeral",
      metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: "fixture-plugin", remoteCwd: "/home/daytona/workspace" } }).returning();
    await beginLegacyShutdownWorkspaceSettlement(db, run, new Date(Date.now() + 30_000));
    return { run, lease };
  }
  async function deleteShutdownSource(f: Awaited<ReturnType<typeof seedShutdownSource>>) {
    await db.delete(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
    await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.run.issueId!));
    await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id));
    adapterExecutionControls.delete(f.run.id);
  }

  it("does not let an elapsed shutdown wait stop a sandbox whose exporter still owns settlement", async () => {
    const f = await seedShutdownSource();
    const control = createAdapterExecutionControl();
    control.controller.abort();
    adapterExecutionControls.set(f.run.id, control);
    const stop = vi.fn(async () => ({ providerLeaseId: f.lease.providerLeaseId, state: "stopped" }));
    const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox",
      resolveCapabilities: async () => classifyEnvironmentCapabilities({}),
      acquireRunLease: async () => { throw new Error("Unexpected acquire"); },
      releaseRunLease: vi.fn(async () => null), retryPendingSandboxTeardown: stop }] });
    try {
      await db.update(heartbeatRuns).set({ status: "interrupted", errorCode: "server_shutdown_interrupted", finishedAt: new Date() })
        .where(eq(heartbeatRuns.id, f.run.id));
      const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
      expect(await allowLegacyShutdownWorkspaceCleanup(db, source, new Date(Date.now() + 60_000))).toBe(false);
      await runtime.releaseRunLeases(f.run.id);
      expect(stop).not.toHaveBeenCalled();
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].status).toBe("active");
      expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.run.issueId!))).toHaveLength(0);
    } finally { control.finish(); await deleteShutdownSource(f); }
  });

  it("renews the durable export fence for other controllers and expires it after owner loss", async () => {
    const f = await seedShutdownSource();
    const [initial] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
    const originalDeadline = (initial.metadata!.legacyShutdownWorkspaceSettlement as { deadlineAt: string }).deadlineAt;
    vi.useFakeTimers({ toFake: ["Date", "setInterval", "clearInterval"] });
    const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(db, f.run);
    try {
      await vi.advanceTimersByTimeAsync(10_000);
      await vi.waitFor(async () => {
        const [current] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
        expect(Date.parse((current.metadata!.legacyShutdownWorkspaceSettlement as { deadlineAt: string }).deadlineAt))
          .toBeGreaterThan(Date.parse(originalDeadline));
      });
      expect(await allowLegacyShutdownWorkspaceCleanup(db, initial, new Date(Date.parse(originalDeadline) + 1))).toBe(false);
      await stopRenewing();
      await vi.advanceTimersByTimeAsync(60_000);
      await db.update(heartbeatRuns).set({ status: "interrupted", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
      await sweepLegacyShutdownWorkspaceSettlements(db);
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0])
        .toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure",
          metadata: { legacyShutdownWorkspaceSettlement: { state: "expired" }, sandboxStopAndRetain: { leaseId: f.lease.id } } });
    } finally { await stopRenewing(); vi.useRealTimers(); await deleteShutdownSource(f); }
  });

  it("bounds a contended settlement renewal without blocking unrelated database work", async () => {
    const f = await seedShutdownSource();
    const warning = vi.spyOn(logger, "warn");
    let unlock!: () => void;
    let locked!: () => void;
    const lockReady = new Promise<void>(resolve => { locked = resolve; });
    const releaseLock = new Promise<void>(resolve => { unlock = resolve; });
    const lock = db.transaction(async tx => {
      await tx.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)).for("update");
      locked();
      await releaseLock;
    });
    await lockReady;
    vi.useFakeTimers({ toFake: ["Date", "setInterval", "clearInterval"] });
    const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(db, f.run);
    try {
      await vi.advanceTimersByTimeAsync(10_000);
      expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))
        .toEqual([{ id: f.run.id }]);
      await stopRenewing();
      expect(warning).toHaveBeenCalledWith({ runId: f.run.id, errorKind: "lock_timeout" },
        "Shutdown workspace settlement lease renewal failed; source remains protected locally");
    } finally {
      unlock();
      await lock;
      await stopRenewing();
      vi.useRealTimers();
      warning.mockRestore();
      await deleteShutdownSource(f);
    }
  });

  it.each([["57014", "statement_cancelled"], ["08006", "connection_failed"], ["private-credential", "unknown"]])(
    "reports renewal failure %s without raw database diagnostics", async (code, errorKind) => {
      const run = { id: randomUUID(), companyId: randomUUID() };
      const privateValue = "postgres://private-credential@example.invalid/database";
      const failure = new Error(privateValue, { cause: { code, query: privateValue, params: [privateValue] } });
      const dbBoundary = { transaction: vi.fn().mockRejectedValue(failure) } as unknown as Db;
      const warning = vi.spyOn(logger, "warn").mockImplementation(() => undefined);
      vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
      const stopRenewing = maintainLegacyShutdownWorkspaceSettlement(dbBoundary, run);
      try {
        await vi.advanceTimersByTimeAsync(10_000);
        await stopRenewing();
        expect(warning).toHaveBeenCalledExactlyOnceWith({ runId: run.id, errorKind },
          "Shutdown workspace settlement lease renewal failed; source remains protected locally");
        expect(JSON.stringify(warning.mock.calls)).not.toContain("private-credential");
      } finally { await stopRenewing(); vi.useRealTimers(); warning.mockRestore(); }
    },
  );

  it("does not turn a stale pending snapshot into a repair hold after settlement", async () => {
    const f = await seedShutdownSource();
    try {
      const old = (await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0];
      await db.update(heartbeatRuns).set({ status: "interrupted", contextSnapshot: { issueId: f.run.issueId }, resultJson: { workspaceRestored: true } }).where(eq(heartbeatRuns.id, f.run.id));
      await finishLegacyShutdownWorkspaceSettlement(db, f.run);
      const single = createDb(externalUrl ?? database.connectionString, { maxConnections: 1 });
      try { expect(await allowLegacyShutdownWorkspaceCleanup(single, old, new Date(Date.now() + 60_000))).toBe(true); }
      finally { await single.$client.end({ timeout: 1 }); }
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0])
        .toMatchObject({ status: "active", leasePolicy: "ephemeral", metadata: { legacyShutdownWorkspaceSettlement: { state: "settled" } } });
      expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.run.issueId!))).toHaveLength(0);
    } finally { await deleteShutdownSource(f); }
  });

  it("allows the aborted owner to stop and retain its command while excluding outside cleanup", async () => {
    const f = await seedShutdownSource();
    const control = createAdapterExecutionControl(); control.controller.abort();
    adapterExecutionControls.set(f.run.id, control);
    const stop = vi.fn(async () => ({ providerLeaseId: f.lease.providerLeaseId, state: "stopped" }));
    const release = vi.fn(async () => null);
    const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", resolveCapabilities: async () => classifyEnvironmentCapabilities({}),
      acquireRunLease: async () => { throw new Error("Unexpected acquire"); }, releaseRunLease: release,
      retryPendingSandboxTeardown: stop }] });
    try {
      await runtime.releaseRunLeases(f.run.id, "released", undefined, "stop_and_retain", false);
      expect(stop).not.toHaveBeenCalled();
      await runtime.releaseRunLeases(f.run.id, "released", undefined, "stop_and_retain", true);
      expect(stop).toHaveBeenCalledOnce(); expect(release).not.toHaveBeenCalled();
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0])
        .toMatchObject({ status: "released", cleanupStatus: "success", metadata: { remoteExecutionTermination: { state: "stopped" },
          sandboxStopAndRetainReceipt: { leaseId: f.lease.id } } });
    } finally { control.finish(); await deleteShutdownSource(f); }
  });

  it.each(["ephemeral", "reuse_by_environment"] as const)("recovers a stopped %s source after its shutdown controller disappears", async leasePolicy => {
    const f = await seedShutdownSource();
    await db.update(environmentLeases).set({ leasePolicy }).where(eq(environmentLeases.id, f.lease.id));
    const control = createAdapterExecutionControl(); control.controller.abort(); adapterExecutionControls.set(f.run.id, control);
    const [environmentRow] = await db.insert(environments).values({ name: "Guarded source", driver: "sandbox" }).returning();
    const environment = (await environmentService(db).getById(environmentRow.id))!;
    const leaseSnapshot = (await environmentService(db).getLeaseById(f.lease.id))!;
    leaseSnapshot.metadata = f.lease.metadata;
    const resume = vi.fn(async () => null); const destroy = vi.fn(async () => null);
    const stop = vi.fn(async () => ({ providerLeaseId: f.lease.providerLeaseId, state: "stopped" }));
    const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", resolveCapabilities: async () => classifyEnvironmentCapabilities({}),
      acquireRunLease: async () => { throw new Error("Unexpected acquire"); }, releaseRunLease: vi.fn(), retryPendingSandboxTeardown: stop, resumeRunLease: resume, destroyRunLease: destroy }] });
    try {
      await runtime.releaseRunLeases(f.run.id, "released", undefined, "stop_and_retain", true);
      adapterExecutionControls.delete(f.run.id); // SIGKILL before export/finally settled.
      await db.update(heartbeatRuns).set({ status: "interrupted", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
      const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
      expect(source.status).toBe("released");
      // An old reuse receipt must not bypass the current allocation's fence.
      expect(await legacyShutdownWorkspaceResourceProtected(db, { provider: source.provider, providerLeaseId: source.providerLeaseId })).toBe(true);
      await sweepLegacyShutdownWorkspaceSettlements(db, new Date(Date.now() + 60_000));
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0])
        .toMatchObject({ status: "released", leasePolicy: "retain_on_failure", metadata: {
          legacyShutdownWorkspaceSettlement: { state: "expired" }, workspaceRestoreRecovery: { runId: f.run.id } } });
      expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].resultJson)
        .toMatchObject({ workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { leaseIds: [f.lease.id] } });
      expect(await legacyShutdownWorkspaceResourceProtected(db, source)).toBe(true);
      // Direct operations and pre-fence snapshots cannot bypass the source pin.
      expect(await runtime.resumeRunLease({ environment, lease: leaseSnapshot })).toBeNull();
      expect(await runtime.destroyRunLease({ environment, lease: leaseSnapshot })).toBeNull();
      expect(resume).not.toHaveBeenCalled(); expect(destroy).not.toHaveBeenCalled();
      expect(stop).toHaveBeenCalledOnce();
    } finally { control.finish(); await deleteShutdownSource(f); await db.delete(environments).where(eq(environments.id, environment.id)); }
  });

  it.each(["ephemeral", "reuse_by_environment"] as const)("recovers failed ordinary cleanup of a fenced %s source", async leasePolicy => {
    const f = await seedShutdownSource();
    const stop = vi.fn(async () => ({ providerLeaseId: f.lease.providerLeaseId, state: "stopped" }));
    const destroy = vi.fn(async () => null);
    const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", resolveCapabilities: async () => classifyEnvironmentCapabilities({}),
      acquireRunLease: async () => { throw new Error("Unexpected acquire"); }, releaseRunLease: destroy, destroyRunLease: destroy,
      retryPendingSandboxTeardown: stop }] });
    try {
      await db.update(heartbeatRuns).set({ status: "interrupted", finishedAt: new Date(), contextSnapshot: { issueId: f.run.issueId } }).where(eq(heartbeatRuns.id, f.run.id));
      await db.update(environmentLeases).set({ status: "pending_cleanup", leasePolicy, cleanupStatus: "failed",
        releasedAt: new Date(), failureReason: "release_cleanup_failed" }).where(eq(environmentLeases.id, f.lease.id));
      const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
      expect(await allowLegacyShutdownWorkspaceCleanup(db, source, new Date(Date.now() + 60_000))).toBe(true);
      const lease = (await environmentService(db).getLeaseById(f.lease.id))!;
      expect(lease).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure", metadata: {
        legacyShutdownWorkspaceSettlement: { state: "expired" }, sandboxStopAndRetain: { leaseId: f.lease.id } } });
      expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].resultJson)
        .toMatchObject({ workspaceRestoreRecovery: { leaseIds: [f.lease.id] } });
      const deps = callbacks(db); deps.environmentRuntime = runtime;
      expect(await createHeartbeatRecovery(db, deps).sweepPendingCleanupLeases({ backoffMs: 0 })).toMatchObject({ destroyed: 1 });
      expect((await environmentService(db).getLeaseById(f.lease.id))!).toMatchObject({ status: "released", cleanupStatus: "success" });
      expect(stop).toHaveBeenCalledOnce(); expect(destroy).not.toHaveBeenCalled();
    } finally { await deleteShutdownSource(f); }
  });

  it("waits for a claimed ordinary cleanup attempt before converting its shutdown source", async () => {
    const f = await seedShutdownSource(); const expiresAt = Date.now() + 120_000;
    try {
      await db.update(heartbeatRuns).set({ status: "interrupted", finishedAt: new Date(), contextSnapshot: { issueId: f.run.issueId } }).where(eq(heartbeatRuns.id, f.run.id));
      const [marked] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
      await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", releasedAt: new Date(),
        metadata: { ...marked.metadata, pendingCleanupInFlight: true, pendingCleanupLeaseExpiresAtMs: expiresAt } }).where(eq(environmentLeases.id, f.lease.id));
      const [source] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id));
      expect(await allowLegacyShutdownWorkspaceCleanup(db, source, new Date(expiresAt - 1))).toBe(false);
      expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].resultJson?.workspaceRestoreRecovery).toBeUndefined();
      expect(await allowLegacyShutdownWorkspaceCleanup(db, source, new Date(expiresAt + 1))).toBe(true);
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0])
        .toMatchObject({ status: "pending_cleanup", metadata: { sandboxStopAndRetain: { leaseId: f.lease.id } } });
    } finally { await deleteShutdownSource(f); }
  });

  it("retries lease release when ordinary completion wins the shutdown status race", async () => {
    const f = await seedShutdownSource(); const control = createAdapterExecutionControl();
    adapterExecutionControls.set(f.run.id, control);
    const deps = callbacks(db);
    deps.setRunStatusIfRunning.mockImplementation(async () => {
      await db.update(heartbeatRuns).set({ status: "succeeded", resultJson: { workspaceRestored: true } }).where(eq(heartbeatRuns.id, f.run.id));
      // The adapter's first release was deferred by the fence; its finally ended.
      control.finish(); return { updated: false, run: null };
    });
    deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null);
    deps.releaseEnvironmentLeasesForRun.mockImplementation(async input => {
      expect(input.status).toBe("succeeded");
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].metadata)
        .toMatchObject({ legacyShutdownWorkspaceSettlement: { state: "settled" } });
      await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() }).where(eq(environmentLeases.id, f.lease.id));
    });
    try {
      await createHeartbeatRecovery(db, deps).drainRunningRunsForShutdown("SIGTERM", new Date(), [f.run.id]);
      expect(deps.releaseEnvironmentLeasesForRun).toHaveBeenCalledOnce();
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].status).toBe("released");
      expect(deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
    } finally { control.finish(); await deleteShutdownSource(f); }
  });

  it("joins a late successful export before cleanup and continuation after the shutdown wait", async () => {
    const run = await seedRun();
    await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: run.issueId } }).where(eq(heartbeatRuns.id, run.id));
    const [lease] = await db.insert(environmentLeases).values({ companyId: run.companyId, heartbeatRunId: run.id,
      issueId: run.issueId, status: "active", provider: "daytona", providerLeaseId: randomUUID(), leasePolicy: "ephemeral",
      metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: "fixture-plugin", remoteCwd: "/home/daytona/workspace" } }).returning();
    await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, run.agentId));
    const deps = callbacks(db);
    deps.setRunStatusIfRunning.mockImplementation(async (id, status, patch) => {
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0].metadata)
        .toMatchObject({ legacyShutdownWorkspaceSettlement: { state: "pending", runId: run.id } });
      const [updated] = await db.update(heartbeatRuns).set({ ...patch, status }).where(eq(heartbeatRuns.id, id)).returning();
      return { updated: true, run: updated };
    });
    deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null);
    deps.releaseEnvironmentLeasesForRun.mockImplementation(async () => {
      expect((await deps.getRun(run.id))?.resultJson).toEqual({ workspaceRestored: true, conversationContinuation: "continue_conversation_v1" });
      expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0]).toMatchObject({
        status: "active", leasePolicy: "ephemeral", metadata: { legacyShutdownWorkspaceSettlement: { state: "settled" } } });
      await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date() }).where(eq(environmentLeases.id, lease.id));
    });
    const control = createAdapterExecutionControl();
    adapterExecutionControls.set(run.id, control);
    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
    const drain = createHeartbeatRecovery(db, deps).drainRunningRunsForShutdown("SIGTERM", new Date(), [run.id]);
    try {
      await vi.waitFor(() => expect(control.controller.signal.aborted).toBe(true));
      await vi.advanceTimersByTimeAsync(30_000);
      expect(await drain).toMatchObject({ interruptedRunIds: [run.id] });
      expect(deps.activeRunExecutionPromises.size).toBe(1);
      expect(deps.releaseEnvironmentLeasesForRun).not.toHaveBeenCalled();
      expect(deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled();
      expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))).toHaveLength(0);
      await db.update(heartbeatRuns).set({ resultJson: { workspaceRestored: true, conversationContinuation: "continue_conversation_v1" } }).where(eq(heartbeatRuns.id, run.id));
      control.finish();
      await Promise.all([...deps.activeRunExecutionPromises]);
      expect(deps.releaseEnvironmentLeasesForRun).toHaveBeenCalledOnce();
      expect(deps.scheduleBoundedRetryForRun).toHaveBeenCalledOnce();
      expect(deps.scheduleBoundedRetryForRun).toHaveBeenCalledAfter(deps.releaseEnvironmentLeasesForRun);
      expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))).toHaveLength(0);
    } finally {
      vi.useRealTimers();
      control.finish();
      adapterExecutionControls.delete(run.id);
      await drain;
      await Promise.all([...deps.activeRunExecutionPromises]);
      await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id));
      await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!));
      await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
    }
  });

  it("uses the shared live-execution set after construction across two factories", async () => {
    const run = await seedRun();
    const deps = callbacks(db);
    const first = createHeartbeatRecovery(db, deps);
    const second = createHeartbeatRecovery(db, deps);
    deps.activeRunExecutions.add(run.id);
    expect(await first.reapOrphanedRuns()).toEqual({ reaped: 0, runIds: [] });
    expect(await second.reapOrphanedRuns()).toEqual({ reaped: 0, runIds: [] });
    expect(deps.setRunStatusFromLive).not.toHaveBeenCalled();
    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("running");
    await Promise.all([...deps.activeRunExecutionPromises]);
    await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
  });

  it("settles persisted native results before claiming restart authority and tracks the exact execution", async () => {
    const deps = callbacks(db);
    const claim: NativeRestartRecoveryClaim = { kind: "bootstrap_incomplete", runId: randomUUID(), leaseOwner: "claimed-owner", controllerGeneration: 3, providerAttempt: 2, restartKind: "hard", recoveryRequestId: null };
    native.claims.mockResolvedValue([claim]);
    let finish!: () => void;
    const execution = new Promise<void>(resolve => { finish = resolve; });
    deps.executeRun.mockReturnValue(execution);
    const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart();
    expect(result.claims).toEqual([claim]);
    expect(native.claims).toHaveBeenCalledAfter(native.finalizations);
    expect(deps.executeRun).toHaveBeenCalledWith(claim.runId, { nativeLeaseOwner: claim.leaseOwner, nativeRestartRecovery: claim });
    expect(deps.activeRunExecutionPromises.size).toBeGreaterThan(0);
    finish();
    await Promise.all([...deps.activeRunExecutionPromises]);
    await vi.waitFor(() => expect(deps.activeRunExecutionPromises.size).toBe(0));
  });

  it.each(["completion", "failure_before_dispatch"] as const)("stops preparation renewal after recovered %s", async outcome => {
    const run = await seedRun();
    const deps = callbacks(db);
    const controller = await currentNativeControllerIdentity();
    const claim: NativeRestartRecoveryClaim = { kind: "reattach_remote_runner", runId: run.id,
      leaseOwner: randomUUID(), controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null,
      remote: { providerLeaseId: randomUUID(), remoteCwd: "/workspace" } };
    const scope = { companyId: run.companyId, issueId: run.issueId!, controller, remote: true };
    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId }).where(eq(heartbeatRuns.id, run.id));
    const expiresAt = new Date(Date.now() + 60_000);
    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, phase: "observed", attempt: 1,
      controllerBootId: controller.bootId, controllerPid: controller.pid,
      controllerProcessStartedAt: controller.processStartedAt,
      leaseOwner: claim.leaseOwner, leaseExpiresAt: expiresAt });
    vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
    native.claims.mockImplementationOnce(async input => {
      await input.onClaim?.(claim, scope);
      return [claim];
    });
    let released!: () => void, finish!: () => void;
    const leaseReleased = new Promise<void>(resolve => { released = resolve; });
    const finalization = new Promise<void>(resolve => { finish = resolve; });
    deps.executeRun.mockImplementationOnce(async (_runId, options) => {
      if (outcome === "failure_before_dispatch") {
        released();
        throw new Error("workspace preparation failed");
      }
      const execution = startNativeSessionExecutionLeaseRenewal({ db, ...scope, runId: run.id,
        leaseOwner: claim.leaseOwner, attempt: claim.providerAttempt,
        onLost: () => native.detach() });
      try {
        // The executor takes its reference before preparation releases its own.
        await options?.onNativeExecutionLeaseAcquired?.();
        await execution.stop();
        await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null })
          .where(eq(nativeRunFinalizations.runId, run.id));
        released();
        // Workspace/result settlement may still be pending after lease release.
        await finalization;
      } finally { await execution.stop(); }
    });
    try {
      await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart(new Date(), [run.id]);
      await leaseReleased;
      if (outcome === "failure_before_dispatch") await Promise.all([...deps.activeRunExecutionPromises]);
      expect(vi.getTimerCount()).toBe(0);
      await vi.advanceTimersByTimeAsync(21_000);
      expect(deps.activeRunExecutionPromises.size).toBe(outcome === "completion" ? 1 : 0);
      expect(native.detach).not.toHaveBeenCalled();
      expect(deps.executeRun).toHaveBeenCalledOnce();
      const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
      expect(coordinator.leaseExpiresAt).toEqual(outcome === "completion" ? null : expiresAt);
    } finally {
      finish();
      await Promise.all([...deps.activeRunExecutionPromises]);
      vi.useRealTimers();
      await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
      await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
    }
  }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);

  it.each(["bootstrap_incomplete", "blocked"] as const)("retries durable remote takeover and preserves its %s disposition", async kind => {
    const run = await seedRun();
    const deps = callbacks(db);
    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId }).where(eq(heartbeatRuns.id, run.id));
    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!,
      phase: "observed", recoveryState: "awaiting_runner_reattach", leaseOwner: "prior-server",
      leaseExpiresAt: new Date(Date.now() + 60_000) });
    await db.insert(nativeRunnerCheckpoints).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!,
      part: "opaque-checkpoint-record", material: { fixture: true } });
    const recovery = createHeartbeatRecovery(db, deps);
    try {
      await recovery.reapOrphanedRuns();
      expect(native.claims).not.toHaveBeenCalled();
      await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() - 1) }).where(eq(nativeRunFinalizations.runId, run.id));
      const claim: NativeRestartRecoveryDisposition = kind === "blocked"
        ? { kind, runId: run.id, reason: "fixture_ambiguous_owner" }
        : { kind, runId: run.id, leaseOwner: "new-server",
          controllerGeneration: 2, providerAttempt: 0, restartKind: "hard", recoveryRequestId: null };
      // Force the old inventory down the unverified-owner path unless recovery
      // dispositions fence it. The runtime's decision must own this run.
      await db.update(nativeRunFinalizations).set({ recoveryState: null }).where(eq(nativeRunFinalizations.runId, run.id));
      native.claims.mockResolvedValue([claim]);
      await recovery.reapOrphanedRuns();
      expect(native.claims).toHaveBeenCalledWith(expect.objectContaining({ runIds: [run.id] }));
      if (kind === "blocked") expect(deps.executeRun).not.toHaveBeenCalled();
      else expect(deps.executeRun).toHaveBeenCalledExactlyOnceWith(run.id, { nativeLeaseOwner: "new-server", nativeRestartRecovery: claim });
      expect(deps.setRunStatusFromLive).not.toHaveBeenCalled();
      expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].errorCode)
        .not.toBe(NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE);
      await Promise.all([...deps.activeRunExecutionPromises]);
      await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, run.id));
      native.claims.mockClear();
      await recovery.reapOrphanedRuns();
      expect(native.claims).not.toHaveBeenCalled();
    } finally {
      await Promise.all([...deps.activeRunExecutionPromises]);
      await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
      await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
    }
  });

  it("notifies a replacement after committing the departing controller's released lease", async () => {
    const run = await seedRun(), deps = callbacks(db);
    const identity = await currentNativeControllerIdentity();
    await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, run.agentId));
    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId }).where(eq(heartbeatRuns.id, run.id));
    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!,
      phase: "observed", controllerBootId: identity.bootId, leaseOwner: "departing-server", leaseExpiresAt: new Date(Date.now() + 60_000) });
    const wake = vi.fn(), unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeProcessSessions, wake);
    try {
      const before = new Date();
      await createHeartbeatRecovery(db, deps).drainRunningRunsForShutdown("SIGTERM", before, [run.id]);
      expect(native.detach).toHaveBeenCalledWith([run.id]);
      expect(wake).toHaveBeenCalled();
      const [saved] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
      expect(saved.leaseExpiresAt).toEqual(before);
      expect(saved.recoveryState).toBe("awaiting_runner_reattach");
    } finally {
      unsubscribe();
      await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
      await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
    }
  });

  it("does not detach a different server's native run during shutdown", async () => {
    const run = await seedRun();
    const deps = callbacks(db);
    const bootId = randomUUID();
    const expiresAt = new Date(Date.now() + 60_000);
    await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, run.agentId));
    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId }).where(eq(heartbeatRuns.id, run.id));
    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!,
      phase: "observed", controllerBootId: bootId, leaseOwner: "other-server", leaseExpiresAt: expiresAt });
    try {
      await createHeartbeatRecovery(db, deps).drainRunningRunsForShutdown("SIGTERM", new Date(), [run.id]);
      expect(native.detach).not.toHaveBeenCalled();
      expect(deps.releaseEnvironmentLeasesForRun).not.toHaveBeenCalled();
      expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0])
        .toMatchObject({ controllerBootId: bootId, leaseOwner: "other-server", leaseExpiresAt: expiresAt, recoveryState: null });
    } finally {
      await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
      await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
    }
  });

  it("does not dispatch recovery for a run still preparing in this heartbeat process", async () => {
    const run = await seedRun();
    await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, run.agentId));
    await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, run.issueId!));
    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId }).where(eq(heartbeatRuns.id, run.id));
    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, phase: "observed" });
    const deps = callbacks(db);
    deps.activeRunExecutions.add(run.id);
    deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null);
    native.claims.mockImplementationOnce(claimNativeRestartRecoveries);
    const before = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
    try {
      const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart();
      expect(result.awaitingEvidenceRunIds).toEqual([run.id]);
      expect(result.claims).toEqual([]);
      expect(deps.executeRun).not.toHaveBeenCalled();
      expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id))).toEqual(before);
      expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))).toHaveLength(0);
      expect((await deps.getRun(run.id))?.status).toBe("running");
    } finally {
      deps.activeRunExecutions.delete(run.id);
      await Promise.all([...deps.activeRunExecutionPromises]);
      await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
      await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
    }
  });

  it.each(["getRun", "appendRunEvent"] as const)("dispatches committed claims when recording a transition fails in %s", async failing => {
    const deps = callbacks(db);
    const deferred = { kind: "awaiting_evidence" as const, runId: randomUUID(), reason: NATIVE_RESTART_RECOVERY_FAILED,
      retryAt: new Date(Date.now() + 30_000) };
    const claim: NativeRestartRecoveryClaim = { kind: "bootstrap_incomplete", runId: randomUUID(), leaseOwner: "claimed-owner",
      controllerGeneration: 1, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null };
    native.claims.mockResolvedValueOnce([deferred, claim]);
    if (failing === "getRun") deps.getRun.mockRejectedValue(new Error("run read failed"));
    else {
      deps.getRun.mockImplementation(async id => ({ id }) as unknown as Awaited<ReturnType<HeartbeatRecoveryDependencies["getRun"]>>);
      deps.appendRunEvent.mockRejectedValue(new Error("event write failed"));
    }
    const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart();
    expect(result.claims).toEqual([claim]);
    expect(result.awaitingEvidenceRunIds).toEqual([deferred.runId]);
    expect(deps.executeRun).toHaveBeenCalledExactlyOnceWith(claim.runId, { nativeLeaseOwner: claim.leaseOwner, nativeRestartRecovery: claim });
    if (failing === "appendRunEvent") {
      expect(deps.appendRunEvent).toHaveBeenCalledWith(expect.objectContaining({ id: deferred.runId }), expect.objectContaining({
        level: "warn", payload: expect.objectContaining({ reason: NATIVE_RESTART_RECOVERY_FAILED }) }));
    }
    await Promise.all([...deps.activeRunExecutionPromises]);
  });

  it("does not claim or execute recovery when persisted finalization fails", async () => {
    const deps = callbacks(db);
    native.finalizations.mockRejectedValueOnce(new Error("result settlement failed"));
    await expect(createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart()).rejects.toThrow("result settlement failed");
    expect(native.claims).not.toHaveBeenCalled();
    expect(deps.executeRun).not.toHaveBeenCalled();
    expect(native.cleanups).not.toHaveBeenCalled();
  });

  it("keeps blocked and ambiguous ownership out of execution", async () => {
    const deps = callbacks(db);
    const dispositions: NativeRestartRecoveryDisposition[] = [
      { kind: "blocked", runId: randomUUID(), reason: "conflicting_owner" },
      { kind: "awaiting_evidence", runId: randomUUID(), reason: "missing_stop_evidence" },
    ];
    native.claims.mockResolvedValue(dispositions);
    const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart();
    expect(result.blockedRunIds).toEqual([dispositions[0].runId]);
    expect(result.awaitingEvidenceRunIds).toEqual([dispositions[1].runId]);
    expect(result.claims).toEqual([]);
    expect(deps.executeRun).not.toHaveBeenCalled();
    await Promise.all([...deps.activeRunExecutionPromises]);
  });

  it.each(["future", "overdue-owner"])("rearms the persisted native retry deadline through the service callback: %s", async scenario => {
    const run = await seedRun();
    const nextAttemptAt = new Date(Date.now() + (scenario === "future" ? 60_000 : -1_000));
    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId, status: "failed" }).where(eq(heartbeatRuns.id, run.id));
    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, phase: "retryable_failure", nextAttemptAt,
      failureCode: scenario === "overdue-owner" ? "native_controller_recovery_waiting_for_owner" : null });
    const deps = callbacks(db);
    const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart();
    expect(result.scheduledRetryRunIds).toEqual([run.id]);
    expect(deps.scheduleNativeSessionResumeDispatch).toHaveBeenCalledWith(run.id, nextAttemptAt);
    expect(deps.executeRun).not.toHaveBeenCalled();
    await Promise.all([...deps.activeRunExecutionPromises]);
    await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
    await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
  });

  it("schedules deferred remote retries without repeating unchanged transition events", async () => {
    const run = await seedRun();
    const deps = callbacks(db), retryAt = new Date(Date.now() + 30_000);
    deps.getRun.mockResolvedValue(run);
    const held = { kind: "blocked" as const, runId: run.id, reason: "native_retry_admission_ineligible", retryAt };
    native.claims.mockResolvedValue([{ ...held, transitionUnchanged: false }]);
    const recovery = createHeartbeatRecovery(db, deps);
    await recovery.recoverNativeRunsAfterRestart(new Date(), [run.id], { skipDelegatedWork: true });
    expect(deps.scheduleNativeSessionResumeDispatch).toHaveBeenCalledWith(run.id, retryAt, "restart");
    expect(deps.appendRunEvent).toHaveBeenCalledTimes(1);
    native.claims.mockResolvedValue([{ ...held, transitionUnchanged: true }]);
    await recovery.recoverNativeRunsAfterRestart(new Date(), [run.id], { skipDelegatedWork: true });
    expect(deps.appendRunEvent).toHaveBeenCalledTimes(1);
    expect(deps.executeRun).not.toHaveBeenCalled();
    await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
  });

  it.each(["nativeWorkspaceSync", "remoteWorkspace"])("routes a due %s retry to remote ownership verification", async reference => {
    const run = await seedRun();
    const leaseId = randomUUID(), workspaceId = randomUUID();
    await db.update(agents).set({ adapterType: "paperclip_runner", status: "active" }).where(eq(agents.id, run.agentId));
    await db.update(issues).set({ status: "in_progress", assigneeAgentId: run.agentId,
      executionRunId: null, checkoutRunId: null }).where(eq(issues.id, run.issueId!));
    const location = { workspaceId, leaseId, providerLeaseId: "retained-sandbox", remoteCwd: "/workspace", state: "prepared" };
    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId, status: "failed",
      processPid: process.pid, processStartedAt: new Date(), runnerProfileJson: {
        [reference]: reference === "remoteWorkspace"
          ? { ...location, schema: "paperclip.remote-workspace-run.v1", epoch: 1, snapshotId: null }
          : { ...location, schema: "paperclip.native-workspace-sync/v3", descriptorSha256: "a".repeat(64),
            baselineSha256: "b".repeat(64), finalHostSha256: null, resourceDisposition: "destroy" },
      } }).where(eq(heartbeatRuns.id, run.id));
    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!,
      phase: "retryable_failure", nextAttemptAt: new Date(0),
      failureDetail: { recoveryMode: "exact_checkpoint_resume" } });
    const deps = callbacks(db);
    deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null);
    native.claims.mockImplementation(claimNativeRestartRecoveries);
    await createHeartbeatRecovery(db, deps).resumeRetryableNativeProcessSessions({ runId: run.id });
    expect(native.claims).toHaveBeenCalledWith(expect.objectContaining({ runIds: [run.id] }));
    expect(deps.executeRun).not.toHaveBeenCalled();
    expect((await deps.getRun(run.id))?.processPid).toBe(process.pid);
    expect((await db.select().from(issues).where(eq(issues.id, run.issueId!)))[0].executionRunId).toBeNull();
    await Promise.all([...deps.activeRunExecutionPromises]);
    await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
    await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
  });

  it("retains a live remote owner's actual lease deadline without dispatching competing work", async () => {
    const deps = callbacks(db);
    const runId = randomUUID(), retryAt = new Date(Date.now() + 60_000);
    native.claims.mockResolvedValue([{ kind: "awaiting_evidence", runId, reason: "remote_controller_lease_live", retryAt }]);
    const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart();
    expect(result.awaitingEvidenceRunIds).toEqual([runId]);
    expect(deps.scheduleNativeSessionResumeDispatch).toHaveBeenCalledWith(runId, retryAt, "restart");
    expect(deps.executeRun).not.toHaveBeenCalled();
  });

  it("retains shutdown ownership until background native cleanup physically settles", async () => {
    const deps = callbacks(db);
    let settle!: () => void;
    native.cleanups.mockReturnValueOnce(new Promise<void>(resolve => { settle = resolve; }));
    await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart();
    await vi.waitFor(() => expect(native.cleanups).toHaveBeenCalled());
    expect(deps.activeRunExecutionPromises.size).toBe(1);
    settle();
    await Promise.all([...deps.activeRunExecutionPromises]);
    await vi.waitFor(() => expect(deps.activeRunExecutionPromises.size).toBe(0));
  });
});