fix(tasks): advertise a native resume only while the scheduler would claim it (#19523)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - When a native run cannot reattach to its provider after a
control-plane restart, the native coordinator resumes the same run by
itself after about 30 seconds, up to three attempts
> - The run projection shows this correctly as "Retry scheduled" with
only `inspect_run` permitted
> - The task interruption shown on the issue read only system-recovery
evidence and `scheduledRetryAt`. It never read the coordinator's
`nextAttemptAt`, so it showed `state: failed, canRetry: true` and
offered a manual Retry while the automatic resume was still pending
> - The two surfaces disagree, and a user click can race the automatic
resume. That is one way a Retry gets parked behind "execution checks"
> - #19503 made the task interruption read the coordinator's pending
resume time. Review found that the coordinator also pushes that time
forward while it defers an ineligible candidate, so a reassigned or
blocked task could hide Retry for good
> - This pull request advertises the resume only while the scheduler's
synchronous admission checks pass
> - The benefit is one consistent story: while the automatic resume is
pending, the task shows it with its time and no competing Retry. If
recovery ends, exactly one Retry returns

## Linked Issues or Issue Description

Follow-up to #19503. Refs #19487, #19497.

**What happened?**
A native run failed with `native_session_interrupted` after a restart.
Its finalization row had `phase: retryable_failure` and `nextAttemptAt`
30 seconds later. The run projection said "Retry scheduled". The issue
interruption said `state: failed, canRetry: true, nextRetryAt: null`.

**What did you expect?**
While the automatic resume is pending, the task shows that it is
resuming, with the time, and no Retry. When the resume fails for good or
never runs, exactly one Retry that works on the first click.

**Steps to reproduce**
1. Run a native task and restart the control plane so the native attach
fails.
2. Read `GET /api/issues/:id` while the coordinator's resume is pending.
3. Observe `interruption.canRetry: true` while `execution.phase` is
`retry_scheduled`.

## What Changed

- `task-interruption-presentation.ts`: the pending-resume read now also
requires the task to be in a resumable status, assigned to the run's
agent, not owned by another execution or checkout run, the run's
question not cancelled, and the agent invokable (not paused, terminated
or pending approval). These mirror `readNativeSessionResumptionAgent`.
- `native-session-resumption-statuses.ts`: one shared status set used by
both the scheduler admission and the read model.
- No time cap: provider cooldowns and remote backoff schedules stay
visible as a pending resume.

## Verification

- `task-interruption-query.test.ts` (embedded Postgres): a reassigned
task and a blocked task show one Retry; a paused agent asks to be
resumed; in each case `readNativeSessionResumptionAgent` also refuses
the candidate, so the read model and the scheduler agree. A resume an
hour out (provider cooldown) stays `retry_scheduled` with no Retry. Red
on master for the reassigned case.
- All 57 presentation and query tests pass. Server typecheck passes.

## Risks

- Low risk. Read model only.
- A review-assignment resume (reviewer is not the assignee) shows Retry
instead of the pending resume. That is the safe fallback.
- Deferrals for a pause hold, dependency readiness or a competing run
are not mirrored. Those gates also hold a manual Retry, so no competing
action appears.
- The `POST /issues/:id/retry` endpoint is unchanged.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with
tool use (shell, file editing, embedded Postgres tests).

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

server/src/__tests__/task-interruption-query.test.ts

@@ -4,6 +4,7 @@ import { createRunSecretRedactionRegistry } from "../services/run-secret-redacti
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
 import { agents, agentWakeupRequests, companies, createDb, applyPendingMigrations, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db";
 import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js";
+import { readNativeSessionResumptionAgent } from "../services/native-runtime/native-session-resumption-admission.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { captureInterruptionQuery, seedInterruptionFixture } from "./helpers/task-interruption-fixture.js";
@@ -39,6 +40,49 @@ describe.skipIf(!support.supported)("task interruption latest run query", () =>
       await taskInterruptionPresentation(db, issue)]) {
       expect(presentation).toMatchObject({ state: "retry_scheduled", canRetry: false, nextRetryAt: resumeAt.toISOString() });
     }
+    // The coordinator also pushes nextAttemptAt while it defers an ineligible
+    // candidate. Those deferrals, or one far past the failure, never hide Retry.
+    const otherAgentId = randomUUID();
+    await db.insert(agents).values({ id: otherAgentId, companyId, name: "Other agent", adapterType: "paperclip_runner" });
+    const admission = async () => {
+      const [currentIssue] = await db.select().from(issues).where(eq(issues.id, issue.id));
+      const [currentRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
+      const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId));
+      return readNativeSessionResumptionAgent(db, { issue: currentIssue!, run: currentRun!, coordinator: coordinator! });
+    };
+    expect(await admission()).not.toBeNull();
+    const otherRunId = randomUUID();
+    await db.insert(heartbeatRuns).values({ id: otherRunId, companyId, agentId: otherAgentId, status: "cancelled",
+      finishedAt: new Date(Date.now() - 60_000), createdAt: new Date(Date.now() - 60 * 60_000) });
+    const terminatedManagerId = randomUUID();
+    await db.insert(agents).values({ id: terminatedManagerId, companyId, name: "Former manager",
+      adapterType: "paperclip_runner", status: "terminated" });
+    for (const [table, change, restore] of [
+      ["issue", { assigneeAgentId: otherAgentId }, { assigneeAgentId: agentId }],
+      ["issue", { status: "blocked" }, { status: "in_progress" }],
+      ["issue", { executionRunId: otherRunId }, { executionRunId: null }],
+      ["issue", { checkoutRunId: otherRunId }, { checkoutRunId: null }],
+      ["run", { contextSnapshot: { nativeQuestionCancellation: { requestId: "q" } } }, { contextSnapshot: {} }],
+      // Invokability is more than status: an idle agent under a terminated manager is refused.
+      ["agent", { reportsTo: terminatedManagerId }, { reportsTo: null }],
+    ] as const) {
+      if (table === "issue") await db.update(issues).set(change).where(eq(issues.id, issue.id));
+      else if (table === "run") await db.update(heartbeatRuns).set(change).where(eq(heartbeatRuns.id, runId));
+      else await db.update(agents).set(change).where(eq(agents.id, agentId));
+      // The scheduler would defer this candidate, so the task offers Retry.
+      if (table !== "run") expect(await admission()).toBeNull();
+      expect((await taskInterruptionPresentations(db, companyId, [issue])).get(issue.id))
+        .toMatchObject({ state: "failed", canRetry: true });
+      if (table === "issue") await db.update(issues).set(restore).where(eq(issues.id, issue.id));
+      else if (table === "run") await db.update(heartbeatRuns).set(restore).where(eq(heartbeatRuns.id, runId));
+      else await db.update(agents).set(restore).where(eq(agents.id, agentId));
+    }
+    expect(await admission()).not.toBeNull();
+    // A provider cooldown can schedule the resume an hour out; it stays visible.
+    const cooldown = new Date(Date.now() + 60 * 60_000);
+    await db.update(nativeRunFinalizations).set({ nextAttemptAt: cooldown }).where(eq(nativeRunFinalizations.runId, runId));
+    expect((await taskInterruptionPresentations(db, companyId, [issue])).get(issue.id))
+      .toMatchObject({ state: "retry_scheduled", canRetry: false, nextRetryAt: cooldown.toISOString() });
     // Exhausted recovery is terminal: exactly one Retry returns.
     await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", nextAttemptAt: null })
       .where(eq(nativeRunFinalizations.runId, runId));
@@ -46,6 +90,29 @@ describe.skipIf(!support.supported)("task interruption latest run query", () =>
       .toMatchObject({ state: "failed", canRetry: true });
   });
 
+  it("decides several pending native resumes in one list read", async () => {
+    const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Resuming list", issuePrefix: `L${companyId.slice(0, 7)}` });
+    await db.insert(agents).values([{ id: agentId, companyId, name: "Resuming agent", adapterType: "paperclip_runner" },
+      { id: otherAgentId, companyId, name: "Other agent", adapterType: "paperclip_runner" }]);
+    const resumeAt = new Date(Date.now() + 30_000);
+    const list = [];
+    for (let index = 0; index < 6; index++) {
+      const [issue] = await db.insert(issues).values({ companyId, title: `Resume ${index}`, status: "in_progress",
+        assigneeAgentId: index % 2 ? otherAgentId : agentId }).returning();
+      const runId = randomUUID();
+      await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issue.id, scopeKind: "issue",
+        status: "failed", runtimeMode: "native", errorCode: "native_session_interrupted", finishedAt: new Date() });
+      await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue.id, runId, phase: "retryable_failure",
+        attempt: 1, nextAttemptAt: resumeAt });
+      list.push(issue);
+    }
+    const result = await taskInterruptionPresentations(db, companyId, list);
+    list.forEach((issue, index) => expect(result.get(issue.id)).toMatchObject(index % 2
+      ? { state: "failed", canRetry: true }
+      : { state: "retry_scheduled", canRetry: false, nextRetryAt: resumeAt.toISOString() }));
+  });
+
   it("redacts workspace preparation errors on task detail and list reads", async () => {
     vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "a".repeat(64));
     try {

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

@@ -1,4 +1,5 @@
 import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
+import { NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES } from "./native-session-resumption-statuses.js";
 import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
 import { pendingNativeRunProjectionCondition } from "./native-run-projection.js";
 import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js";
@@ -158,7 +159,7 @@ export async function claimNativeSessionResumptions(input: {
       if (await retireSupersededNativeRetry(tx as unknown as Db, {
         run, coordinator, issueExecutionRunId: issue.executionRunId, now,
       })) return false;
-      if (!["todo", "in_progress", "in_review"].includes(issue.status)) return false;
+      if (!(NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES as readonly string[]).includes(issue.status)) return false;
       const row = { run, coordinator };
       if (
         row.run.runtimeMode !== "native" ||

server/src/services/native-runtime/native-session-resumption-admission.ts

@@ -8,6 +8,7 @@ import { getNativeReviewAssignment } from "./native-review-participant.js";
 import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
 import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
 import { normalizeAgentNameKey } from "../agent-name-key.js";
+import { NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES } from "./native-session-resumption-statuses.js";
 
 export async function restoreNativeSessionExecution(db: Db, input: {
   issue: Pick<typeof issues.$inferSelect, "id" | "executionRunId">;
@@ -48,7 +49,7 @@ export async function readNativeSessionResumptionAgent(db: Db, input: {
   const { issue, run, coordinator } = input;
   if (issue.companyId !== run.companyId || issue.id !== run.nativeIssueId ||
       coordinator.companyId !== run.companyId || coordinator.issueId !== issue.id || coordinator.runId !== run.id ||
-      !["todo", "in_progress", "in_review"].includes(issue.status) ||
+      !(NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES as readonly string[]).includes(issue.status) ||
       (issue.executionRunId && issue.executionRunId !== run.id) ||
       (issue.checkoutRunId && issue.checkoutRunId !== run.id)) return null;
   const [agent] = await db.select().from(agents).where(and(

server/src/services/native-runtime/native-session-resumption-statuses.ts

@@ -0,0 +1,4 @@
+/** Task statuses from which a native same-run resume may be claimed. The task
+ * interruption read model shares this set with resumption admission so it
+ * never advertises a resume the scheduler would refuse. */
+export const NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES = ["todo", "in_progress", "in_review"] as const;

server/src/services/task-interruption-presentation.ts

@@ -216,10 +216,15 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i
         or ${heartbeatRuns.errorCode} = 'workspace_validation_failed' then ${heartbeatRuns.error} else null end`,
       errorCode: heartbeatRuns.errorCode, livenessState: heartbeatRuns.livenessState, scheduledRetryAt: heartbeatRuns.scheduledRetryAt,
       scheduledRetryReason: heartbeatRuns.scheduledRetryReason,
+      // Candidate only: the coordinator also pushes nextAttemptAt forward while
+      // it defers an ineligible run. The scheduler's own admission check below
+      // decides whether this resume is real.
       nativeResumeAt: sql<Date | null>`(select ${nativeRunFinalizations.nextAttemptAt} from ${nativeRunFinalizations}
         where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id}
           and ${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'failed'
-          and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'retryable_failure')`.mapWith(value => value == null ? null : new Date(value)),
+          and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'retryable_failure'
+          and ${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null
+          and ${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration' is null)`.mapWith(value => value == null ? null : new Date(value)),
       // The run whose resultJson supplied providerFailure. A scheduled retry
       // shows its predecessor's reason, so redaction must use that run's registry.
       providerFailureRunId: sql<string>`case when ${heartbeatRuns.resultJson}->'providerFailure' is null
@@ -279,6 +284,38 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i
         inArray(approvals.status, ["pending", "revision_requested"]))),
     db.select({ status: companies.status, pauseReason: companies.pauseReason }).from(companies).where(eq(companies.id, companyId)),
   ]);
+  // Advertise a pending resume only when the scheduler would admit it, using
+  // its own check, so the task never hides Retry behind a deferred candidate
+  // and never offers Retry beside a resume that will run. Rare rows only.
+  // Candidates are failed native runs inside an active coordinator retry, so
+  // there are few. Their rows load in three batched reads; admission itself
+  // runs a few bounded reads per candidate, four at a time.
+  const resumeCandidates = runs.filter(run => run.nativeResumeAt);
+  if (resumeCandidates.length) {
+    const { readNativeSessionResumptionAgent } = await import("./native-runtime/native-session-resumption-admission.js");
+    const candidateIds = resumeCandidates.map(run => run.id);
+    const [fullRuns, coordinators] = await Promise.all([
+      db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.id, candidateIds))),
+      db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId),
+        inArray(nativeRunFinalizations.runId, candidateIds))),
+    ]);
+    const issueIds = [...new Set(fullRuns.flatMap(run => run.nativeIssueId ? [run.nativeIssueId] : []))];
+    const issueRows = issueIds.length ? await db.select().from(issues).where(and(eq(issues.companyId, companyId),
+      inArray(issues.id, issueIds))) : [];
+    const runById = new Map(fullRuns.map(run => [run.id, run]));
+    const coordinatorByRun = new Map(coordinators.map(row => [row.runId, row]));
+    const issueById = new Map(issueRows.map(row => [row.id, row]));
+    for (let start = 0; start < resumeCandidates.length; start += 4) {
+      await Promise.all(resumeCandidates.slice(start, start + 4).map(async candidate => {
+        const run = runById.get(candidate.id);
+        const issue = run?.nativeIssueId ? issueById.get(run.nativeIssueId) : undefined;
+        const coordinator = coordinatorByRun.get(candidate.id);
+        const admitted = run && issue && coordinator
+          && await readNativeSessionResumptionAgent(db, { issue, run, coordinator }).catch(() => null);
+        if (!admitted) candidate.nativeResumeAt = null;
+      }));
+    }
+  }
   const runIds = runs.map(run => run.id);
   const actions = runIds.length ? await db.selectDistinctOn([issueRecoveryActions.sourceIssueId]).from(issueRecoveryActions).where(and(
     eq(issueRecoveryActions.companyId, companyId), inArray(issueRecoveryActions.sourceIssueId, ids),