fix(heartbeat): cancel queued runs whose claim is permanently rejected (#14738)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - The heartbeat scheduler claims queued runs for each agent in
`startNextQueuedRunForAgent`, and startup recovery calls it through
`resumeQueuedRuns`
> - `claimQueuedRun` can throw a 4xx `HttpError` that the run's own rows
decide, for example the 403 from #14043 ("Queued-message interrupt
authority is unavailable")
> - The claim loop does not catch that error. The run stays `queued`, so
every recovery pass and every restart claims it again and gets the same
error
> - During periodic recovery the error is logged, but it stops the claim
loop, so all queued runs of every agent after it wait. During startup
recovery the error is rethrown, so the server cannot boot
> - We saw this on a self-hosted instance: one card-answer run from
#14043 put the server in a crash loop for about 11 hours (216 failed
boots) until the run was cancelled by hand in the database
> - This pull request cancels a queued run when its claim fails with a
403 that cannot clear, keeps a run queued when its 4xx rejection can
clear later, and in both cases continues with the rest of the queue
> - The benefit is that one bad queued run can no longer stop the server
from starting or stall other agents. #14510 fixes the specific identity
mismatch; this change is the safety net for this whole class of error

## Linked Issues or Issue Description

Refs #14043

Refs #14510, #13539, #13315

## What Changed

- `server/src/services/heartbeat.ts`: The claim loop in
`startNextQueuedRunForAgent` catches errors from `claimQueuedRun` for
each run.
- A 403 is a permanent rejection. In the claim path it only comes from
the run's persisted identity (an unverifiable interrupt receipt, a
manual wake without a user), and those rows do not change. The loop
cancels that run with error code `queued_run_claim_rejected`.
`cancelRunInternal` also cancels the wakeup request and releases the
issue execution lock.
- The cancel runs after the agent start lock is released (`.finally()`),
because `cancelRunInternal` promotes the next queued run under the same
lock. A cancel inside the lock waits 30 seconds for the stale-lock
timeout.
- Any other 4xx (for example a 422 `responsible_user_unresolved`, or a
409) can clear later. The run stays queued, the loop logs a warning, and
it continues with the next run. A later recovery pass tries the run
again.
- Non-HTTP errors and 5xx errors (for example database errors) keep
propagating, as before. Startup still fails when the database or another
dependency is broken.
- If a cancel itself fails, the loop logs the error. The run stays
queued for the next recovery pass.
- `server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts`:
New embedded-Postgres tests with a mocked adapter:
- A queued-comment interrupt run with an unverifiable receipt is
cancelled, `resumeQueuedRuns()` resolves, and a second pass stays clean.
- A claimable run behind a rejected run of the same agent is claimed and
executed.
- A run with an unresolvable responsible user (422) stays queued, and
the claimable run behind it is executed.
- A rejected run for one agent does not stop the queue of another agent.

## Verification

- `cd server && npx vitest run
src/__tests__/heartbeat-queued-run-claim-isolation.test.ts` passes.
- Without the change in `heartbeat.ts`, all four tests fail (the 403
`Queued-message interrupt authority is unavailable` and the 422
`responsible_user_unresolved` escape `resumeQueuedRuns()`).
- All `heartbeat*`, `*queued*` and `run-identity` server suites pass
locally (1252/1253). The one local failure, `heartbeat-process-recovery`
"redacts opaque environment-bound credentials from Sentry diagnostics",
fails the same way on unchanged `master` on my machine and passes in CI.
- `pnpm -r typecheck`, `pnpm test:run` and `pnpm build` pass locally.

## Risks

- Behavior shift: a queued run whose claim fails with a 403 is now
cancelled, not left queued. Such a run could never start before this
change, so no runnable work is lost. The cancel reason and the
`queued_run_claim_rejected` error code make the cancel visible on the
run.
- A run with another 4xx rejection stays queued, as before, but it no
longer stops the rest of the queue. It logs a warning on each recovery
pass until the cause clears.
- Low risk otherwise: no schema, API or UI change.

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use and
code execution.

## 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
- [x] All Paperclip CI gates are green
- [x] 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__/heartbeat-queued-run-claim-isolation.test.ts

@@ -0,0 +1,245 @@
+import { randomUUID } from "node:crypto";
+import { sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
+import {
+  agents,
+  agentWakeupRequests,
+  companies,
+  createDb,
+  heartbeatRuns,
+} from "@paperclipai/db";
+import {
+  getEmbeddedPostgresTestSupport,
+  startEmbeddedPostgresTestDatabase,
+} from "./helpers/embedded-postgres.js";
+import { heartbeatService } from "../services/heartbeat.ts";
+import { runningProcesses } from "../adapters/index.ts";
+
+const mockAdapterExecute = vi.hoisted(() =>
+  vi.fn(async () => ({
+    exitCode: 0,
+    signal: null,
+    timedOut: false,
+    errorMessage: null,
+    summary: "Queued-run claim isolation test run.",
+    provider: "test",
+    model: "test-model",
+  })),
+);
+
+vi.mock("../adapters/index.ts", async () => {
+  const actual = await vi.importActual<typeof import("../adapters/index.ts")>("../adapters/index.ts");
+  return {
+    ...actual,
+    getServerAdapter: vi.fn(() => ({
+      supportsLocalAgentJwt: false,
+      execute: mockAdapterExecute,
+    })),
+  };
+});
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
+
+if (!embeddedPostgresSupport.supported) {
+  console.warn(
+    `Skipping embedded Postgres queued-run claim isolation tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
+  );
+}
+
+describeEmbeddedPostgres("heartbeat queued-run claim isolation", () => {
+  let db!: ReturnType<typeof createDb>;
+  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+  let heartbeat!: ReturnType<typeof heartbeatService>;
+
+  beforeAll(async () => {
+    tempDb = await startEmbeddedPostgresTestDatabase("heartbeat-queued-run-claim-isolation-");
+    db = createDb(tempDb.connectionString);
+    heartbeat = heartbeatService(db);
+  }, 20_000);
+
+  afterEach(async () => {
+    await heartbeat.drainActiveRunExecutions();
+    mockAdapterExecute.mockClear();
+    runningProcesses.clear();
+    // Executed runs write to many company-scoped tables; clear them all.
+    await db.execute(sql`truncate table ${companies} cascade`);
+  });
+
+  afterAll(async () => {
+    await tempDb?.cleanup();
+  });
+
+  async function insertAgent() {
+    const companyId = randomUUID();
+    const agentId = randomUUID();
+
+    await db.insert(companies).values({
+      id: companyId,
+      name: "Claim Isolation Co",
+      status: "active",
+      issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
+      requireBoardApprovalForNewAgents: false,
+    });
+
+    await db.insert(agents).values({
+      id: agentId,
+      companyId,
+      name: "Claim Isolation Agent",
+      role: "engineer",
+      status: "idle",
+      adapterType: "codex_local",
+      adapterConfig: {},
+      runtimeConfig: {
+        heartbeat: {
+          enabled: true,
+          intervalSec: 60,
+          wakeOnDemand: true,
+          maxConcurrentRuns: 1,
+        },
+      },
+      permissions: {},
+    });
+
+    return { companyId, agentId };
+  }
+
+  async function insertQueuedRun(
+    companyId: string,
+    agentId: string,
+    wake: Partial<typeof agentWakeupRequests.$inferInsert>,
+    createdAt: Date,
+  ) {
+    const wakeupRequestId = randomUUID();
+    const runId = randomUUID();
+
+    await db.insert(agentWakeupRequests).values({
+      id: wakeupRequestId,
+      companyId,
+      agentId,
+      source: "on_demand",
+      triggerDetail: "manual",
+      status: "queued",
+      runId,
+      ...wake,
+    });
+    await db.insert(heartbeatRuns).values({
+      id: runId,
+      companyId,
+      agentId,
+      invocationSource: "on_demand",
+      triggerDetail: "manual",
+      status: "queued",
+      wakeupRequestId,
+      createdAt,
+    });
+
+    return { runId, wakeupRequestId };
+  }
+
+  // A queued-comment interrupt wake whose receipt cannot be verified. The run
+  // identity check rejects it with a 403 every time the run is claimed.
+  function insertUnverifiableInterruptRun(companyId: string, agentId: string, createdAt = new Date(Date.now() - 60_000)) {
+    return insertQueuedRun(companyId, agentId, {
+      reason: "issue_commented",
+      idempotencyKey: `queued-comment-interrupt:${randomUUID()}`,
+      requestedByActorType: "user",
+      requestedByActorId: "board-user",
+    }, createdAt);
+  }
+
+  // A system wake with no issue in a company without members. Its responsible
+  // user cannot be resolved yet (422), but a user added later would fix that.
+  function insertUnresolvableOwnerRun(companyId: string, agentId: string, createdAt = new Date(Date.now() - 60_000)) {
+    return insertQueuedRun(companyId, agentId, {
+      source: "automation",
+      triggerDetail: "system",
+      requestedByActorType: "system",
+      requestedByActorId: "heartbeat_test",
+    }, createdAt);
+  }
+
+  // A manual wake from a user. It resolves its identity from the receipt, so it
+  // can always be claimed.
+  function insertClaimableRun(companyId: string, agentId: string, createdAt = new Date()) {
+    return insertQueuedRun(companyId, agentId, {
+      requestedByActorType: "user",
+      requestedByActorId: "board-user",
+      payload: { manualUserWake: true },
+    }, createdAt);
+  }
+
+  async function runStatus(runId: string) {
+    return db
+      .select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, error: heartbeatRuns.error })
+      .from(heartbeatRuns)
+      .where(sql`${heartbeatRuns.id} = ${runId}`)
+      .then((rows) => rows[0] ?? null);
+  }
+
+  it("cancels a queued run whose claim is rejected instead of failing recovery", async () => {
+    const { companyId, agentId } = await insertAgent();
+    const { runId, wakeupRequestId } = await insertUnverifiableInterruptRun(companyId, agentId);
+
+    await expect(heartbeat.resumeQueuedRuns()).resolves.toBeUndefined();
+
+    expect(await runStatus(runId)).toMatchObject({
+      status: "cancelled",
+      errorCode: "queued_run_claim_rejected",
+      error: "Cancelled because the queued run cannot be claimed: Queued-message interrupt authority is unavailable",
+    });
+    const wakeup = await db
+      .select({ status: agentWakeupRequests.status })
+      .from(agentWakeupRequests)
+      .where(sql`${agentWakeupRequests.id} = ${wakeupRequestId}`)
+      .then((rows) => rows[0] ?? null);
+    expect(wakeup).toMatchObject({ status: "cancelled" });
+
+    // Recovery runs again on the next cycle and on every restart. The run must
+    // stay settled instead of failing the claim loop again.
+    await expect(heartbeat.resumeQueuedRuns()).resolves.toBeUndefined();
+    expect(await runStatus(runId)).toMatchObject({ status: "cancelled" });
+  });
+
+  it("claims the agent's next queued run after a rejected one", async () => {
+    const { companyId, agentId } = await insertAgent();
+    const { runId: rejectedRunId } = await insertUnverifiableInterruptRun(companyId, agentId);
+    const { runId: healthyRunId } = await insertClaimableRun(companyId, agentId);
+
+    await expect(heartbeat.resumeQueuedRuns()).resolves.toBeUndefined();
+    await heartbeat.drainActiveRunExecutions();
+
+    expect(await runStatus(rejectedRunId)).toMatchObject({
+      status: "cancelled",
+      errorCode: "queued_run_claim_rejected",
+    });
+    expect(mockAdapterExecute).toHaveBeenCalledOnce();
+    expect(await runStatus(healthyRunId)).toMatchObject({ status: "succeeded" });
+  });
+
+  it("keeps a recoverable rejection queued without blocking the runs behind it", async () => {
+    const { companyId, agentId } = await insertAgent();
+    const { runId: deferredRunId } = await insertUnresolvableOwnerRun(companyId, agentId);
+    const { runId: healthyRunId } = await insertClaimableRun(companyId, agentId);
+
+    await expect(heartbeat.resumeQueuedRuns()).resolves.toBeUndefined();
+    await heartbeat.drainActiveRunExecutions();
+
+    expect(await runStatus(deferredRunId)).toMatchObject({ status: "queued", errorCode: null });
+    expect(mockAdapterExecute).toHaveBeenCalledOnce();
+    expect(await runStatus(healthyRunId)).toMatchObject({ status: "succeeded" });
+  });
+
+  it("keeps processing other agents' queued runs after one agent's run is rejected", async () => {
+    const first = await insertAgent();
+    const second = await insertAgent();
+    const { runId: rejectedRunId } = await insertUnverifiableInterruptRun(first.companyId, first.agentId);
+    const { runId: healthyRunId } = await insertClaimableRun(second.companyId, second.agentId);
+
+    await expect(heartbeat.resumeQueuedRuns()).resolves.toBeUndefined();
+    await heartbeat.drainActiveRunExecutions();
+
+    expect(await runStatus(rejectedRunId)).toMatchObject({ status: "cancelled" });
+    expect(await runStatus(healthyRunId)).toMatchObject({ status: "succeeded" });
+  });
+});

server/src/services/heartbeat.ts

@@ -19878,9 +19878,51 @@ export function heartbeatService(
     }
   }
 
+  // A 403 from claimQueuedRun comes from the run's own persisted identity
+  // (an unverifiable interrupt receipt, a manual wake with no user). Those rows
+  // do not change, so the claim fails the same way on every pass and restart.
+  function isPermanentClaimRejection(err: unknown): err is HttpError {
+    return err instanceof HttpError && err.status === 403;
+  }
+
+  // Other 4xx rejections can clear later (a responsible user gets assigned, a
+  // conflicting claim finishes). Keep the run queued, but do not let it stop
+  // the rest of the queue or startup recovery.
+  function isDeferrableClaimRejection(err: unknown): err is HttpError {
+    return err instanceof HttpError && err.status >= 400 && err.status < 500;
+  }
+
+  // Settle runs that can never be claimed. Letting the error escape stalls the
+  // agent's queue and, during startup recovery, stops the server from booting.
+  async function cancelRejectedQueuedRuns(
+    rejected: Array<{ run: typeof heartbeatRuns.$inferSelect; err: HttpError }>,
+  ) {
+    for (const { run, err } of rejected) {
+      logger.warn(
+        { err, runId: run.id, agentId: run.agentId, companyId: run.companyId },
+        "cancelling queued heartbeat run whose claim was rejected",
+      );
+      try {
+        await cancelRunInternal(
+          run.id,
+          `Cancelled because the queued run cannot be claimed: ${err.message}`,
+          { errorCode: "queued_run_claim_rejected" },
+        );
+      } catch (cancelErr) {
+        logger.error(
+          { err: cancelErr, runId: run.id },
+          "failed to cancel queued heartbeat run whose claim was rejected; it stays queued for the next recovery pass",
+        );
+      }
+    }
+  }
+
   async function startNextQueuedRunForAgent(agentId: string) {
     if ((await getSchedulingSuppression()).suppressed) return [];
     const cutoff = await getWorktreeExecutionCutoff();
+    // Cancelled after the start lock is released: cancelRunInternal promotes the
+    // agent's next queued run, which takes this same lock.
+    const rejectedClaims: Array<{ run: typeof heartbeatRuns.$inferSelect; err: HttpError }> = [];
 
     return withAgentStartLock(agentId, async () => {
       const agent = await getAgent(agentId);
@@ -19992,7 +20034,21 @@ export function heartbeatService(
       const claimedRuns: Array<typeof heartbeatRuns.$inferSelect> = [];
       for (const queuedRun of prioritizedRuns) {
         if (claimedRuns.length >= availableSlots) break;
-        const claimed = await claimQueuedRun(queuedRun, companyAgents);
+        let claimed: typeof heartbeatRuns.$inferSelect | null;
+        try {
+          claimed = await claimQueuedRun(queuedRun, companyAgents);
+        } catch (err) {
+          if (isPermanentClaimRejection(err)) {
+            rejectedClaims.push({ run: queuedRun, err });
+            continue;
+          }
+          if (!isDeferrableClaimRejection(err)) throw err;
+          logger.warn(
+            { err, runId: queuedRun.id, agentId: queuedRun.agentId, companyId: queuedRun.companyId },
+            "queued heartbeat run claim was rejected; leaving it queued for the next recovery pass",
+          );
+          continue;
+        }
         if (claimed) claimedRuns.push(claimed);
       }
       if (claimedRuns.length === 0) return [];
@@ -20016,7 +20072,7 @@ export function heartbeatService(
         });
       }
       return claimedRuns;
-    });
+    }).finally(() => cancelRejectedQueuedRuns(rejectedClaims));
   }
 
   // Await every background heartbeat execution that is currently in flight. A