fix(native): reclaim task execution before same-run recovery

Restore the original task execution binding atomically during native same-run recovery. Preserve current owners, stops, pauses, assignments, and dependency waits.

Isolate candidate failures, bound deferral writes, prioritize undated retries, and allow recovery ahead of queued waiters without premature queue promotion. Add PostgreSQL attachment, ownership, contention, fairness, and queue regressions.

Verification: 99 focused tests, workspace typecheck, build, full current-head CI, and Apex 5/5.

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

server/src/__tests__/native-session-resumption.test.ts

@@ -95,6 +95,9 @@ describe("P6-25 pre-result native session recovery", () => {
   const bootstrapRetryRunId = "79000000-0000-4000-8000-000000000009";
   const observedExpiredRunId = "79000000-0000-4000-8000-000000000010";
   const observedLivePidRunId = "79000000-0000-4000-8000-000000000011";
+  const fixtureRunIds = [runId, cancelledRunId, exhaustedRunId, missingCheckpointRunId,
+    initialRunId, bootstrapRetryRunId, observedExpiredRunId, observedLivePidRunId];
+  const issueForRun = (id: string) => id === runId ? issueId : id;
   const persistedProfile = {
     mode: "native",
     nativeExecutionInput: { schema: "paperclip.native-execution-input.v1", binding: { runId } },
@@ -118,14 +121,10 @@ describe("P6-25 pre-result native session recovery", () => {
       adapterType: "codex_local",
       status: "running",
     });
-    await db.insert(issues).values({
-      id: issueId,
-      companyId,
-      title: "Resume the same native run",
-      status: "in_progress",
-      assigneeAgentId: agentId,
-      workMode: "standard",
-    });
+    await db.insert(issues).values(fixtureRunIds.map(id => ({
+      id: issueForRun(id), companyId, title: "Resume the same native run",
+      status: "in_progress", assigneeAgentId: agentId, workMode: "standard" as const,
+    })));
     await db.insert(heartbeatRuns).values([
       {
         id: runId,
@@ -230,7 +229,10 @@ describe("P6-25 pre-result native session recovery", () => {
         },
         contextSnapshot: { issueId },
       })),
-    ]);
+    ].map(run => ({ ...run, runtimeMode: "native" as const,
+      nativeIssueId: issueForRun(run.id), contextSnapshot: { issueId: issueForRun(run.id) },
+    })));
+
     await db.insert(nativeRunFinalizations).values([
       { runId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) },
       { runId: cancelledRunId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) },
@@ -263,7 +265,10 @@ describe("P6-25 pre-result native session recovery", () => {
         leaseOwner: "prior-native-owner",
         leaseExpiresAt: new Date(0),
       })),
-    ]);
+    ].map(value => ({ ...value, issueId: issueForRun(value.runId),
+      phase: value.phase as typeof nativeRunFinalizations.$inferInsert.phase,
+    })));
+
   }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
 
   afterAll(async () => temporary?.cleanup());
@@ -276,12 +281,15 @@ describe("P6-25 pre-result native session recovery", () => {
     );
     const eligibleRunId = "79100000-0000-4000-8000-000000000099";
     const candidateRunIds = [...heldRunIds, eligibleRunId];
+    await db.insert(issues).values(candidateRunIds.map(id => ({
+      id, companyId, title: "Independent recovery candidate", status: "in_progress", assigneeAgentId: agentId,
+    })));
     await db.insert(heartbeatRuns).values(
       candidateRunIds.map((candidateRunId) => ({
         id: candidateRunId,
         companyId,
         agentId,
-        nativeIssueId: issueId,
+        nativeIssueId: candidateRunId,
         status: "running",
         runtimeMode: "native",
         nativePhase:
@@ -309,7 +317,7 @@ describe("P6-25 pre-result native session recovery", () => {
       candidateRunIds.map((candidateRunId) => ({
         runId: candidateRunId,
         companyId,
-        issueId,
+        issueId: candidateRunId,
         phase: "retryable_failure",
         attempt: 1,
         nextAttemptAt: new Date(0),
@@ -341,12 +349,14 @@ describe("P6-25 pre-result native session recovery", () => {
         ),
       ).toBe(true);
     } finally {
+      await db.delete(heartbeatRunEvents).where(inArray(heartbeatRunEvents.runId, candidateRunIds));
       await db
         .delete(nativeRunFinalizations)
         .where(inArray(nativeRunFinalizations.runId, candidateRunIds));
       await db
         .delete(heartbeatRuns)
         .where(inArray(heartbeatRuns.id, candidateRunIds));
+      await db.delete(issues).where(inArray(issues.id, candidateRunIds));
     }
   });
 
@@ -523,7 +533,7 @@ describe("P6-25 pre-result native session recovery", () => {
       expect.objectContaining({ phase: "terminal_failure", failureCode: "native_session_interrupted" }),
     ]);
     await expect(db.select().from(issueRecoveryActions)
-      .where(eq(issueRecoveryActions.sourceIssueId, issueId))).resolves.toEqual([
+      .where(eq(issueRecoveryActions.sourceIssueId, missingCheckpointRunId))).resolves.toEqual([
       expect.objectContaining({ cause: "native_session_interrupted", wakePolicy: null }),
     ]);
   });
@@ -542,11 +552,13 @@ describe("P6-25 pre-result native session recovery", () => {
 
   it("sends exactly one Sentry event when a sweep resolves a run from running to failed", async () => {
     const freshRunId = "79000000-0000-4000-8000-000000000101";
+    await db.insert(issues).values({ id: freshRunId, companyId, title: "Independent failure report",
+      status: "in_progress", assigneeAgentId: agentId });
     await db.insert(heartbeatRuns).values({
       id: freshRunId,
       companyId,
       agentId,
-      nativeIssueId: issueId,
+      nativeIssueId: freshRunId,
       status: "running",
       runtimeMode: "native",
       runtimeModeResolvedAt: new Date(),
@@ -556,12 +568,12 @@ describe("P6-25 pre-result native session recovery", () => {
           binding: { runId: freshRunId },
         },
       },
-      contextSnapshot: { issueId },
+      contextSnapshot: { issueId: freshRunId },
     });
     await db.insert(nativeRunFinalizations).values({
       runId: freshRunId,
       companyId,
-      issueId,
+      issueId: freshRunId,
       phase: "retryable_failure",
       attempt: 1,
     });
@@ -578,11 +590,13 @@ describe("P6-25 pre-result native session recovery", () => {
 
   it("sends no Sentry event when a sweep finds a run that is already failed", async () => {
     const freshRunId = "79000000-0000-4000-8000-000000000102";
+    await db.insert(issues).values({ id: freshRunId, companyId, title: "Independent failure report",
+      status: "in_progress", assigneeAgentId: agentId });
     await db.insert(heartbeatRuns).values({
       id: freshRunId,
       companyId,
       agentId,
-      nativeIssueId: issueId,
+      nativeIssueId: freshRunId,
       status: "failed",
       runtimeMode: "native",
       runtimeModeResolvedAt: new Date(),
@@ -592,12 +606,12 @@ describe("P6-25 pre-result native session recovery", () => {
           binding: { runId: freshRunId },
         },
       },
-      contextSnapshot: { issueId },
+      contextSnapshot: { issueId: freshRunId },
     });
     await db.insert(nativeRunFinalizations).values({
       runId: freshRunId,
       companyId,
-      issueId,
+      issueId: freshRunId,
       phase: "retryable_failure",
       attempt: 1,
     });

server/src/services/heartbeat/queue.test.ts

@@ -1,5 +1,6 @@
+import { claimNativeSessionResumptions } from "../native-runtime/native-finalization-reconciler.js";
 import type { Db } from "@paperclipai/db";
-import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues } from "@paperclipai/db";
+import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db";
 import { eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
@@ -171,6 +172,33 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => {
     expect(current.executionRunId).toBe(newOwner.id);
   });
 
+  it.each(["running", "failed"])("resumes the %s native owner before promoting its queued waiter", async status => {
+    const { run, issue, agent, company } = await fixture(status);
+    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id, runnerProfileJson: {
+      nativeExecutionInput: { binding: { runId: run.id } },
+      sessionCheckpoint: { providerSessionId: "original-session" },
+    } }).where(eq(heartbeatRuns.id, run.id));
+    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: company.id,
+      issueId: issue.id, phase: "retryable_failure", attempt: 1 });
+    const [waiter] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id,
+      status: "queued", invocationSource: "assignment", responsibleUserId: "queue-test-user", contextSnapshot: { issueId: issue.id },
+    }).returning();
+    const deps = callbacks(db);
+    deps.getAgent.mockResolvedValue(agent);
+    deps.instanceSettings.getExperimental.mockResolvedValue({ enableNativeRunner: true } as Awaited<ReturnType<HeartbeatQueueDependencies["instanceSettings"]["getExperimental"]>>);
+    const queue = createHeartbeatQueue(db, deps);
+    expect(await queue.claimQueuedRun(waiter)).toBeNull();
+    const claims = await claimNativeSessionResumptions({ db, runnerInstanceId: "owner", runIds: [run.id] });
+    expect(claims.map(claim => claim.runId)).toEqual([run.id]);
+    expect(await queue.claimQueuedRun(waiter)).toBeNull();
+    await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, run.id));
+    await db.update(nativeRunFinalizations).set({ phase: "committed", leaseOwner: null, leaseExpiresAt: null })
+      .where(eq(nativeRunFinalizations.runId, run.id));
+    await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, issue.id));
+    expect(await queue.claimQueuedRun(waiter)).toMatchObject({ id: waiter.id, status: "running" });
+    expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].executionRunId).toBe(waiter.id);
+  });
+
   it("cancels at the daily run cap before claiming or dispatching", async () => {
     const { run, agent } = await fixture("queued");
     const cappedAgent = { ...agent, runtimeConfig: { heartbeat: { maxDailyRuns: 0 } } };

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

@@ -3,10 +3,11 @@ import { hasPendingNativeChildCompletion } from "./native-child-completion-deliv
 import { dismissAutomaticCompletionReviews, decisionHasRetiredAutomaticReview } from "./automatic-completion-reviews.js";
 import { logger } from "../../middleware/logger.js";
 import { createHash, randomUUID } from "node:crypto";
-import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lte, notInArray, or, sql } from "drizzle-orm";
+import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lte, ne, notInArray, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
   completionContracts,
+  agents,
   heartbeatRunEvents,
   heartbeatRuns,
   issueWorkProducts,
@@ -55,6 +56,12 @@ import {
   type NativeStatusDecision,
 } from "./status-arbiter.js";
 
+import { getNativeReviewAssignment } from "./native-review-participant.js";
+import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js";
+import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold-guard.js";
+import { readRunCancellation } from "../run-cancellation.js";
+import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
+
 export type NativeReconciliationFacts = {
   equivalentAttentionFamily?: boolean;
   canonicalReplay?: boolean;
@@ -343,7 +350,7 @@ export async function claimNativeSessionResumptions(input: {
 }): Promise<NativeSessionResumeClaim[]> {
   const now = input.now ?? new Date();
   const candidates = await input.db
-    .select({ runId: heartbeatRuns.id })
+    .select({ runId: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: nativeRunFinalizations.issueId })
     .from(heartbeatRuns)
     .innerJoin(
       nativeRunFinalizations,
@@ -371,6 +378,7 @@ export async function claimNativeSessionResumptions(input: {
           : []),
       ),
     )
+    .orderBy(sql`${nativeRunFinalizations.nextAttemptAt} asc nulls first`, asc(nativeRunFinalizations.runId))
     .limit(input.limit ?? 25);
 
   const claims: NativeSessionResumeClaim[] = [];
@@ -378,16 +386,21 @@ export async function claimNativeSessionResumptions(input: {
     const leaseOwner = `${input.runnerInstanceId}:resume:${randomUUID()}`;
     let terminalRunToEmit: typeof heartbeatRuns.$inferSelect | null = null;
     const claimed = await input.db.transaction(async (tx) => {
-      const row = await tx.select({
-        run: heartbeatRuns,
-        coordinator: nativeRunFinalizations,
-      }).from(heartbeatRuns)
-        .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
-        .where(eq(heartbeatRuns.id, candidate.runId))
-        .for("update")
-        .limit(1)
-        .then((rows) => rows[0] ?? null);
-      if (!row) return false;
+      await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`);
+      const [issue] = await tx.select().from(issues).where(and(
+        eq(issues.companyId, candidate.companyId), eq(issues.id, candidate.issueId),
+      )).for("update").limit(1);
+      if (!issue || !["todo", "in_progress", "in_review"].includes(issue.status)) return false;
+      const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and(
+        eq(nativeRunFinalizations.companyId, candidate.companyId), eq(nativeRunFinalizations.runId, candidate.runId),
+        eq(nativeRunFinalizations.issueId, issue.id),
+      )).for("update").limit(1);
+      const [run] = await tx.select().from(heartbeatRuns).where(and(
+        eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, candidate.runId),
+        eq(heartbeatRuns.nativeIssueId, issue.id),
+      )).for("update").limit(1);
+      if (!run || !coordinator) return false;
+      const row = { run, coordinator };
       if (
         row.run.runtimeMode !== "native" ||
         isNativeRunnerOwnershipHeld(row.run) ||
@@ -404,10 +417,37 @@ export async function claimNativeSessionResumptions(input: {
       )
         return false;
 
+      if ((issue.executionRunId && issue.executionRunId !== run.id) ||
+          (issue.checkoutRunId && issue.checkoutRunId !== run.id)) return false;
+      const [agent] = await tx.select().from(agents).where(and(
+        eq(agents.companyId, candidate.companyId), eq(agents.id, run.agentId),
+      )).limit(1);
+      if (!agent || !(await evaluateAgentInvokabilityFromDb(tx as unknown as Db, agent)).invokable) return false;
+      if (issue.assigneeAgentId !== run.agentId && !await getNativeReviewAssignment(tx as unknown as Db, {
+        companyId: run.companyId, issueId: issue.id, agentId: run.agentId,
+        contextSnapshot: run.contextSnapshot,
+      })) return false;
+      if (await isAutomaticRecoverySuppressedByPauseHold(tx as unknown as Db, run.companyId, issue.id)) return false;
+      if (!(await issueService(tx as unknown as Db).getDependencyReadiness(issue.id, tx)).isDependencyReady) return false;
+      const [competing] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+        eq(heartbeatRuns.companyId, run.companyId), ne(heartbeatRuns.id, run.id),
+        inArray(heartbeatRuns.status, issue.executionRunId === run.id
+          ? ["running", "scheduled_retry"] : ["queued", "running", "scheduled_retry"]),
+        or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id),
+          sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`),
+      )).limit(1);
+      if (competing) return false;
+
       const profile = row.run.runnerProfileJson ?? {};
       const persistedInput = profile.nativeExecutionInput;
       const checkpoint = profile.sessionCheckpoint;
       const failureDetail = record(row.coordinator.failureDetail);
+      const cancellation = readRunCancellation(run.resultJson);
+      const nativeCancellation = record(record(run.resultJson).nativeCancellation);
+      if ((cancellation?.source === "operator" && cancellation.expected) ||
+          (["run", "issue"].includes(String(nativeCancellation.scope)) &&
+            ["pending", "acknowledged"].includes(String(nativeCancellation.dispatchState))) ||
+          typeof failureDetail.successorRunId === "string") return false;
       const originalFailureCode =
         typeof failureDetail.originalFailureCode === "string"
           ? failureDetail.originalFailureCode
@@ -534,7 +574,24 @@ export async function claimNativeSessionResumptions(input: {
         nativePhaseUpdatedAt: now,
         updatedAt: now,
       }).where(eq(heartbeatRuns.id, row.run.id));
+      await tx.update(issues).set({
+        executionRunId: run.id,
+        executionAgentNameKey: agent.name.trim().toLowerCase() || null,
+        executionLockedAt: now,
+        updatedAt: now,
+      }).where(and(eq(issues.id, issue.id), eq(issues.companyId, run.companyId)));
+      if (issue.executionRunId !== run.id) await appendHeartbeatRunEvent(tx as unknown as Db, {
+        companyId: run.companyId, runId: run.id, agentId: run.agentId,
+        eventType: "native.recovery.execution_reclaimed", stream: "system", level: "info",
+        message: "Reclaimed the original task execution for native session recovery.",
+        payload: { issueId: issue.id },
+      });
       return true;
+    }).catch((error: unknown) => {
+      terminalRunToEmit = null;
+      logger.warn({ err: error, companyId: candidate.companyId, runId: candidate.runId },
+        "failed to claim native session recovery candidate");
+      return false;
     });
     // Telemetry is best-effort background work; it must not delay claiming
     // the remaining candidates in this loop, so fire it and do not await it.
@@ -543,6 +600,24 @@ export async function claimNativeSessionResumptions(input: {
       void reportRunFailure(input.db, terminalRunToEmit);
     }
     if (claimed) claims.push({ runId: candidate.runId, leaseOwner });
+    else await input.db.transaction(async tx => {
+      await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`);
+      await tx.update(nativeRunFinalizations).set({
+        nextAttemptAt: new Date(now.getTime() + 30_000),
+        updatedAt: now,
+      }).where(and(
+        eq(nativeRunFinalizations.companyId, candidate.companyId),
+        eq(nativeRunFinalizations.issueId, candidate.issueId),
+        eq(nativeRunFinalizations.runId, candidate.runId),
+        eq(nativeRunFinalizations.phase, "retryable_failure"),
+        isNull(nativeRunFinalizations.resultId),
+        or(isNull(nativeRunFinalizations.nextAttemptAt), lte(nativeRunFinalizations.nextAttemptAt, now)),
+        or(isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.leaseExpiresAt), lte(nativeRunFinalizations.leaseExpiresAt, now)),
+      ));
+    }).catch((error: unknown) => {
+      logger.warn({ err: error, companyId: candidate.companyId, runId: candidate.runId },
+        "failed to defer native session recovery candidate");
+    });
   }
   return claims;
 }

server/src/services/native-runtime/native-session-resumption-binding.test.ts

@@ -0,0 +1,209 @@
+import { randomUUID } from "node:crypto";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { eq } from "drizzle-orm";
+import {
+  agents, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS,
+  issueRelations, issueTreeHolds, issueTreeHoldMembers,
+} from "@paperclipai/db";
+import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
+import { claimNativeSessionResumptions, dispatchNativeSessionResumptions } from "./native-finalization-reconciler.js";
+import { stageNativeRunnerWakeAttachments } from "./native-runner-file-handoff.js";
+
+describe("native recovery execution binding", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  beforeAll(async () => {
+    temporary = await startEmbeddedPostgresTestDatabase("native-resume-binding-");
+    db = createDb(temporary.connectionString);
+  }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
+  afterAll(async () => temporary?.cleanup());
+
+  async function fixture() {
+    const companyId = randomUUID();
+    const agentId = randomUUID();
+    const issueId = randomUUID();
+    const runId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Recovery binding", issuePrefix: "N" + companyId.replaceAll("-", "").slice(0, 6).toUpperCase() });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Recovery owner", status: "idle", adapterType: "paperclip_runner" });
+    await db.insert(issues).values({ id: issueId, companyId, title: "Resume original work", status: "in_progress", assigneeAgentId: agentId });
+    await db.insert(heartbeatRuns).values({
+      id: runId, companyId, agentId, issueId, nativeIssueId: issueId, status: "failed", runtimeMode: "native",
+      finishedAt: new Date(), contextSnapshot: { issueId }, runnerProfileJson: {
+        nativeExecutionInput: { schema: "paperclip.native-execution-input.v1", binding: { runId, companyId, agentId, issueId, executionWorkspaceId: runId } },
+        sessionCheckpoint: { sessionId: "provider-session", providerSessionId: "provider-session", identity: { runId, companyId, agentId, issueId } },
+      },
+    });
+    await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) });
+    return { companyId, agentId, issueId, runId };
+  }
+
+  it("restores the original execution before the real attachment authorization gate", async () => {
+    const f = await fixture();
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "owner", runIds: [f.runId] })).toHaveLength(1);
+    const staged = await stageNativeRunnerWakeAttachments({ db, binding: {
+      ...f, workspaceRoot: "/remote/workspace", executionTargetKind: "remote",
+    } });
+    expect(staged.attachments).toEqual([]);
+    await staged.cleanup();
+    expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({
+      executionRunId: f.runId, executionAgentNameKey: "recovery owner", status: "in_progress", assigneeAgentId: f.agentId,
+    });
+  });
+
+  it.each(["execution", "checkout", "queued"] as const)("does not replace another %s owner", async (kind) => {
+    const f = await fixture();
+    const otherRunId = randomUUID();
+    await db.insert(heartbeatRuns).values({ id: otherRunId, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId,
+      status: kind === "queued" ? "queued" : "failed", runtimeMode: "legacy", contextSnapshot: { issueId: f.issueId } });
+    if (kind !== "queued") await db.update(issues).set(kind === "execution"
+      ? { executionRunId: otherRunId } : { checkoutRunId: otherRunId }).where(eq(issues.id, f.issueId));
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]);
+    expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
+      .toMatchObject({ phase: "retryable_failure", leaseOwner: null });
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].status).toBe("failed");
+  });
+
+  it.each(["done", "cancelled", "blocked"])("preserves a task moved to %s", async status => {
+    const f = await fixture();
+    await db.update(issues).set({ status }).where(eq(issues.id, f.issueId));
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]);
+    expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status, executionRunId: null });
+  });
+
+  it("does not recover work after reassignment", async () => {
+    const f = await fixture();
+    const nextAgent = randomUUID();
+    await db.insert(agents).values({ id: nextAgent, companyId: f.companyId, name: "New owner", status: "idle" });
+    await db.update(issues).set({ assigneeAgentId: nextAgent }).where(eq(issues.id, f.issueId));
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]);
+  });
+
+  it("preserves an agent pause", async () => {
+    const f = await fixture();
+    await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId));
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]);
+    expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0].status).toBe("paused");
+  });
+
+  it("waits for a first-class dependency", async () => {
+    const f = await fixture();
+    const blocker = randomUUID();
+    await db.insert(issues).values({ id: blocker, companyId: f.companyId, title: "Unfinished prerequisite", status: "todo" });
+    await db.insert(issueRelations).values({ companyId: f.companyId, issueId: blocker, relatedIssueId: f.issueId, type: "blocks" });
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]);
+  });
+
+  it("preserves a task-tree pause hold", async () => {
+    const f = await fixture();
+    const hold = randomUUID();
+    await db.insert(issueTreeHolds).values({ id: hold, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause" });
+    await db.insert(issueTreeHoldMembers).values({ holdId: hold, companyId: f.companyId, issueId: f.issueId,
+      issueTitle: "Paused work", issueStatus: "in_progress" });
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]);
+    expect((await db.select().from(issueTreeHolds).where(eq(issueTreeHolds.id, hold)))[0].status).toBe("active");
+  });
+
+  it.each(["run", "issue"])("preserves a recorded native %s stop", async scope => {
+    const f = await fixture();
+    await db.update(heartbeatRuns).set({ resultJson: { nativeCancellation: { scope, dispatchState: "acknowledged" } } })
+      .where(eq(heartbeatRuns.id, f.runId));
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]);
+  });
+
+  it("preserves operator cancellation when terminal status was overwritten", async () => {
+    const f = await fixture();
+    await db.update(heartbeatRuns).set({ resultJson: { cancellation: {
+      source: "operator", expected: true, initiator: { type: "user" }, reason: "Stop", recordedAt: new Date().toISOString(),
+    } } }).where(eq(heartbeatRuns.id, f.runId));
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]);
+  });
+
+  it("does not revive a superseded source", async () => {
+    const f = await fixture();
+    await db.update(nativeRunFinalizations).set({ failureDetail: { successorRunId: randomUUID() } })
+      .where(eq(nativeRunFinalizations.runId, f.runId));
+    expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "stale", runIds: [f.runId] })).toEqual([]);
+  });
+
+  it.each(["pause", "dependency"])("keeps a %s wait from starving the next bounded claim", async kind => {
+    const held = await fixture();
+    const eligible = await fixture();
+    if (kind === "pause") {
+      await db.update(agents).set({ status: "paused" }).where(eq(agents.id, held.agentId));
+    } else {
+      const blocker = randomUUID();
+      await db.insert(issues).values({ id: blocker, companyId: held.companyId, title: "Waiting prerequisite", status: "todo" });
+      await db.insert(issueRelations).values({ companyId: held.companyId, issueId: blocker, relatedIssueId: held.issueId, type: "blocks" });
+    }
+    const now = new Date();
+    await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(1) })
+      .where(eq(nativeRunFinalizations.runId, eligible.runId));
+    const input = { db, runnerInstanceId: "owner", runIds: [held.runId, eligible.runId], limit: 1, now };
+    const first = await claimNativeSessionResumptions(input);
+    const second = await claimNativeSessionResumptions(input);
+    expect([...first, ...second].map(claim => claim.runId)).toEqual([eligible.runId]);
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, held.runId)))[0].status).toBe("failed");
+  });
+
+  it.each(["task", "coordinator"])("dispatches earlier claims when a later %s lock times out", async kind => {
+    const first = await fixture();
+    const locked = await fixture();
+    await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(1) })
+      .where(eq(nativeRunFinalizations.runId, locked.runId));
+    let release!: () => void;
+    let acquired!: () => void;
+    const ready = new Promise<void>(resolve => { acquired = resolve; });
+    const wait = new Promise<void>(resolve => { release = resolve; });
+    const blocker = db.transaction(async tx => {
+      if (kind === "task") await tx.select().from(issues).where(eq(issues.id, locked.issueId)).for("update");
+      else await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, locked.runId)).for("update");
+      acquired();
+      await wait;
+    });
+    try {
+      await ready;
+      const dispatch = vi.fn();
+      const claims = await dispatchNativeSessionResumptions({ db, runnerInstanceId: "owner",
+        runIds: [first.runId, locked.runId], dispatch });
+      expect(claims.map(claim => claim.runId)).toEqual([first.runId]);
+      expect(dispatch).toHaveBeenCalledExactlyOnceWith(claims[0]);
+    } finally {
+      release();
+      await blocker;
+    }
+  }, 15_000);
+
+  it.each(["pause", "dependency"])("reaches undated retries past %s waits at real sweep intervals", async kind => {
+    const held = await fixture();
+    const eligible = await fixture();
+    if (kind === "pause") {
+      await db.update(agents).set({ status: "paused" }).where(eq(agents.id, held.agentId));
+    } else {
+      const blocker = randomUUID();
+      await db.insert(issues).values({ id: blocker, companyId: held.companyId, title: "Waiting prerequisite", status: "todo" });
+      await db.insert(issueRelations).values({ companyId: held.companyId, issueId: blocker, relatedIssueId: held.issueId, type: "blocks" });
+    }
+    await db.update(nativeRunFinalizations).set({ nextAttemptAt: null })
+      .where(eq(nativeRunFinalizations.runId, eligible.runId));
+    const now = Date.now();
+    const claims = [];
+    for (const offset of [0, 30_000, 60_000]) claims.push(...await claimNativeSessionResumptions({
+      db, runnerInstanceId: "owner", runIds: [held.runId, eligible.runId], limit: 1, now: new Date(now + offset),
+    }));
+    expect(claims.map(claim => claim.runId)).toEqual([eligible.runId]);
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, held.runId)))[0].status).toBe("failed");
+  });
+
+  it("admits only one recovering run for a task", async () => {
+    const f = await fixture();
+    const second = randomUUID();
+    const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
+    await db.insert(heartbeatRuns).values({ ...source, id: second, runnerProfileJson: {
+      ...source.runnerProfileJson, nativeExecutionInput: { binding: { runId: second } },
+    } });
+    await db.insert(nativeRunFinalizations).values({ runId: second, companyId: f.companyId, issueId: f.issueId, phase: "retryable_failure", attempt: 1 });
+    const results = await Promise.all([f.runId, second].map(runId => claimNativeSessionResumptions({ db, runnerInstanceId: "owner", runIds: [runId] })));
+    expect(results.flat()).toHaveLength(1);
+    expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(results.flat()[0]!.runId);
+  });
+});