Integrate bounded issue recovery and canonical scope hints

server/src/__tests__/company-portability.test.ts

@@ -4168,6 +4168,7 @@ describe("company portability", () => {
     const insertedRelationValues: Array<Record<string, unknown>> = [];
     const monitorUpdates: Array<Record<string, unknown>> = [];
     const db = {
+      transaction: async (write: (tx: unknown) => Promise<unknown>) => write(db),
       insert: () => ({
         values: (rows: Array<Record<string, unknown>>) => ({
           onConflictDoNothing: async () => {

server/src/__tests__/issue-recovery-deadlines.test.ts

@@ -2,14 +2,18 @@ import { idleWorkSnapshot } from "../services/task-admission.js";
 import { randomUUID } from "node:crypto";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
-import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issues, issueWatchdogs } from "@paperclipai/db";
+import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issueRecoveryActions, issueRelations, issues, issueWatchdogs } from "@paperclipai/db";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { heartbeatService } from "../services/heartbeat.js";
 import { issueService } from "../services/issues.js";
+import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
+import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js";
 import { taskWatchdogService } from "../services/task-watchdogs.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 
+vi.setConfig({ testTimeout: 90_000 });
+
 const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSUE_RECOVERY_TEST_DATABASE_URL");
 const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
 describe.skipIf(!support.supported)("issue recovery production producers and deadlines", () => {
@@ -81,13 +85,14 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
       expect(wake, detail).toBeNull();
       expect(extraDeadlines.map(next => next())).toEqual(extraDeadlines.map(() => null));
       expect(idleWorkSnapshot().active).toBe(0);
-    }, { timeout: 10_000 });
+    }, { timeout: 40_000 });
   }
   async function quiet() {
     await settled();
     const reads = vi.spyOn(db, "select"); const executes = vi.spyOn(db, "execute");
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
-    try { await vi.advanceTimersByTimeAsync(20 * 60_000); expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); }
+    const selected = reads.mock.calls.length, executed = executes.mock.calls.length;
+    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try { await vi.advanceTimersByTimeAsync(20 * 60_000); expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed); }
     finally { vi.useRealTimers(); reads.mockRestore(); executes.mockRestore(); }
   }
   async function finish(issueId: string) {
@@ -95,12 +100,10 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     await heartbeat.drainActiveRunExecutions();
     const saved = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
       sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`));
+    expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0);
+    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     if (saved.length > 0) {
       const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime()));
-      expect(await db.select().from(heartbeatRuns).where(and(
-        sql`${heartbeatRuns.status} in ('queued', 'running')`,
-        sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`))).toHaveLength(0);
-      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
       await vi.advanceTimersByTimeAsync(Math.max(0, due - Date.now()) + 1);
       await vi.waitFor(async () => {
         const remaining = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
@@ -120,20 +123,85 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
       const run = runs.find(row => row.status === "running" || row.status === "succeeded");
       expect(run).toBeDefined(); runId = run!.id;
     }, { timeout: 15_000 });
-    await heartbeat.drainActiveRunExecutions();
+    await vi.waitFor(async () => {
+      await heartbeat.drainActiveRunExecutions();
+      const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
+      expect(current.status, JSON.stringify({ error: current.error, code: current.errorCode, stderr: current.stderrExcerpt })).toBe("succeeded");
+    }, { timeout: 15_000 });
     const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
-    expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode, stderr: run.stderrExcerpt, result: run.resultJson })).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("recovery-proof");
+    expect(run.stdoutExcerpt).toContain("recovery-proof");
     return run;
   }
   it("keeps empty, configured idle, and terminal history quiet", async () => {
     await fixture(); await start(); await quiet();
   });
+  it("creates a blocking child through the ordinary root service transaction", async () => {
+    const f = await fixture();
+    const parent = await issueService(db).create(f.companyId, { title: "Parent", status: "todo" });
+    const child = await issueService(db).createChild(parent.id, { title: "Blocking child", status: "todo", blockParentUntilDone: true });
+    expect(child.parentBlockerAdded).toBe(true);
+    expect(await db.select().from(issueRelations).where(and(eq(issueRelations.companyId, f.companyId),
+      eq(issueRelations.issueId, child.issue.id), eq(issueRelations.relatedIssueId, parent.id)))).toHaveLength(1);
+  });
+  it.each(["startup", "committed"])("restores blocker-only and completed-action debt with no assignment candidates (%s)", async mode => {
+    const f = await fixture();
+    if (mode === "committed") { await start(); await quiet(); }
+    const blocker = await issueService(db).create(f.companyId, { title: "Orphan blocker", status: "blocked", createdByAgentId: f.agentId });
+    const dependent = await issueService(db).create(f.companyId, { title: "Blocked dependent", status: "blocked", blockedByIssueIds: [blocker.id] });
+    const completedSource = await issueService(db).create(f.companyId, { title: "Completed source", status: "done" });
+    const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: completedSource.id,
+      kind: "stranded_assigned_issue", ownerType: "agent", ownerAgentId: f.agentId, cause: "stranded_assigned_issue",
+      fingerprint: "source-terminal", nextAction: "Restore work", wakePolicy: { type: "wake_owner" }, evidence: {} });
+    if (mode === "startup") await start();
+    await vi.waitFor(async () => {
+      expect((await db.select().from(issues).where(eq(issues.id, blocker.id)))[0].assigneeAgentId).toBe(f.agentId);
+      expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0].status).toBe("resolved");
+    }, { timeout: 40_000 });
+    await finish(blocker.id); await finish(dependent.id); await quiet();
+  });
+  it("bounds subsidiary scans for100 assignment candidates and coalesces repeated company writes", async () => {
+    const f = await fixture(), foreign = await fixture();
+    await db.update(agents).set({ lifecycleState: "preparing" }).where(eq(agents.id, f.agentId));
+    await db.insert(issues).values(Array.from({ length: 100 }, (_, index) => ({ companyId: f.companyId,
+      title: `Waiting setup ${index}`, status: "todo", assigneeAgentId: f.agentId })));
+    const [source] = await db.insert(issues).values({ companyId: f.companyId, title: "Done source", status: "done" }).returning();
+    const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: source.id,
+      kind: "stranded_assigned_issue", ownerType: "agent", ownerAgentId: f.agentId, cause: "stranded_assigned_issue",
+      fingerprint: "query-budget", nextAction: "Restore work", wakePolicy: { type: "wake_owner" }, evidence: {} });
+    const scans = { blockers: 0, actions: 0, loads: 0 };
+    const originalSelect = db.select.bind(db);
+    vi.spyOn(db, "select").mockImplementation(((fields?: Record<string, unknown>) => {
+      const query = originalSelect(fields as never); const from = query.from.bind(query);
+      query.from = ((table: unknown) => {
+        if (table === issues && fields?.id === issues.id && fields.companyId === issues.companyId && Object.keys(fields).length === 2) scans.loads++;
+        if (table === issueRelations && fields?.createdByAgentId === issues.createdByAgentId) scans.blockers++;
+        if (table === issueRecoveryActions && fields?.action === issueRecoveryActions) scans.actions++;
+        return from(table as never);
+      }) as typeof query.from;
+      return query;
+    }) as typeof db.select);
+    await start(); await settled();
+    expect(scans.actions).toBe(1); expect(scans.blockers).toBe(0); expect(scans.loads).toBeLessThanOrEqual(8);
+    expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0].status).toBe("resolved");
+    const [foreignSource] = await db.insert(issues).values({ companyId: foreign.companyId, title: "Foreign done", status: "done" }).returning();
+    const [foreignAction] = await db.insert(issueRecoveryActions).values({ companyId: foreign.companyId, sourceIssueId: foreignSource.id,
+      kind: "stranded_assigned_issue", ownerType: "board", cause: "stranded_assigned_issue", fingerprint: "foreign-query-budget", nextAction: "Inspect" }).returning();
+    const before = { ...scans };
+    for (let index = 0; index < 100; index++) await db.transaction(tx => notifyIssueRecovery(tx, { companyId: f.companyId }));
+    await new Promise(resolve => setTimeout(resolve, 100));
+    expect(scans.loads).toBe(before.loads);
+    await settled();
+    expect(scans.actions).toBe(before.actions); expect(scans.blockers).toBe(before.blockers);
+    expect(scans.loads).toBeLessThanOrEqual(before.loads + 4);
+    expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, foreignAction.id)))[0].status).toBe("active");
+    expect(await originalSelect().from(heartbeatRuns)).toHaveLength(0); await quiet();
+  });
   it("dispatches an ordinary new assignment after commit and returns to quiet after completion", async () => {
     const f = await fixture(); await start(); await quiet();
     const issue = await issueService(db).create(f.companyId, { title: "Recover assignment", status: "todo", assigneeAgentId: f.agentId });
     await completed(issue.id);
     await finish(issue.id); await quiet();
-  }, 45_000);
+  }, 90_000);
   it("reacts to blocker resolution once and dispatches the durable dependency wake", async () => {
     const f = await fixture();
     const blocker = await issueService(db).create(f.companyId, { title: "Blocker", status: "todo" });
@@ -147,7 +215,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     await finish(dependent.id); await quiet();
     expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
       sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'issue_blockers_resolved'`))).toHaveLength(1);
-  }, 45_000);
+  }, 90_000);
   it("uses ordinary watchdog configuration and subtree closure to dispatch a reusable review", async () => {
     const f = await fixture();
     const source = await issueService(db).create(f.companyId, { title: "Finished subtree", status: "done" });
@@ -163,8 +231,9 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
       sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'task_watchdog_stopped_subtree'`))).toHaveLength(1);
     await finish(reviewId!);
-    await taskWatchdogService(db).disableForIssue(f.companyId, source.id); await quiet();
-  }, 45_000);
+    await taskWatchdogService(db).disableForIssue(f.companyId, source.id);
+    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await quiet();
+  }, 90_000);
   it("bounds ordinary writes to their company and reconstructs foreign stale debt after restart", async () => {
     const selected = await fixture(); await start(); await quiet();
     const foreign = await fixture();
@@ -198,19 +267,19 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     })).rejects.toThrow("rollback control");
     await quiet();
     expect(await db.select().from(issues).where(eq(issues.title, "Nested rollback"))).toHaveLength(0);
-  }, 60_000);
+  }, 90_000);
   it("retains the first committed assignment during drain and executes on resume", async () => {
     const f = await fixture(); await start(); await quiet(); draining = true;
     const issue = await issueService(db).create(f.companyId, { title: "Drained assignment", status: "todo", assigneeAgentId: f.agentId });
     await new Promise(resolve => setTimeout(resolve, 100));
     const reads = vi.spyOn(db, "select");
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     try { await vi.advanceTimersByTimeAsync(90_000); expect(reads).not.toHaveBeenCalled(); }
     finally { vi.useRealTimers(); reads.mockRestore(); }
     expect(await db.select().from(heartbeatRuns)).toHaveLength(0);
     draining = false; worker.wake(); dispatch.wake();
     await completed(issue.id); await finish(issue.id); await quiet();
-  }, 45_000);
+  }, 90_000);
   it("retains a fresh subtree's exact first-run grace then disarms on disable", async () => {
     const f = await fixture();
     const source = await issueService(db).create(f.companyId, { title: "Fresh unassigned subtree", status: "todo" });
@@ -218,7 +287,8 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     await taskWatchdogService(db).upsertForIssue(f.companyId, source.id, { agentId: f.agentId });
     await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(source.createdAt.getTime() + 15_000), { timeout: 5_000 });
     expect(await db.select().from(issues).where(eq(issues.originKind, "task_watchdog"))).toHaveLength(0);
-    await taskWatchdogService(db).disableForIssue(f.companyId, source.id); await quiet();
+    await taskWatchdogService(db).disableForIssue(f.companyId, source.id);
+    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await quiet();
   });
   it("waits for a live foreign controller's exact stored lease before clearing a terminal lock", async () => {
     const f = await fixture();
@@ -232,6 +302,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(lease.getTime()), { timeout: 3_000 });
     expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("running");
     await vi.waitFor(async () => expect((await db.select().from(issues).where(eq(issues.id, source.id)))[0].executionRunId).toBeNull(), { timeout: 12_000 });
-    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("succeeded"); await quiet();
-  }, 30_000);
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("succeeded");
+    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await quiet();
+  }, 90_000);
 });

server/src/__tests__/issue-recovery-work-lifecycle.test.ts

@@ -1,6 +1,6 @@
 import { PgDialect } from "drizzle-orm/pg-core";
 import { afterEach, describe, expect, it, vi } from "vitest";
-import { issueWatchdogs } from "@paperclipai/db";
+import { issueRecoveryActions, issueWatchdogs } from "@paperclipai/db";
 import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js";
@@ -12,29 +12,34 @@ afterEach(async () => { for (const stop of stops.splice(0)) await stop(); vi.use
 function fixture() {
   vi.useFakeTimers(); vi.setSystemTime(1_000_000);
   const dialect = new PgDialect();
-  const rows = { issues: [] as Array<{ id: string; companyId: string }>, watchdogs: [] as Array<{ id: string; companyId: string }> };
+  const rows = { issues: [] as Array<{ id: string; companyId: string }>, watchdogs: [] as Array<{ id: string; companyId: string }>, blockers: [] as Array<{ id: string; companyId: string }>, actions: [] as Array<{ id: string; companyId: string }> };
   const execute = vi.fn(async (query: any) => {
     if (dialect.sqlToQuery(query).sql.includes("pg_current_xact_id")) return [{ xid: "42" }];
     throw new Error("unexpected SQL");
   });
-  const load = vi.fn((table: unknown) => table === issueWatchdogs ? rows.watchdogs : rows.issues);
+  const load = vi.fn((table: unknown, distinct = false) => distinct
+    ? table === issueRecoveryActions ? rows.actions : rows.blockers
+    : table === issueWatchdogs ? rows.watchdogs : rows.issues);
+  const query = (distinct = false) => ({ from: (table: unknown) => ({ where: () => ({ orderBy: () => ({ limit: async () => load(table, distinct) }) }) }) });
   const db: any = installDatabaseWorkSignals({ execute,
-    select: () => ({ from: (table: unknown) => ({ where: async () => load(table) }) }),
-    transaction: async (callback: (tx: any) => Promise<unknown>) => callback({ execute, select: db.select, transaction: db.transaction }),
+    select: () => query(), selectDistinct: () => query(true),
+    transaction: async (callback: (tx: any) => Promise<unknown>) => callback({ execute, select: db.select, selectDistinct: db.selectDistinct, transaction: db.transaction }),
   });
   const state = { draining: false, standby: false, suppressed: false, deadline: null as number | null, fail: false, incomplete: false };
   const check = vi.fn(async (scope: { onDeferredDeadline(at: number): void }) => {
     if (state.fail) throw new Error("database unavailable");
     if (state.deadline !== null && state.deadline > Date.now()) scope.onDeferredDeadline(state.deadline);
   });
-  const owner = { stranded: check, dependencies: async (scope: any) => { await check(scope); return { enqueueFailed: 0 }; },
+  const orphanBlockers = vi.fn(async () => {}), recoveryActions = vi.fn(async () => {});
+  const suppression = vi.fn(async () => state.suppressed);
+  const owner = { orphanBlockers, recoveryActions, stranded: check, dependencies: async (scope: any) => { await check(scope); return { enqueueFailed: 0 }; },
     locks: check, watchdogs: async (scope: any) => { await check(scope); return { incomplete: state.incomplete ? 1 : 0 }; },
-    schedulingSuppressed: async () => state.suppressed };
+    schedulingSuppressed: suppression };
   const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !state.standby,
     canReconcile: () => !state.standby, onError: vi.fn() });
   const worker = registerIssueRecoveryWork(db, coordinator, owner, () => !state.draining && !state.standby);
   stops.push(() => coordinator.stop());
-  return { db, rows, state, check, load, coordinator, worker,
+  return { db, rows, state, check, load, coordinator, worker, orphanBlockers, recoveryActions, suppression,
     notify: () => db.transaction((tx: any) => notifyIssueRecovery(tx, { companyId: "selected" })),
     settle: () => vi.advanceTimersByTimeAsync(1_000) };
 }
@@ -45,6 +50,51 @@ describe("issue recovery obligation lifecycle", () => {
     await vi.advanceTimersByTimeAsync(20 * 60_000);
     expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).not.toHaveBeenCalled();
   });
+  it("reads scheduling suppression once per pass instead of once per issue", async () => {
+    const f = fixture();
+    f.rows.issues.push(...Array.from({ length: 25 }, (_, index) => ({ id: `issue-${index}`, companyId: "selected" })));
+    await f.worker.ready;
+    expect(f.check).toHaveBeenCalledTimes(75);
+    expect(f.suppression).toHaveBeenCalledTimes(6);
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("backs off a failed suppression read without probing an empty or failed owner repeatedly", async () => {
+    const f = fixture(); f.suppression.mockRejectedValueOnce(new Error("settings unavailable"));
+    await f.worker.ready; const reads = f.suppression.mock.calls.length;
+    expect(f.coordinator.nextWakeAt()).toBe(1_030_000);
+    await vi.advanceTimersByTimeAsync(29_999); expect(f.suppression).toHaveBeenCalledTimes(reads);
+    await vi.advanceTimersByTimeAsync(1); expect(f.suppression).toHaveBeenCalledTimes(reads + 1);
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("normalizes a wide scope with cyclic SQL objects before publishing work hints", async () => {
+    const f = fixture(); await f.worker.ready;
+    const wide = { companyId: "selected", endpointId: issueRecoveryActions.id, table: issueRecoveryActions };
+    await f.db.transaction((tx: object) => notifyIssueRecovery(tx, wide)); await f.settle();
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+    const loads = f.load.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.load).toHaveBeenCalledTimes(loads);
+  });
+  it("runs orphan blocker and action duties without assignment entries", async () => {
+    const f = fixture(); f.rows.blockers.push({ id: "selected", companyId: "selected" });
+    f.rows.actions.push({ id: "selected", companyId: "selected" }); await f.worker.ready;
+    expect(f.check).not.toHaveBeenCalled();
+    expect(f.orphanBlockers).toHaveBeenCalledExactlyOnceWith({ companyId: "selected" });
+    expect(f.recoveryActions).toHaveBeenCalledExactlyOnceWith({ companyId: "selected" });
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("coalesces repeated company reconstruction without delaying an existing lease deadline", async () => {
+    const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" });
+    f.state.deadline = 1_015_000; await f.worker.ready;
+    await f.notify(); await f.settle(); const loaded = f.load.mock.calls.length, checked = f.check.mock.calls.length;
+    for (let index = 0; index < 100; index++) { await f.notify(); await vi.advanceTimersByTimeAsync(10); }
+    expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).toHaveBeenCalledTimes(checked);
+    await vi.advanceTimersByTimeAsync(13_000);
+    expect(f.state.deadline).toBe(1_015_000);
+    expect(f.coordinator.nextWakeAt()).toBe(1_031_000);
+    await vi.advanceTimersByTimeAsync(16_000);
+    expect(f.load.mock.calls.length).toBeGreaterThan(loaded);
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
   it("retains exact grace and lease deadlines without early owner probes", async () => {
     const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" });
     f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); f.state.deadline = 1_015_000;
@@ -85,7 +135,7 @@ describe("issue recovery obligation lifecycle", () => {
     const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); await f.worker.ready;
     let release!: () => void; const held = new Promise<void>(resolve => { release = resolve; });
     f.check.mockImplementationOnce(() => held);
-    await f.notify(); await f.settle(); await f.notify(); release(); await f.settle(); await f.settle();
+    await f.notify(); await f.settle(); await f.notify(); release(); await f.settle(); await vi.advanceTimersByTimeAsync(30_000);
     expect(f.coordinator.nextWakeAt()).toBeNull();
     expect(f.check.mock.calls.length).toBeGreaterThan(3);
   });

server/src/services/delivery-work-notifications.ts

@@ -2,6 +2,8 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  orphanBlockers: "orphan-blockers",
+  activeRecoveryActions: "active-recovery-actions",
   strandedAssignments: "stranded-assignments",
   resolvedDependencies: "resolved-dependencies",
   taskWatchdogs: "task-watchdogs",

server/src/services/heartbeat.ts

@@ -4245,7 +4245,9 @@ export function heartbeatService(
 
     registerIssueStateRecoveryWork: (coordinator: Parameters<typeof registerIssueRecoveryWork>[1], canRun: () => boolean) =>
       registerIssueRecoveryWork(db, coordinator, {
-        stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true }),
+        stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true, skipRelatedRecovery: true }),
+        orphanBlockers: recovery.reconcileUnassignedBlockingIssues,
+        recoveryActions: recovery.reconcileActiveRecoveryActions,
         dependencies: reconcileResolvedDependencyWakes,
         locks: sweepStaleIssueLocks,
         watchdogs: reconcileTaskWatchdogs,

server/src/services/issue-recovery-notifications.ts

@@ -3,8 +3,8 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type IssueRecoveryScope = { companyId: string };
 
 export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) {
-  const hint = scope ? JSON.stringify(scope) : "*";
-  for (const queue of [DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies,
+  const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
+  for (const queue of [DELIVERY_QUEUES.orphanBlockers, DELIVERY_QUEUES.activeRecoveryActions, DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies,
     DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) {
     await notifyDeliveryWork(transaction, queue, hint);
   }

server/src/services/native-runtime/native-workspace-finalizer.test.ts

@@ -131,7 +131,7 @@ describe("native workspace finalization recovery", () => {
     const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_WORKSPACE_FINALIZER_TEST_DATABASE_URL");
     if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-workspace-finalizer-");
     connectionString = external ?? temporary!.connectionString;
-    db = createDb(connectionString);
+    db = createDb(connectionString, { maxConnections: 6 });
 
     await db.insert(companies).values([
       { id: companyId, name: "Native workspace owner", issuePrefix: "NWO" },
@@ -452,7 +452,7 @@ describe("native workspace finalization recovery", () => {
     const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Live copyback" });
     const live = await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async (ownership) => {
       await ownership.assertHeld();
-      expect(await resumeNativeWorkspaceFinalization({ db: createDb(connectionString), runId: seeded.runId })).toBeNull();
+      expect(await resumeNativeWorkspaceFinalization({ db: createDb(connectionString, { maxConnections: 6 }), runId: seeded.runId })).toBeNull();
       expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, seeded.runId))).toEqual([]);
       return "copied";
     });
@@ -476,7 +476,7 @@ describe("native workspace finalization recovery", () => {
     try {
       await entered;
       await db.execute(sql`select pg_terminate_backend(pid) from pg_locks where locktype = 'advisory' and database = (select oid from pg_database where datname = current_database()) and pid <> pg_backend_pid()`);
-      const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "must not run");
+      const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString, { maxConnections: 6 }), companyId, runId: seeded.runId }, async () => "must not run");
       expect(second).toEqual({ acquired: false, reason: "busy" });
       const [held] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
       expect(held.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toBeTruthy();
@@ -489,7 +489,7 @@ describe("native workspace finalization recovery", () => {
 
   it("recovers its own joined copyback receipt after the application pool disconnects during cleanup", async () => {
     const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Reconnect joined copyback" });
-    const interruptedDb = createDb(connectionString);
+    const interruptedDb = createDb(connectionString, { maxConnections: 6 });
     let joined = false;
     await expect(withNativeWorkspaceFinalizationOwnership({ db: interruptedDb, companyId, runId: seeded.runId }, async () => {
       await fs.writeFile(path.join(workspaceRoot, "joined-copyback.txt"), "durable work");
@@ -507,7 +507,7 @@ describe("native workspace finalization recovery", () => {
         .toEqual({ acquired: false, reason: "owner_unverified" });
     }
     await db.update(heartbeatRuns).set({ runnerProfileJson: beforeReconnect.runnerProfileJson }).where(eq(heartbeatRuns.id, seeded.runId));
-    expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "resumed"))
+    expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString, { maxConnections: 6 }), companyId, runId: seeded.runId }, async () => "resumed"))
       .toEqual({ acquired: true, value: "resumed" });
   });
 

server/src/services/recovery/service.ts

@@ -2203,7 +2203,7 @@ export function recoveryService(
     return Boolean(budgetBlock);
   }
 
-  async function reconcileUnassignedBlockingIssues() {
+  async function reconcileUnassignedBlockingIssues(opts?: { companyId?: string }) {
     const candidates = await db
       .select({
         id: issues.id,
@@ -2217,6 +2217,8 @@ export function recoveryService(
       .where(
         and(
           eq(issueRelations.type, "blocks"),
+          eq(issueRelations.companyId, issues.companyId),
+          opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined,
           inArray(issues.status, ["todo", "blocked"]),
           isNull(issues.assigneeAgentId),
           isNull(issues.assigneeUserId),
@@ -3501,7 +3503,7 @@ export function recoveryService(
     return Boolean(run || wake);
   }
 
-  async function reconcileActiveRecoveryActions() {
+  async function reconcileActiveRecoveryActions(opts?: { companyId?: string }) {
     const rows = await db
       .select({
         action: issueRecoveryActions,
@@ -3515,7 +3517,8 @@ export function recoveryService(
           eq(issues.companyId, issueRecoveryActions.companyId),
         ),
       )
-      .where(inArray(issueRecoveryActions.status, ["active", "escalated"]));
+      .where(and(inArray(issueRecoveryActions.status, ["active", "escalated"]),
+        opts?.companyId ? eq(issueRecoveryActions.companyId, opts.companyId) : undefined));
 
     const result = {
       requeued: 0,
@@ -4532,6 +4535,7 @@ export function recoveryService(
     onDeferredDeadline?: (at: number) => void;
     issueCreatedAtGte?: Date | null;
     skipExternalConversationSettlement?: boolean;
+    skipRelatedRecovery?: boolean;
   }) {
     const candidates = await db
       .select(recoveryIssueFields)
@@ -5780,16 +5784,18 @@ export function recoveryService(
       }
     }
 
-    const orphanBlockerRecovery = await reconcileUnassignedBlockingIssues();
-    result.orphanBlockersAssigned = orphanBlockerRecovery.assigned;
-    result.skipped += orphanBlockerRecovery.skipped;
-    result.issueIds.push(...orphanBlockerRecovery.issueIds);
+    if (!opts?.skipRelatedRecovery) {
+      const orphanBlockerRecovery = await reconcileUnassignedBlockingIssues(opts);
+      result.orphanBlockersAssigned = orphanBlockerRecovery.assigned;
+      result.skipped += orphanBlockerRecovery.skipped;
+      result.issueIds.push(...orphanBlockerRecovery.issueIds);
 
-    const activeRecovery = await reconcileActiveRecoveryActions();
-    result.continuationRequeued += activeRecovery.requeued;
-    result.escalated += activeRecovery.escalated;
-    result.skipped += activeRecovery.skipped;
-    result.issueIds.push(...activeRecovery.issueIds);
+      const activeRecovery = await reconcileActiveRecoveryActions(opts);
+      result.continuationRequeued += activeRecovery.requeued;
+      result.escalated += activeRecovery.escalated;
+      result.skipped += activeRecovery.skipped;
+      result.issueIds.push(...activeRecovery.issueIds);
+    }
     result.issueIds = [...new Set(result.issueIds)];
 
     return result;
@@ -6563,6 +6569,8 @@ export function recoveryService(
     recordWatchdogDecision,
     scanSilentActiveRuns,
     reconcileStrandedAssignedIssues,
+    reconcileUnassignedBlockingIssues,
+    reconcileActiveRecoveryActions,
     reconcileLegacyContinuation,
     legacyRepairDispatchBlock,
     sweepStaleIssueLocks,

server/src/services/recovery/work.ts

@@ -1,6 +1,6 @@
 import { registerTaskWatchdogWork } from "../task-watchdog-work.js";
-import { and, eq, inArray, isNotNull, isNull, or } from "drizzle-orm";
-import { issues, type Db } from "@paperclipai/db";
+import { and, asc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
+import { issueRecoveryActions, issueRelations, issues, type Db } from "@paperclipai/db";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { DELIVERY_QUEUES, type DeliveryQueue } from "../delivery-work-notifications.js";
 import type { IssueRecoveryScope } from "../issue-recovery-notifications.js";
@@ -8,12 +8,15 @@ import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
 
 type Entry = { id: string; companyId: string; dueAt: number };
 type Scope = { companyId: string; issueId: string; onDeferredDeadline: (at: number) => void };
+const PAGE_SIZE = 100;
 
 export function registerIssueRecoveryWork(
   db: Db,
   coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
   owner: {
     stranded(scope: Scope): Promise<unknown>;
+    orphanBlockers(scope: { companyId: string }): Promise<unknown>;
+    recoveryActions(scope: { companyId: string }): Promise<unknown>;
     dependencies(scope: Scope): Promise<{ enqueueFailed: number }>;
     locks(scope: Scope): Promise<unknown>;
     watchdogs(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>;
@@ -22,20 +25,32 @@ export function registerIssueRecoveryWork(
   canRun: () => boolean,
 ) {
   function register(queue: DeliveryQueue, statuses: string[] | null, check: (scope: Scope) => Promise<unknown>) {
+    const cursors = new Map<string, string>();
+    let enabled = false;
     return registerScopedDeadlineWork<Entry, IssueRecoveryScope>(coordinator, {
       queue,
+      reconstructionIntervalMs: 30_000,
+      beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); },
       decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope,
       matches: (entry, scope) => !scope || entry.companyId === scope.companyId,
       load: async scope => {
+        const key = scope?.companyId ?? "*";
+        if (!enabled) { cursors.delete(key); return []; }
+        const cursor = cursors.get(key);
         const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and(
           scope ? eq(issues.companyId, scope.companyId) : undefined,
+          cursor ? gt(issues.id, cursor) : undefined,
           statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)),
           statuses ? isNull(issues.hiddenAt) : undefined,
-        ));
-        return rows.map(row => ({ ...row, dueAt: Date.now() }));
+          statuses ? isNull(issues.assigneeUserId) : undefined,
+          statuses ? or(isNotNull(issues.assigneeAgentId), eq(issues.status, "in_review")) : undefined,
+        )).orderBy(asc(issues.id)).limit(PAGE_SIZE);
+        const complete = rows.length < PAGE_SIZE;
+        if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
+        return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete };
       },
       check: async entry => {
-        if (await owner.schedulingSuppressed()) return null;
+        if (!enabled) return null;
         let next: number | null = null;
         await check({ companyId: entry.companyId, issueId: entry.id,
           onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } });
@@ -43,7 +58,45 @@ export function registerIssueRecoveryWork(
       },
     }, canRun);
   }
+  function registerCompany(queue: DeliveryQueue, kind: "blockers" | "actions", check: (scope: { companyId: string }) => Promise<unknown>) {
+    const cursors = new Map<string, string>();
+    let enabled = false;
+    return registerScopedDeadlineWork<Entry, IssueRecoveryScope>(coordinator, {
+      queue,
+      reconstructionIntervalMs: 30_000,
+      beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); },
+      decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope,
+      matches: (entry, scope) => !scope || entry.companyId === scope.companyId,
+      load: async scope => {
+        const key = scope?.companyId ?? "*";
+        if (!enabled) { cursors.delete(key); return []; }
+        const cursor = cursors.get(key);
+        const rows = kind === "actions"
+          ? await db.selectDistinct({ id: issueRecoveryActions.companyId, companyId: issueRecoveryActions.companyId })
+            .from(issueRecoveryActions).where(and(inArray(issueRecoveryActions.status, ["active", "escalated"]),
+              scope ? eq(issueRecoveryActions.companyId, scope.companyId) : undefined,
+              cursor ? gt(issueRecoveryActions.companyId, cursor) : undefined))
+            .orderBy(asc(issueRecoveryActions.companyId)).limit(PAGE_SIZE)
+          : await db.selectDistinct({ id: issues.companyId, companyId: issues.companyId }).from(issues).where(and(
+              inArray(issues.status, ["todo", "blocked"]), isNull(issues.assigneeAgentId), isNull(issues.assigneeUserId),
+              isNotNull(issues.createdByAgentId),
+              scope ? eq(issues.companyId, scope.companyId) : undefined,
+              cursor ? gt(issues.companyId, cursor) : undefined,
+              sql`exists (select 1 from ${issueRelations} r inner join ${issues} dependent
+                on dependent.id = r.related_issue_id and dependent.company_id = r.company_id
+                where r.company_id = ${issues.companyId} and r.issue_id = ${issues.id}
+                and r.type = 'blocks' and dependent.status not in ('done', 'cancelled'))`,
+            )).orderBy(asc(issues.companyId)).limit(PAGE_SIZE);
+        const complete = rows.length < PAGE_SIZE;
+        if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
+        return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete };
+      },
+      check: async entry => { if (enabled) await check({ companyId: entry.companyId }); return null; },
+    }, canRun);
+  }
   const assignments = register(DELIVERY_QUEUES.strandedAssignments, ["todo", "in_progress", "in_review"], owner.stranded);
+  const orphanBlockers = registerCompany(DELIVERY_QUEUES.orphanBlockers, "blockers", owner.orphanBlockers);
+  const recoveryActions = registerCompany(DELIVERY_QUEUES.activeRecoveryActions, "actions", owner.recoveryActions);
   const dependencies = register(DELIVERY_QUEUES.resolvedDependencies, ["blocked"], async scope => {
     const result = await owner.dependencies(scope);
     if (result.enqueueFailed > 0) throw new Error("Resolved dependency wake remains undelivered");
@@ -52,7 +105,8 @@ export function registerIssueRecoveryWork(
   const watchdogs = registerTaskWatchdogWork(db, coordinator, {
     reconcile: owner.watchdogs, enabled: async () => !(await owner.schedulingSuppressed()),
   }, canRun);
-  return { ready: Promise.all([assignments.ready, dependencies.ready, locks.ready, watchdogs.ready]),
-    wake() { assignments.invalidate(); dependencies.invalidate(); locks.invalidate(); watchdogs.invalidate(); },
-    async stop() { await Promise.all([assignments.stop(), dependencies.stop(), locks.stop(), watchdogs.stop()]); } };
+  const workers = [assignments, orphanBlockers, recoveryActions, dependencies, locks, watchdogs];
+  return { ready: Promise.all(workers.map(worker => worker.ready)),
+    wake() { for (const worker of workers) worker.invalidate(); },
+    async stop() { await Promise.all(workers.map(worker => worker.stop())); } };
 }

server/src/services/scoped-deadline-work.ts

@@ -12,6 +12,8 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
     queue: DeliveryQueue;
     batchSize?: number;
     minimumPassIntervalMs?: number;
+    reconstructionIntervalMs?: number;
+    beforePass?(): Promise<void>;
     decodeScope(hint: string): Scope | undefined;
     matches(entry: Entry, scope?: Scope): boolean;
     load(scope?: Scope, signal?: AbortSignal): Promise<Entry[] | { rows: Entry[] | null; complete: boolean }>;
@@ -25,6 +27,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
   const entries = new Map<string, Entry>();
   const failures = new Map<string, { count: number; retryAt: number }>();
   const scopeFailures = new Map<string, { count: number; retryAt: number }>();
+  const reconstructAfter = new Map<string, number>();
   let earliest = 0;
   let lastCompany: string | null = null;
   function failed(map: typeof failures, id: string) {
@@ -34,13 +37,14 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
   function dirtyScopes() {
     return [...dirty].filter(hint => hint !== "*" || !scopeFailures.has(hint))
       .filter(hint => !scans.has(hint) || scopeFailures.has(hint))
+      .filter(hint => (reconstructAfter.get(hint) ?? 0) <= Date.now() || scopeFailures.has(hint))
       .map(hint => { try { return options.decodeScope(hint); } catch { return undefined; } });
   }
   function next() {
     if (!canRun()) return dirty.size > 0 || entries.size > 0 ? Date.now() + 30_000 : null;
     let nextAt: number | null = null;
     const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); };
-    for (const hint of dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0));
+    for (const hint of dirty) take(Math.max(earliest, reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0));
     const unavailable = dirtyScopes();
     for (const entry of entries.values()) if (!unavailable.some(scope => options.matches(entry, scope)))
       take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0));
@@ -54,9 +58,18 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
     } },
     run: async signal => {
       if (!canRun() || Date.now() < earliest) return;
+      if ((options.reconstructionIntervalMs ?? 0) > 0) for (const [hint, until] of reconstructAfter) {
+        if (until <= Date.now()) reconstructAfter.delete(hint);
+      }
+      const ready = [...dirty].filter(hint => Math.max(reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
+      if (ready.length === 0 && ![...entries.values()].some(entry => Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now())) return;
       earliest = Date.now() + (options.minimumPassIntervalMs ?? 1000);
+      try { await options.beforePass?.(); } catch (error) {
+        for (const hint of ready) failed(scopeFailures, hint);
+        for (const entry of entries.values()) if (entry.dueAt <= Date.now()) failed(failures, entry.id);
+        throw error;
+      }
       const errors: unknown[] = [];
-      const ready = [...dirty].filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
       const requested = ready.includes("*")
         ? ["*", ...ready.filter(hint => hint !== "*")].slice(0, 25)
         : ready.slice(0, 25);
@@ -81,6 +94,9 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
             entries.delete(entry.id); failures.delete(entry.id);
           }
           scans.delete(hint);
+          if ((options.reconstructionIntervalMs ?? 0) > 0) {
+            reconstructAfter.set(hint, Date.now() + options.reconstructionIntervalMs!);
+          }
           if ((versions.get(hint) ?? 0) !== scan.version) dirty.add(hint); else versions.delete(hint);
           if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint);
         } catch (error) {

server/src/services/task-watchdog-work.ts

@@ -1,4 +1,4 @@
-import { and, eq } from "drizzle-orm";
+import { and, asc, eq, gt } from "drizzle-orm";
 import { issueWatchdogs, type Db } from "@paperclipai/db";
 import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
@@ -16,16 +16,29 @@ export function registerTaskWatchdogWork(
   },
   canRun: () => boolean,
 ) {
+  const cursors = new Map<string, string>();
+  let enabled = false;
   return registerScopedDeadlineWork<Entry, IssueRecoveryScope>(coordinator, {
     queue: DELIVERY_QUEUES.taskWatchdogs,
+    reconstructionIntervalMs: 30_000,
+    beforePass: async () => { enabled = await owner.enabled(); },
     decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope,
     matches: (entry, scope) => !scope || entry.companyId === scope.companyId,
-    load: async scope => (await db.select({ id: issueWatchdogs.id, companyId: issueWatchdogs.companyId })
-      .from(issueWatchdogs).where(and(eq(issueWatchdogs.status, "active"),
-        scope ? eq(issueWatchdogs.companyId, scope.companyId) : undefined)))
-      .map(row => ({ ...row, dueAt: Date.now() })),
+    load: async scope => {
+      const key = scope?.companyId ?? "*";
+      if (!enabled) { cursors.delete(key); return []; }
+      const cursor = cursors.get(key);
+      const rows = await db.select({ id: issueWatchdogs.id, companyId: issueWatchdogs.companyId })
+        .from(issueWatchdogs).where(and(eq(issueWatchdogs.status, "active"),
+          scope ? eq(issueWatchdogs.companyId, scope.companyId) : undefined,
+          cursor ? gt(issueWatchdogs.id, cursor) : undefined))
+        .orderBy(asc(issueWatchdogs.id)).limit(100);
+      const complete = rows.length < 100;
+      if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
+      return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete };
+    },
     check: async entry => {
-      if (!(await owner.enabled())) return null;
+      if (!enabled) return null;
       let next: number | null = null;
       const result = await owner.reconcile({ companyId: entry.companyId, watchdogId: entry.id,
         onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } });