fix(heartbeat): bound the I7 queued-run check with a resumable cursor (#21022)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - #20986 added a reconciler check that reports a queued run whose
dispatch signal was lost (I7 `queued_run_not_dispatched`), and a worker
that re-drives such runs.
> - The Ponytail review of #20986 (4/5 on `fbab23cce6`) found three
issues. #20986 merged at that head before the fix commit was pushed.
> - The new I7 check read every queued run with no cap, while the report
still said that each I7 check reads at most 2,000 records.
> - This pull request adds that fix commit to master. The check now
reads a bounded slice per pass and continues from a cursor. The page
cursor no longer depends on the session date style. An agent with a
scheduled retry now counts as busy.
> - The benefit is that the reconciler pass stays bounded when many runs
are held on purpose, and the coverage statement is true again.

## Linked Issues or Issue Description

Refs #20986

Follow-up to the Ponytail review of #20986:
https://github.com/paperclipai/paperclip/pull/20986#pullrequestreview-5868879754
(review of `fbab23cce6`, score 4/5).

**What is wrong?**
1. `checkQueuedRunsDispatched`
(`server/src/services/recovery/ownership-invariant-report.ts`) paged
through every undispatched queued run with no cap.
`COVERAGE_GAPS.record_cap` still said each I7 check reads at most 2,000
records per pass.
2. The page cursor cast `created_at::text` back with `::timestamptz`.
That works only with the default `DateStyle`.
3. A queued run on an agent with a `scheduled_retry` run was reported
and re-driven, but `checkVisibleNextAction` already treats
`scheduled_retry` as an active run.

**Expected behavior**
The I7 queued-run check reads a bounded slice per pass and reaches every
run over later passes. The cursor format does not depend on session
settings. A scheduled retry counts as the agent's next action.

## What Changed

- `server/src/services/queued-run-dispatch-liveness.ts`: new
`readUndispatchedQueuedRuns`. It reads from an optional cursor until the
set ends or `maxRuns` rows were read, and returns the next cursor.
`listUndispatchedQueuedRuns` uses it with no limit. The cursor text is
`to_char(created_at at time zone 'UTC',
'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, which keeps microsecond precision and
does not depend on `DateStyle`. The live-run exclusion now also covers
`status = 'scheduled_retry'`.
- `server/src/services/recovery/ownership-invariant-report.ts`: the I7
queued-run check reads at most `MAX_RECORDS_PER_KIND` (2,000) per pass.
It keeps a keyset cursor in `OwnershipReportState.queuedRunCursor`, and
the next pass continues from that cursor. A pass that stops at the cap
keeps a follow-up pass due (`retryNeeded`), like the I3 and I4 backlogs.
`COVERAGE_GAPS.record_cap` now describes this check.
- `server/src/__tests__/ownership-reconciliation-report.test.ts`: new
test. Bounded slices resume from the cursor and end with `next: null`. A
`scheduled_retry` run on the agent removes its queued runs from the set.

## Verification

- `pnpm --filter @paperclipai/server exec vitest run
src/__tests__/ownership-reconciliation-report.test.ts
src/services/queued-run-dispatch-liveness.test.ts
src/services/heartbeat/dispatch-work.test.ts
src/__tests__/heartbeat-held-comment-wake.test.ts`: 75 tests pass on
this branch (fresh master + this commit).
- Server `tsc --noEmit` passes.
- The restart integration test from #20986 ("starts a continuation that
queued behind its predecessor's sandbox release after a graceful
restart") passed with this change on the #20986 branch, in both cases
(signalled and unsignalled).
- Restart-chaos CR1 360 s, Codex W2, `--daytona-warm
--ephemeral-server`, seed 9512 (`w-20261112T040314-s9512`, on the #20986
change set before this commit): the fault crossed the task and every
rule passed (R1–R6, T, U1–U5).

## Risks

- Low risk. The report reads less per pass. A very large held queue is
now covered over several passes instead of one.
- A queued run behind a `scheduled_retry` is no longer re-driven by the
backstop. The scheduled retry's own promotion signals dispatch.
- No schema 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`), Anthropic, via Claude Code with
tool use (shell, file edits, test runs).

## 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__/ownership-reconciliation-report.test.ts

@@ -34,7 +34,7 @@ import { applyOwnershipReportBackstops, registerOwnershipReconciliationWork } fr
 import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js";
 import { instanceSettingsService } from "../services/instance-settings.js";
-import { listUndispatchedQueuedRuns } from "../services/queued-run-dispatch-liveness.js";
+import { listUndispatchedQueuedRuns, readUndispatchedQueuedRuns } from "../services/queued-run-dispatch-liveness.js";
 import { releaseLapsedCoordinatorOwner } from "../services/workspace-recovery-stop-proof.js";
 import { subscribeDatabaseWork } from "@paperclipai/db/work-signals";
 import {
@@ -1072,6 +1072,38 @@ describe("report-only ownership reconciliation", () => {
     }
   });
 
+  it("reads undispatched queued runs in bounded slices that resume from a cursor, and skips an agent with a scheduled retry", async () => {
+    const created = [30, 29, 28].map(minutesAgo);
+    const runs = await db.insert(heartbeatRuns).values(created.map(createdAt =>
+      ({ companyId, agentId, status: "queued", invocationSource: "automation", createdAt }))).returning();
+    const first = await readUndispatchedQueuedRuns(db, { pageSize: 1, maxRuns: 2 });
+    expect(first.runs.map(run => run.runId)).toEqual([runs[0]!.id, runs[1]!.id]);
+    expect(first.next).not.toBeNull();
+    const rest = await readUndispatchedQueuedRuns(db, { pageSize: 1, maxRuns: 2, after: first.next });
+    expect(rest).toEqual({ runs: [expect.objectContaining({ runId: runs[2]!.id })], next: null });
+    await db.insert(heartbeatRuns).values({ companyId, agentId, status: "scheduled_retry", invocationSource: "automation",
+      scheduledRetryAt: new Date(Date.now() + 60_000) });
+    expect(await listUndispatchedQueuedRuns(db)).toEqual([]);
+  });
+
+  it("keeps the earliest grace deadline when the queued-run check reaches its cap, and rescans once it ends", async () => {
+    const createdAt = minutesAgo(1);
+    await db.insert(heartbeatRuns).values(Array.from({ length: 2_000 }, () =>
+      ({ companyId, agentId, status: "queued", invocationSource: "automation", createdAt })));
+    const dueAt = createdAt.getTime() + 5 * 60_000;
+    const state = createOwnershipReportState();
+    const queuedRuns = (report: OwnershipReport) => report.violations.filter(violation => violation.kind === "queued_run");
+    const first = await runOwnershipReconciliationPass(db, { state, record: false });
+    expect(queuedRuns(first)).toEqual([]);
+    expect(first.retryNeeded).toBe(true);
+    expect(first.nextDueAt?.getTime()).toBeLessThanOrEqual(dueAt);
+    const tail = await runOwnershipReconciliationPass(db, { state, record: false });
+    expect(queuedRuns(tail)).toEqual([]);
+    expect(tail.nextDueAt?.getTime()).toBeLessThanOrEqual(dueAt);
+    const overdue = await runOwnershipReconciliationPass(db, { state, record: false, now: new Date(dueAt + 1) });
+    expect(queuedRuns(overdue)).toHaveLength(2_000);
+  }, 60_000);
+
   it("signals queued-run dispatch when finalization work settles or a finished coordinator's lapsed owner is released", async () => {
     const issueId = randomUUID();
     const runId = randomUUID();

server/src/services/queued-run-dispatch-liveness.ts

@@ -18,7 +18,8 @@ export type UndispatchedQueuedRun = {
   createdAt: Date;
 };
 
-type PageCursor = { at: string; runId: string };
+export type UndispatchedQueuedRunCursor = { at: string; runId: string };
+type PageCursor = UndispatchedQueuedRunCursor;
 
 async function listUndispatchedQueuedRunPage(db: Db, companyId: string | undefined, createdAfter: Date | null | undefined,
   after: PageCursor | null, limit: number):
@@ -29,7 +30,7 @@ async function listUndispatchedQueuedRunPage(db: Db, companyId: string | undefin
     agentId: heartbeatRuns.agentId,
     issueId: sql<string | null>`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`,
     createdAt: heartbeatRuns.createdAt,
-    cursorAt: sql<string>`${heartbeatRuns.createdAt}::text`,
+    cursorAt: sql<string>`to_char(${heartbeatRuns.createdAt} at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`,
   }).from(heartbeatRuns)
     .innerJoin(companies, eq(companies.id, heartbeatRuns.companyId))
     .innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, heartbeatRuns.companyId)))
@@ -43,7 +44,7 @@ async function listUndispatchedQueuedRunPage(db: Db, companyId: string | undefin
       after ? sql`(${heartbeatRuns.createdAt}, ${heartbeatRuns.id}) > (${after.at}::timestamptz, ${after.runId}::uuid)` : undefined,
       sql`not exists (select 1 from ${heartbeatRuns} live
         where live.company_id = ${heartbeatRuns.companyId} and live.agent_id = ${heartbeatRuns.agentId}
-          and live.status = 'running')`,
+          and live.status in ('running', 'scheduled_retry'))`,
     ))
     .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id))
     .limit(limit);
@@ -55,15 +56,26 @@ async function listUndispatchedQueuedRunPage(db: Db, companyId: string | undefin
 export async function listUndispatchedQueuedRuns(db: Db,
   options: { companyId?: string; createdAfter?: Date | null; pageSize?: number } = {}):
   Promise<UndispatchedQueuedRun[]> {
+  return (await readUndispatchedQueuedRuns(db, options)).runs;
+}
+
+/** Reads from `after` until the set ends or `maxRuns` rows were read. `next`
+ * is where a later read continues, or null when the set was read to its end. */
+export async function readUndispatchedQueuedRuns(db: Db, options: { companyId?: string; createdAfter?: Date | null;
+  pageSize?: number; after?: UndispatchedQueuedRunCursor | null; maxRuns?: number } = {}):
+  Promise<{ runs: UndispatchedQueuedRun[]; next: UndispatchedQueuedRunCursor | null }> {
   const pageSize = options.pageSize ?? UNDISPATCHED_PAGE_SIZE;
+  const maxRuns = options.maxRuns ?? Number.POSITIVE_INFINITY;
   const runs: UndispatchedQueuedRun[] = [];
-  let after: PageCursor | null = null;
+  let after: PageCursor | null = options.after ?? null;
   for (;;) {
-    const page = await listUndispatchedQueuedRunPage(db, options.companyId, options.createdAfter, after, pageSize);
+    const want = Math.min(pageSize, maxRuns - runs.length);
+    const page = await listUndispatchedQueuedRunPage(db, options.companyId, options.createdAfter, after, want);
     runs.push(...page.map(({ cursorAt: _cursorAt, ...run }) => run));
-    if (page.length < pageSize) return runs;
+    if (page.length < want) return { runs, next: null };
     const last = page[page.length - 1]!;
     after = { at: last.cursorAt, runId: last.runId };
+    if (runs.length >= maxRuns) return { runs, next: after };
   }
 }
 

server/src/services/recovery/ownership-invariant-report.ts

@@ -15,7 +15,8 @@ import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js";
 import { logger } from "../../middleware/logger.js";
 import { getServerProcessStartedAt } from "../../server-info.js";
 import { persistActivity } from "../activity-log.js";
-import { listUndispatchedQueuedRuns, QUEUED_RUN_NOT_DISPATCHED, readQueuedRunExecutionWindow } from "../queued-run-dispatch-liveness.js";
+import { QUEUED_RUN_NOT_DISPATCHED, readQueuedRunExecutionWindow, readUndispatchedQueuedRuns,
+  type UndispatchedQueuedRunCursor } from "../queued-run-dispatch-liveness.js";
 import { currentProcessProbe, proveOwnerGone, type OwnerProof, type ProcessProbe } from "../owner-proof.js";
 import { LEGACY_OWNER_REASONS, recordVersionAttestations } from "./owner-gone-attestation.js";
 import {
@@ -103,7 +104,8 @@ const COVERAGE_GAPS: Record<string, string> = {
   I8_fast_turns: "Measured per run by the speed-budget check in the run log, not by this pass.",
   I7_scope: "I7 measures agent-assigned in_progress issues. todo, in_review and blocked issues are not measured yet.",
   record_cap: "Each kind and the I7 check read at most 2,000 records per pass; a larger kind is reported as truncated. "
-    + "The I3 and I4 scans read ten pages per pass and continue from a cursor on the next pass.",
+    + "The I3 and I4 scans read ten pages per pass and continue from a cursor on the next pass. The I7 queued-run "
+    + "check reads at most 2,000 undispatched queued runs per pass and continues from a cursor on the next pass.",
   wake_coverage: "Writers of workspace runtime services, remote workspace operations, snapshots, restore points, "
     + "git cache intents, provider artifacts and harness backups publish no work signal. Those kinds are read at "
     + "startup and on any other report wake. Phase 2 adds a signal when each kind moves to the shared settle path.",
@@ -127,11 +129,16 @@ export type OwnershipReportState = {
   admissionRunCursor: { at: string; id: string } | null;
   /** True when an I3 scan stopped with more rows to read. */
   admissionBacklog: boolean;
+  /** Keyset cursor of the I7 queued-run check (created_at, id); null starts over. */
+  queuedRunCursor: UndispatchedQueuedRunCursor | null;
+  /** Earliest grace deadline seen in the earlier slices of the current queued-run scan. */
+  queuedRunDueAt: Date | null;
 };
 
 export function createOwnershipReportState(): OwnershipReportState {
   return { goneLastPass: new Set(), reported: new Set(), intentCursor: null, unwritten: new Map(), intentBacklog: false,
-    admissionWakeCursor: null, admissionRunCursor: null, admissionBacklog: false };
+    admissionWakeCursor: null, admissionRunCursor: null, admissionBacklog: false, queuedRunCursor: null,
+    queuedRunDueAt: null };
 }
 
 const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"];
@@ -644,10 +651,18 @@ async function checkVisibleNextAction(db: Db, now: Date, limit: number): Promise
 /** A queued run counts as an issue's next action only while dispatch can still
  * start it. One that stays queued past the grace window while its ready agent
  * runs nothing has lost its dispatch signal (invariant 7). */
-async function checkQueuedRunsDispatched(db: Db, now: Date, limit: number): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> {
+async function checkQueuedRunsDispatched(db: Db, state: OwnershipReportState, now: Date, limit: number):
+  Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> {
   const window = await readQueuedRunExecutionWindow(db);
-  if (!window.dispatches) return { violations: [], nextDueAt: null };
-  const runs = await listUndispatchedQueuedRuns(db, { pageSize: limit, createdAfter: window.createdAfter });
+  if (!window.dispatches) {
+    state.queuedRunCursor = null; state.queuedRunDueAt = null;
+    return { violations: [], nextDueAt: null };
+  }
+  if (state.queuedRunDueAt && state.queuedRunDueAt <= now) { state.queuedRunCursor = null; state.queuedRunDueAt = null; }
+  const { runs, next } = await readUndispatchedQueuedRuns(db, { pageSize: limit, createdAfter: window.createdAfter,
+    after: state.queuedRunCursor, maxRuns: MAX_RECORDS_PER_KIND });
+  state.queuedRunCursor = next;
+  const carried = state.queuedRunDueAt;
   const cutoff = now.getTime() - QUEUED_RUN_DISPATCH_GRACE_MS;
   const violations: InvariantViolation[] = [];
   let nextDueAt: Date | null = null;
@@ -659,6 +674,8 @@ async function checkQueuedRunsDispatched(db: Db, now: Date, limit: number): Prom
       nextDueAt ??= new Date(run.createdAt.getTime() + QUEUED_RUN_DISPATCH_GRACE_MS);
     }
   }
+  if (carried && (!nextDueAt || carried < nextDueAt)) nextDueAt = carried;
+  state.queuedRunDueAt = next ? nextDueAt : null;
   return { violations, nextDueAt };
 }
 
@@ -777,7 +794,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
       return result.violations;
     },
     async () => {
-      const result = await checkQueuedRunsDispatched(db, now, limit);
+      const result = await checkQueuedRunsDispatched(db, state, now, limit);
       if (result.nextDueAt && (!nextDueAt || result.nextDueAt < nextDueAt)) nextDueAt = result.nextDueAt;
       return result.violations;
     },
@@ -805,7 +822,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
 
   // Only a first proof needs a second pass; a record already reported for I2
   // is not a reason to keep scanning.
-  if (state.intentBacklog || state.admissionBacklog) retryNeeded = true;
+  if (state.intentBacklog || state.admissionBacklog || state.queuedRunCursor) retryNeeded = true;
   const awaitingSecondProof = [...goneThisPass].filter(key => !state.goneLastPass.has(key)).length;
   state.goneLastPass = goneThisPass;
   // Violations whose activity row failed last time are written again, even