Merge commit '1f1b046d72' into codex/sleep-09-aggregate

* commit '1f1b046d72':
  fix(server): bound graph queries and scope issue retention work
  fix: scope status-card deletion notifications to its company
  Restore independent recovery duties and bound committed-source work

doc/plans/2026-10-30-attention-retention-sleep.md

@@ -25,7 +25,7 @@ The feed and worker share confirmation collapse, newer-run probes, issue visibil
 
 ## Existing mutation boundaries reactivate work
 
-Committed hints identify a company, source kind, and source ID where those values are known. Retention's own writes use exact source hints; blocker retention changes invalidate the company/source because multiple blocked roots can project one terminal subject. Reconstruction pages and due batches remain bounded. Review attention and interaction invokability are computed in company batches per candidate page; a full 100-candidate review page uses at most 20 statements and 100 confirmations use at most six. Decisions, recovery source links, runs, eligible agents, issue visibility, newer-run probes and blocker graphs also load in company batches per page. A 100-source decision or recovery page uses at most three statements; 100 failed runs use at most six; 100 blocker roots use at most 30 graph/source statements plus retention-state reads for their emitted entries. These evidence maps exist only for one load. The blocker owner shares fetched node and edge facts while tracking depth, node count, expanded nodes and live-work reachability independently for each root. An unrelated graph reaching its limit cannot change a shallow root’s terminal subject or deadline. Archive checks still lock and freshly re-read sources, terminal subjects and related eligibility. Ordinary execution, decision and budget mutations pass their affected company; exact decision sources are included.
+Committed hints identify a company, source kind, source ID or changed issue where those values are known. Routine issue writers pass identity to the retention domain helper before their mutation. The helper captures old and new parent paths, reverse blocker ancestors, run-linked issues and liveness-origin references; issue-scoped source queries then select reviews, blockers, interactions, decisions, recovery actions and failed runs by their issue identity. Budget incidents and agent errors remain tied to their own source changes. Ordinary issue writes do not reconstruct the unrelated company feed. Queue keys explicitly pick supported identity fields. Retention's own writes use exact source hints; blocker retention changes invalidate the company/source because multiple blocked roots can project one terminal subject. Reconstruction pages and due batches remain bounded. Review attention and interaction invokability are computed in company batches per candidate page; a full 100-candidate review page uses at most 20 statements and 100 confirmations use at most six. Decisions, recovery source links, runs, eligible agents, issue visibility, newer-run probes and blocker graphs also load in company batches per page. A 100-source decision or recovery page uses at most three statements; 100 failed runs use at most six; 100 blocker roots use at most 30 graph/source statements plus retention-state reads for their emitted entries. These evidence maps exist only for one load. The blocker owner shares fetched node and edge facts while tracking depth, node count, expanded nodes and live-work reachability independently for each root. An unrelated graph reaching its limit cannot change a shallow root’s terminal subject or deadline. All later graph queries, including recovery source IDs, agent IDs and workspace-finalize inputs, use bounded chunks. A SQL-generated 70,000-node graph proves PostgreSQL parameter limits are respected at the driver boundary. Archive checks still lock and freshly re-read sources, terminal subjects and related eligibility. Ordinary execution, decision and budget mutations pass their affected company; exact decision sources are included.
 
 Producer coverage includes decision creation and changes; interaction creation, resolution, expiry and supersession; issue creation, updates, blocker relations, checkout, adoption and release; agent hire, configuration, lifecycle, error clearing and deletion; budget incidents and company pause/resume; exhausted-run events and retry-run creation; native status, review, finalization and reconstruction; external conversation transitions; and imported blocker relations. Existing service transactions retain outer commit ownership. Savepoint release does not publish work before outer commit.
 

doc/plans/2026-10-30-issue-recovery-sleep.md

@@ -1,14 +1,18 @@
 # Issue recovery without periodic scans
 
-Four existing owners reconstruct their current obligations once at startup. Committing source writes invalidate only their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database.
+The issue recovery owners reconstruct their current obligations once at startup. Committing source writes invalidate only their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database.
 
 | Owner | Durable truth and producer boundary | Deadline or event | Exit |
 | --- | --- | --- | --- |
 | Stranded assignment | Visible assigned todo/in-progress/review issues; issue create/update/checkout/release; run lifecycle and execution control; agent/company/budget/configuration; interaction, approval and subtree hold changes | A committed eligibility change. A transient continuation uses the latest finish plus the existing exponential delay. | An execution path, durable wait, escalation, or current ineligibility. |
+| Orphan blocker assignment | Agent-created unassigned todo/blocked blockers with a nonterminal dependent; issue, relation, agent and company changes | Once per changed company, independent of assignment candidates. | Assigned, no longer blocking, or creator cannot execute until its next eligibility change. |
+| Active recovery actions | Active/escalated action records, including actions on completed sources; action, issue, run and wait changes | Once per changed company, independent of assignment candidates. | Resolved, restored path, explicit hold or a separately owned retry. |
 | Resolved dependency | Blocked assigned issues and company-scoped blocker relations; blocker status or relation changes through issue service and import; run/control and pause gate changes | The committing change. A failed enqueue retains that issue with error backoff. | Existing idempotent wake, live path, unresolved dependency or deliberate self-block. |
 | Task watchdog | Active issue watchdogs and their current source subtree; normal watchdog upsert/disable; issue/comment/document/work-product and wait changes | Exact source creation plus the existing 15-second first-run grace. An incomplete scan retains the watchdog with error backoff. | Live path, open/reviewed reusable review, disabled configuration, or ineligible source. |
 | Stale issue lock | Existing checkout/execution run references; issue and run lifecycle changes | A committing ownership change, or the fresh owning controller's saved lease expiry. | No stale reference, a valid owner, or the existing guarded terminalization and lock clear. |
 
+Reconstruction uses keyset pages of at most100rows. Repeated company hints coalesce reconstruction for30seconds. Already-known entry deadlines preempt that spacing without an early probe on each comment. Scheduling suppression is read once per pass. A failed read keeps scoped backoff debt. The two subsidiary company repairs do not run inside the per-issue loop. Legacy direct aggregate API calls retain both repairs.
+
 The owner methods accept company and issue/watchdog identity. A narrowed lock check still reads other same-company references before using terminal-issue authority. A stale terminal alias cannot end an active issue's shared run. The controller lease owner supplies its fresh expiry; the existing write predicate revalidates the observed boot ID and database-clock expiry.
 
 Producer coverage uses the existing run-dispatch, execution-control, issue-review-state, native status reassessment and external conversation settlement boundaries. Direct gaps have explicit signals inside their owning transaction: document writes/import, attachment removal, blocker import, recovery-action adoption/mutation, conversation pause release, Telegram failure comment replacement, native artifact title/metadata updates and company deletion. Metadata-only comment classification that does not change timestamp, visibility or execution eligibility needs no additional watchdog activation. Approval links and decisions use their existing issue-review/continuation effects. Work-product create/update/delete/import uses its existing native status boundary.
@@ -16,3 +20,5 @@ Producer coverage uses the existing run-dispatch, execution-control, issue-revie
 The new owners replace only these four startup and periodic calls. Deferred comments, silent active runs and residual orphan/resource recovery retain their existing backstop until the next slice. Every previously migrated native-owner skip remains unchanged.
 
 A controlled process test prints an execution proof but cannot apply a task disposition. The test operator completes that task through issue service. If the existing continuation policy already saved a disposition-repair retry, the test preserves it until its stored deadline. Only after all provider work has joined does it advance the clock; dispatch then revalidates the completed task and cancels the retry. The final quiet proof requires no registered deadlines, no tracked in-flight work and no owner SQL over twenty former polling minutes. It does not claim a saved future retry is already drained.
+
+Active action repair has no separate grace or lease countdown. Its delayed disposition repair first writes a scheduled-retry run, then mirrors the date into action wakePolicy.retryAt and timeoutAt. The dispatch owner owns that saved run deadline. Other skipped gates wait for their existing execution, wait, pause, budget, or finalizer producer; the company action worker does not add a polling fallback.

doc/plans/2026-10-30-status-cards-sleep.md

@@ -135,3 +135,9 @@ scheduled work after settlement. Real issue/comment producer acceptance measures
 due/next queries and actual searches under 100 writes, preserves both companies'
 hints, dispatches another company's saved deadline inside the batching window,
 and checks SQL quietness after authorized results.
+
+Card deletion resolves the card's owning company and emits only that company's
+deadline-reconstruction hint. Every card mutation requires a company ID; only
+instance-level settings retain wildcard ingress. Deleting an already-missing
+card emits no hint. The real two-company regression proves deletion does not
+re-run unchanged reactive searches or wake either summarizer, then stays quiet.

server/src/__tests__/attention-retention-work.test.ts

@@ -1,7 +1,7 @@
 import { randomUUID } from "node:crypto";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
-import { authUsers, companyMemberships, agents, approvals, budgetIncidents, budgetPolicies, companies, createDb, applyPendingMigrations, decisionArchiveNotificationOutbox,
+import { subscribeDatabaseWork, authUsers, companyMemberships, agents, approvals, budgetIncidents, budgetPolicies, companies, createDb, applyPendingMigrations, decisionArchiveNotificationOutbox,
   decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions,
   issueRelations, issueThreadInteractions, issues, type Db } from "@paperclipai/db";
 import { attentionRetentionService, registerAttentionRetentionWork } from "../services/attention-retention.js";
@@ -13,6 +13,7 @@ import { companyService } from "../services/companies.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../services/decision-retention.js";
 import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
+import { buildIssueGraphLivenessIncidentKey, RECOVERY_ORIGIN_KINDS } from "../services/recovery/origins.js";
 import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
 import { createAgentLifecycle } from "../services/agent-lifecycle.js";
 import { BLOCKER_ATTENTION_MAX_DEPTH, BLOCKER_ATTENTION_MAX_NODES, issueService } from "../services/issues.js";
@@ -32,12 +33,13 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
   let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
   const errors: unknown[] = [];
   const statements: string[] = [];
+  const parameterCounts: number[] = [];
   beforeAll(async () => {
     if (externalUrl) await applyPendingMigrations(externalUrl);
     else temporary = await startEmbeddedPostgresTestDatabase("attention-retention-");
     db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
     db.$client.options.onnotice = () => {};
-    db.$client.options.debug = (_connection, query) => statements.push(query);
+    db.$client.options.debug = (_connection, query, parameters) => { statements.push(query); parameterCounts.push(parameters.length); };
   }, 30_000);
   beforeEach(async () => {
     companyId = randomUUID(); agentId = randomUUID(); draining = false; errors.length = 0;
@@ -311,6 +313,111 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
     }
   });
 
+  it("bounds every graph query when independent root trees contain over 65535 unfinished nodes", async () => {
+    await db.execute(sql`insert into issues (id, company_id, title, status, updated_at)
+      select gen_random_uuid(), ${companyId}::uuid, 'parameter-root-' || n, 'blocked', ${old().toISOString()}::timestamptz
+      from generate_series(1, 70) n`);
+    await db.execute(sql`insert into issues (id, company_id, parent_id, title, status, assignee_agent_id, updated_at)
+      select gen_random_uuid(), ${companyId}::uuid, root.id, 'parameter-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz
+      from issues root cross join generate_series(1, 999) n
+      where root.company_id = ${companyId}::uuid and root.status = 'blocked'`);
+    const roots = await db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.status, "blocked")));
+    parameterCounts.length = 0;
+    const result = await issueService(db).listBlockerAttention(companyId, roots).catch(error => {
+      throw new Error(`Large graph failed: ${String(error.cause?.message ?? error.message).slice(0, 300)}`);
+    });
+    expect(result.size).toBe(70);
+    expect([...result.values()].every(item => item.state === "needs_attention" && item.terminalBlockerIssueId)).toBe(true);
+    expect(Math.max(...parameterCounts)).toBeLessThanOrEqual(1100);
+  }, 120_000);
+
+  it("canonicalizes wide circular notification variables to the supported identity fields", async () => {
+    const sourceId = randomUUID(), issueId = randomUUID(), hints: string[] = [];
+    const wide = { companyId, sourceKind: "review" as const, sourceId, issueId, ignored: {} as unknown };
+    wide.ignored = wide;
+    const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, values) => {
+      if (event === "settled") hints.push(...values ?? []);
+    });
+    try { await db.transaction(tx => notifyAttentionRetentionWork(tx, wide)); } finally { unsubscribe(); }
+    expect(hints).toEqual([JSON.stringify({ companyId, sourceKind: "review", sourceId, issueId })]);
+  });
+
+  it("keeps held, burst and sustained ordinary issue writes scoped while an unrelated deadline fires", async () => {
+    const row = await issueService(db).create(companyId, { title: "Active review", status: "in_review", assigneeUserId: "board", createdByAgentId: agentId });
+    const urgent = await insertIssue("in_review", { assigneeUserId: "board", updatedAt: new Date(Date.now() - delay + 10_000) });
+    const otherCompany = randomUUID();
+    await db.insert(companies).values({ id: otherCompany, name: "Unrelated", issuePrefix: "BURST" });
+    const foreign = await db.insert(issues).values(Array.from({ length: 125 }, () => ({ companyId: otherCompany,
+      title: randomUUID(), status: "in_review", assigneeUserId: "board", updatedAt: old() }))).returning();
+    await db.insert(decisionRetention).values(foreign.map(issue => ({ companyId: otherCompany, sourceKind: "review", sourceId: issue.id, sourceActivityAt: issue.updatedAt, keep: true })));
+    const service = attentionRetentionService(db), original = service.load, load = vi.spyOn(service, "load");
+    await start(service); await settle(); load.mockClear();
+    let release!: () => void, entered = false;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    load.mockImplementationOnce(async (...args) => { const page = await original(...args); entered = true; await gate; return page; });
+    statements.length = 0;
+    await issueService(db).update(row.id, { title: "Held mutation" });
+    await vi.waitFor(() => expect(entered).toBe(true), { timeout: 15_000 });
+    await Promise.all(Array.from({ length: 5 }, (_, index) => issueService(db).update(row.id, { title: `Burst ${index}` })));
+    release(); await settle();
+    for (let index = 0; index < 8; index++) {
+      await issueService(db).update(row.id, { title: `Steady ${index}` }); await vi.advanceTimersByTimeAsync(1000); await settle();
+    }
+    await archived("review", urgent.id);
+    const routine = load.mock.calls.map(([scope]) => scope).filter(scope => !scope?.sourceId);
+    expect(routine.length).toBeGreaterThan(0);
+    expect(routine.every(scope => scope?.companyId === companyId && scope.issueId === row.id)).toBe(true);
+    const scans = statements.filter(query => query.includes("with source_candidates"));
+    expect(scans.length).toBeGreaterThan(0);
+    expect(scans.every(query => !query.includes("budget_incidents") && !query.includes("agent_error_alert"))).toBe(true);
+    expect(await state("review", row.id)).toBeNull();
+  });
+
+  it("wakes all linked issue sources when an ordinary visibility mutation reveals their origin", async () => {
+    const origin = await insertIssue("in_review", { hiddenAt: old(), assigneeUserId: "board", responsibleUserId: "board" });
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "failed", contextSnapshot: { issueId: origin.id }, createdAt: old(), finishedAt: old() }).returning();
+    await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: run.id, seq: 1, eventType: "lifecycle", message: "Bounded retry exhausted: visibility" });
+    const [decision] = await db.insert(decisions).values({ companyId, originAgentId: agentId, originIssueId: origin.id, originRunId: run.id,
+      title: "Decision", body: "Choose", options: [], signedSpec: "fixture", targetSnapshots: {}, expiresAt: new Date(Date.now() + 86_400_000), updatedAt: old() }).returning();
+    const [recovery] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: origin.id, kind: "execution", ownerType: "board", cause: "fixture", fingerprint: "fixture", nextAction: "Choose", updatedAt: old() }).returning();
+    const [interaction] = await db.insert(issueThreadInteractions).values({ companyId, issueId: origin.id, kind: "ask_user_questions", payload: { questions: [] }, updatedAt: old() }).returning();
+    await start(); await quiet();
+    await issueService(db).update(origin.id, { hiddenAt: null });
+    for (const [kind, id] of [["decision", decision.id], ["recovery_action", recovery.id], ["failed_run", run.id], ["issue_thread_interaction", interaction.id]]) await archived(kind, id);
+    await settle(); expect(await state("review", origin.id)).toBeNull(); expect(coordinator!.nextWakeAt()).not.toBeNull();
+  });
+
+  it("invalidates old and new parents, terminal visibility and repointed blocker relations", async () => {
+    const oldParent = await insertIssue("blocked", { updatedAt: new Date(), responsibleUserId: "board" });
+    const newParent = await insertIssue("blocked", { updatedAt: new Date(), responsibleUserId: "board" });
+    const terminal = await insertIssue("backlog", { parentId: oldParent.id, hiddenAt: old(), assigneeAgentId: agentId, responsibleUserId: "board" });
+    const service = attentionRetentionService(db), load = vi.spyOn(service, "load"); await start(service); await settle(); load.mockClear();
+    await issueService(db).update(terminal.id, { parentId: newParent.id }); await settle();
+    const ids = load.mock.calls.map(([scope]) => scope?.issueId);
+    expect(ids).toEqual(expect.arrayContaining([oldParent.id, newParent.id, terminal.id]));
+    load.mockClear();
+    await issueService(db).update(terminal.id, { hiddenAt: null }); await settle();
+    expect(await state("blocker_attention", terminal.id)).toBeNull();
+    const visible = await service.load({ companyId, issueId: newParent.id });
+    expect(visible.rows!.some(entry => entry.sourceId === terminal.id && entry.dueAt > Date.now())).toBe(true);
+    expect(load.mock.calls.some(([scope]) => scope?.issueId === newParent.id)).toBe(true);
+    const next = await insertIssue("backlog", { assigneeAgentId: agentId });
+    await issueService(db).update(oldParent.id, { blockedByIssueIds: [next.id] }); await archived("blocker_attention", next.id);
+  });
+
+  it("cancels and reactivates a blocker deadline through ordinary liveness incident creation and hiding", async () => {
+    const root = await insertIssue("blocked");
+    const terminal = await insertIssue("backlog", { assigneeAgentId: agentId, updatedAt: new Date(Date.now() - delay + 10_000) });
+    await db.insert(issueRelations).values({ companyId, issueId: terminal.id, relatedIssueId: root.id, type: "blocks" });
+    await start(); await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull();
+    const incident = await issueService(db).create(companyId, { title: "Resolve liveness", status: "todo", assigneeUserId: "board",
+      createdByUserId: "board", originKind: RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation,
+      originId: buildIssueGraphLivenessIncidentKey({ companyId, issueId: root.id, state: "needs_attention", blockerIssueId: terminal.id }) });
+    await quiet(); expect(await state("blocker_attention", terminal.id)).toBeNull();
+    await issueService(db).update(incident.id, { hiddenAt: new Date() });
+    await archived("blocker_attention", terminal.id); await quiet();
+  });
+
   it("scopes ordinary execution-control writes to their company", async () => {
     const otherCompany = randomUUID();
     await db.insert(companies).values({ id: otherCompany, name: "Other", issuePrefix: "OTHER" });

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/__tests__/status-card-evaluation-work.test.ts

@@ -144,6 +144,38 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     expect(errors).toEqual([]);
   });
 
+  it("keeps card deletion scoped to its company and a missing deletion quiet", async () => {
+    const removed = await card();
+    await card();
+    const ownCompany = companyId, ownAgent = agentId;
+    companyId = randomUUID(); agentId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Other", issuePrefix: `O${companyId.slice(0, 6)}` });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Other summarizer", role: "general", status: "idle", adapterType: "process", adapterConfig: {} });
+    const other = await card();
+    companyId = ownCompany; agentId = ownAgent;
+    const factory = companySearch.companySearchService;
+    const searchedCompanies: string[] = [];
+    vi.spyOn(companySearch, "companySearchService").mockImplementation(owner => {
+      const search = factory(owner), actual = search.search;
+      search.search = async (...args) => { searchedCompanies.push(args[0]); return actual(...args); };
+      return search;
+    });
+    await start(); await quiet(); searchedCompanies.length = 0;
+    const cards = statusCardService(db);
+    expect(await cards.remove(removed.id)).toMatchObject({ id: removed.id, companyId: ownCompany });
+    await settled();
+    // Deleting a card only reconstructs deadlines; it is not an issue change
+    // that warrants re-running any unchanged reactive card's saved search.
+    expect(searchedCompanies).toEqual([]);
+    expect(await cards.getById(other.id)).toMatchObject({ nextEvalAt: null, generatingIssueId: null, pendingChangeCount: 0 });
+    expect(wakeup).not.toHaveBeenCalled();
+    searchedCompanies.length = 0;
+    expect(await cards.remove(removed.id)).toBeNull();
+    await quiet();
+    expect(searchedCompanies).toEqual([]);
+    expect(errors).toEqual([]);
+  });
+
   it("coalesces one hundred committed source writes without delaying another company's saved deadline", async () => {
     const reactive = await card();
     const originalCompany = companyId, originalAgent = agentId;

server/src/services/attention-retention.ts

@@ -18,8 +18,8 @@ const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
 const kinds = ["decision", "issue_thread_interaction", "recovery_action", "blocker_attention", "review",
   "failed_run", "budget_alert", "agent_error_alert"] as const;
 type Kind = typeof kinds[number];
-type Scope = { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string };
-type Candidate = { sourceKind: Kind; sourceId: string; companyId: string; activityAt: string; deadlineAt: number };
+type Scope = { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string; issueId?: string };
+type Candidate = { sourceKind: Kind; sourceId: string; companyId: string; activityAt: string; deadlineAt: number; issueId: string | null };
 type Entry = Candidate & { id: string; dueAt: number; rootId: string; variant?: "owner" | "terminal" };
 type Cursor = { sourceKind: string; sourceId: string };
 
@@ -42,37 +42,39 @@ const sourceTables: Record<Kind, SQL> = {
 };
 
 const sources: Record<Kind, SQL> = {
-  decision: sql`select 'decision' as source_kind, d.id::text as source_id, d.company_id, d.updated_at as activity_at
+  decision: sql`select 'decision' as source_kind, d.id::text as source_id, d.company_id, d.updated_at as activity_at, d.origin_issue_id::text as issue_id
     from decisions d where d.status = 'open'`,
-  issue_thread_interaction: sql`select 'issue_thread_interaction', i.id::text, i.company_id, i.updated_at
+  issue_thread_interaction: sql`select 'issue_thread_interaction', i.id::text, i.company_id, i.updated_at, i.issue_id::text
     from issue_thread_interactions i where i.status = 'pending' and i.addressee_user_id is null`,
-  recovery_action: sql`select 'recovery_action', r.id::text, r.company_id, r.updated_at
+  recovery_action: sql`select 'recovery_action', r.id::text, r.company_id, r.updated_at, r.source_issue_id::text
     from issue_recovery_actions r where r.status in ('active', 'escalated') and r.owner_type in (${sql.join(HUMAN_RECOVERY_OWNER_TYPES.map(owner => sql`${owner}`), sql`, `)})`,
-  blocker_attention: sql`select 'blocker_attention', i.id::text, i.company_id, i.updated_at
+  blocker_attention: sql`select 'blocker_attention', i.id::text, i.company_id, i.updated_at, i.id::text
     from issues i where i.status = 'blocked' and i.hidden_at is null and i.harness_kind is null`,
-  review: sql`select 'review', i.id::text, i.company_id, i.updated_at
+  review: sql`select 'review', i.id::text, i.company_id, i.updated_at, i.id::text
     from issues i where i.status = 'in_review' and i.hidden_at is null and i.harness_kind is null`,
-  failed_run: sql`select 'failed_run', r.id::text, r.company_id, coalesce(r.finished_at, r.updated_at, r.created_at)
+  failed_run: sql`select 'failed_run', r.id::text, r.company_id, coalesce(r.finished_at, r.updated_at, r.created_at), coalesce(r.context_snapshot->>'issueId', r.context_snapshot->>'taskId')
     from heartbeat_runs r where r.status in ('failed', 'timed_out') and exists (
       select 1 from heartbeat_run_events e where e.run_id = r.id and e.company_id = r.company_id
         and e.event_type = 'lifecycle' and e.message like ${`${RETRY_EXHAUSTION_MESSAGE_PREFIX}%`})`,
-  budget_alert: sql`select 'budget_alert', b.id::text, b.company_id, b.updated_at from budget_incidents b
+  budget_alert: sql`select 'budget_alert', b.id::text, b.company_id, b.updated_at, null::text from budget_incidents b
     where b.status = 'open' and (b.threshold_type = 'hard' or (b.amount_limit > 0
       and round(b.amount_observed::numeric / b.amount_limit * 10000) / 100 >= ${ATTENTION_BUDGET_WARNING_PERCENT}))`,
-  agent_error_alert: sql`select 'agent_error_alert', a.id::text, a.company_id, a.updated_at from agents a where a.status = ${ATTENTION_AGENT_ERROR_STATUS}`,
+  agent_error_alert: sql`select 'agent_error_alert', a.id::text, a.company_id, a.updated_at, null::text from agents a where a.status = ${ATTENTION_AGENT_ERROR_STATUS}`,
 };
 
 export function attentionRetentionService(db: Db) {
   const cursors = new Map<string, Cursor>();
   async function candidates(database: Db, scope: Scope = {}, cursor?: Cursor, sourceId?: string): Promise<Candidate[]> {
-    const selected = kinds.filter(kind => !scope.sourceKind || scope.sourceKind === kind);
+    const selected = kinds.filter(kind => (!scope.sourceKind || scope.sourceKind === kind)
+      && (!scope.issueId || !["budget_alert", "agent_error_alert"].includes(kind)));
     if (!selected.length) return [];
-    const rows = await database.execute(sql`with source_candidates(source_kind, source_id, company_id, activity_at) as (${sql.join(selected.map(kind => sources[kind]), sql` union all `)})
+    const rows = await database.execute(sql`with source_candidates(source_kind, source_id, company_id, activity_at, issue_id) as (${sql.join(selected.map(kind => sources[kind]), sql` union all `)})
       select s.source_kind as "sourceKind", s.source_id as "sourceId", s.company_id as "companyId",
-        s.activity_at::text as "activityAt", ceil(extract(epoch from s.activity_at) * 1000 + ${archiveDelay}) as "deadlineAt"
+        s.issue_id as "issueId", s.activity_at::text as "activityAt", ceil(extract(epoch from s.activity_at) * 1000 + ${archiveDelay}) as "deadlineAt"
       from source_candidates s inner join companies c on c.id = s.company_id
       left join decision_retention r on r.company_id = s.company_id and r.source_kind = s.source_kind and r.source_id = s.source_id
       where c.status = 'active'
+        and (${scope.issueId ?? null}::text is null or s.issue_id = ${scope.issueId ?? null})
         and (${scope.companyId ?? null}::uuid is null or s.company_id = ${scope.companyId ?? null}::uuid)
         and (${sourceId ?? scope.sourceId ?? null}::text is null or s.source_id = ${sourceId ?? scope.sourceId ?? null})
         and (s.source_kind = 'blocker_attention' or ((r.id is null or r.keep = false) and r.archived_at is null))
@@ -302,7 +304,8 @@ export function registerAttentionRetentionWork(coordinator: ReturnType<typeof cr
     decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as Scope,
     matches: (entry, scope) => (!scope?.companyId || scope.companyId === entry.companyId)
       && (!scope?.sourceKind || scope.sourceKind === entry.sourceKind)
-      && (!scope?.sourceId || scope.sourceId === entry.sourceId),
+      && (!scope?.sourceId || scope.sourceId === entry.sourceId)
+      && (!scope?.issueId || scope.issueId === entry.issueId),
     load: service.load, check: service.check,
   }, canRun);
 }

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/issue-review-state-notifications.ts

@@ -1,6 +1,5 @@
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
-import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
@@ -10,7 +9,6 @@ import { notifyNativeStatusReassessment } from "./native-status-reassessment-not
 export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) {
   await notifyRunDispatch(tx, scope);
   await notifyStatusCardIssueChange(tx, scope.companyId);
-  await notifyAttentionRetentionWork(tx, { companyId: scope.companyId });
   await notifyRoutineExecutionStatus(tx, scope.issueId, scope.companyId);
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
   await notifyExternalConversationSettlement(tx, scope);

server/src/services/issues.ts

@@ -1,7 +1,7 @@
 import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
-import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
+import { notifyIssueAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
 import { notifyMergeConfirmationWork } from "./merged-pr-confirmation-notifications.js";
 import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
@@ -2545,33 +2545,37 @@ async function listPendingFinalizeBlockerIssueIds(
     ),
   );
 
-  if (lockSources) {
+  const rows: Array<{ issueId: string | null; executionWorkspaceId: string | null;
+    phase: string; status: string; startedAt: Date }> = [];
+  for (const chunk of chunkList(executionWorkspaceIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
+    if (lockSources) {
 
 
-    await dbOrTx.select({ id: executionWorkspaces.id }).from(executionWorkspaces)
-      .where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, executionWorkspaceIds)))
-      .for("update", { noWait: true });
-  }
-  const query = dbOrTx
-    .select({
-      issueId: workspaceOperations.issueId,
-      executionWorkspaceId: workspaceOperations.executionWorkspaceId,
-      phase: workspaceOperations.phase,
-      status: workspaceOperations.status,
-      startedAt: workspaceOperations.startedAt,
-    })
-    .from(workspaceOperations)
-    .where(
-      and(
-        eq(workspaceOperations.companyId, companyId),
-        inArray(
-          workspaceOperations.executionWorkspaceId,
-          executionWorkspaceIds,
+      await dbOrTx.select({ id: executionWorkspaces.id }).from(executionWorkspaces)
+        .where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, chunk)))
+        .for("update", { noWait: true });
+    }
+    const query = dbOrTx
+      .select({
+        issueId: workspaceOperations.issueId,
+        executionWorkspaceId: workspaceOperations.executionWorkspaceId,
+        phase: workspaceOperations.phase,
+        status: workspaceOperations.status,
+        startedAt: workspaceOperations.startedAt,
+      })
+      .from(workspaceOperations)
+      .where(
+        and(
+          eq(workspaceOperations.companyId, companyId),
+          inArray(
+            workspaceOperations.executionWorkspaceId,
+            chunk,
+          ),
         ),
-      ),
-    );
+      );
 
-  const rows = await (lockSources ? query.for("share", { noWait: true }) : query);
+    rows.push(...await (lockSources ? query.for("share", { noWait: true }) : query));
+  }
 
   const latestAttributedByBlockerWorkspace = new Map<
     string,
@@ -4126,23 +4130,26 @@ async function listIssueBlockerAttentionMap(
       ownerType: string;
       ownerAgentId: string | null;
       ownerUserId: string | null;
-    }> = await dbOrTx
-      .select({
-        id: issueRecoveryActions.id,
-        sourceIssueId: issueRecoveryActions.sourceIssueId,
-        status: issueRecoveryActions.status,
-        ownerType: issueRecoveryActions.ownerType,
-        ownerAgentId: issueRecoveryActions.ownerAgentId,
-        ownerUserId: issueRecoveryActions.ownerUserId,
-      })
-      .from(issueRecoveryActions)
-      .where(
-        and(
-          eq(issueRecoveryActions.companyId, companyId),
-          inArray(issueRecoveryActions.status, ["active", "escalated"]),
-          inArray(issueRecoveryActions.sourceIssueId, explicitWaitCandidateIds),
-        ),
-      );
+    }> = [];
+    for (const chunk of chunkList(explicitWaitCandidateIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
+      recoveryActionRows.push(...await dbOrTx
+        .select({
+          id: issueRecoveryActions.id,
+          sourceIssueId: issueRecoveryActions.sourceIssueId,
+          status: issueRecoveryActions.status,
+          ownerType: issueRecoveryActions.ownerType,
+          ownerAgentId: issueRecoveryActions.ownerAgentId,
+          ownerUserId: issueRecoveryActions.ownerUserId,
+        })
+        .from(issueRecoveryActions)
+        .where(
+          and(
+            eq(issueRecoveryActions.companyId, companyId),
+            inArray(issueRecoveryActions.status, ["active", "escalated"]),
+            inArray(issueRecoveryActions.sourceIssueId, chunk),
+          ),
+        ));
+    }
     const recoveryActionIds = recoveryActionRows.map((row) => row.id);
     const liveRecoveryActionIds = new Set<string>();
     for (const chunk of chunkList(
@@ -4205,22 +4212,11 @@ async function listIssueBlockerAttentionMap(
     }
   }
 
-  const agentRows: IssueBlockerAttentionAgentRow[] =
-    agentIds.size > 0
-      ? await dbOrTx
-          .select({
-            id: agents.id,
-            companyId: agents.companyId,
-            status: agents.status,
-          })
-          .from(agents)
-          .where(
-            and(
-              eq(agents.companyId, companyId),
-              inArray(agents.id, [...agentIds]),
-            ),
-          )
-      : [];
+  const agentRows: IssueBlockerAttentionAgentRow[] = [];
+  for (const chunk of chunkList([...agentIds], ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
+    agentRows.push(...await dbOrTx.select({ id: agents.id, companyId: agents.companyId, status: agents.status })
+      .from(agents).where(and(eq(agents.companyId, companyId), inArray(agents.id, chunk))));
+  }
   const agentsById = new Map(agentRows.map((agent) => [agent.id, agent]));
 
   type PathClassification = {
@@ -7648,7 +7644,7 @@ export function issueService(db: Db) {
       await assertNoBlockingCycles(companyId, issueId, deduped, dbOrTx);
     }
 
-    await notifyAttentionRetentionWork(dbOrTx, { companyId });
+    await notifyIssueAttentionRetentionWork(dbOrTx, { companyId, issueId });
     await notifyIssueRecovery(dbOrTx, { companyId });
     await dbOrTx
       .delete(issueRelations)
@@ -10638,8 +10634,9 @@ export function issueService(db: Db) {
         if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyRunDispatch(tx, { companyId: values.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+        values.id ??= randomUUID();
         await notifyStatusCardIssueChange(tx, companyId);
-        await notifyAttentionRetentionWork(tx, { companyId });
+        await notifyIssueAttentionRetentionWork(tx, { companyId, issueId: values.id, parentId: values.parentId, originKind: values.originKind, originId: values.originId });
         if (values.originKind === "routine_execution") await notifyRoutineExecutionStatus(tx, values.id, companyId);
         const [issue] = await tx.insert(issues).values(values).returning();
         await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId });
@@ -11532,7 +11529,7 @@ export function issueService(db: Db) {
           || issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) {
           await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
         }
-        await notifyAttentionRetentionWork(tx, { companyId: receiptExisting.companyId });
+        await notifyIssueAttentionRetentionWork(tx, { companyId: receiptExisting.companyId, issueId: id, parentId: patch.parentId, originKind: patch.originKind, originId: patch.originId });
         if (patch.status !== undefined || patch.hiddenAt !== undefined) {
           await notifyMergeConfirmationWork(tx, { companyId: receiptExisting.companyId, issueId: id });
         }
@@ -12007,7 +12004,7 @@ export function issueService(db: Db) {
           await requestRestorePointDeletion(tx, { sourceIssueId: id });
           const [owner] = await tx.select({ companyId: issues.companyId }).from(issues).where(eq(issues.id, id));
           if (owner) {
-            await notifyAttentionRetentionWork(tx, { companyId: owner.companyId });
+            await notifyIssueAttentionRetentionWork(tx, { companyId: owner.companyId, issueId: id });
             await notifyExternalConversationSettlement(tx, { companyId: owner.companyId });
             await notifyExternalObjectRefresh(tx, owner.companyId);
             await notifyApprovalWork(tx, owner.companyId);
@@ -12145,7 +12142,7 @@ export function issueService(db: Db) {
         : isNull(issues.executionRunId);
       const updated = await db.transaction(async tx => {
         await notifyStatusCardIssueChange(tx, issueCompany.companyId);
-        await notifyAttentionRetentionWork(tx, { companyId: issueCompany.companyId });
+        await notifyIssueAttentionRetentionWork(tx, { companyId: issueCompany.companyId, issueId: id });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyRunDispatch(tx, { companyId: issueCompany.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
@@ -12281,7 +12278,7 @@ export function issueService(db: Db) {
           const staleExecutionRunId = current.executionRunId;
           const adopted = await db.transaction(async tx => {
             await notifyStatusCardIssueChange(tx, issueCompany.companyId);
-            await notifyAttentionRetentionWork(tx, { companyId: issueCompany.companyId });
+            await notifyIssueAttentionRetentionWork(tx, { companyId: issueCompany.companyId, issueId: id });
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
             await notifyRunDispatch(tx, { companyId: issueCompany.companyId });
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
@@ -12539,7 +12536,7 @@ export function issueService(db: Db) {
           existing.status === "in_progress" ? "todo" : existing.status;
         await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
         await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id });
-        await notifyAttentionRetentionWork(tx, { companyId: existing.companyId });
+        await notifyIssueAttentionRetentionWork(tx, { companyId: existing.companyId, issueId: id });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyRunDispatch(tx, { companyId: existing.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
@@ -12595,7 +12592,7 @@ export function issueService(db: Db) {
         }
 
         await notifyStatusCardIssueChange(tx, existing.companyId);
-        await notifyAttentionRetentionWork(tx, { companyId: existing.companyId });
+        await notifyIssueAttentionRetentionWork(tx, { companyId: existing.companyId, issueId: id });
         await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
         if (existing.checkoutRunId || existing.executionRunId || options.clearAssignee) {
           await notifyExternalConversationSettlement(tx, { companyId: existing.companyId, issueId: id });

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/native-status-reassessment-notifications.ts

@@ -1,5 +1,5 @@
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
-import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
+import { notifyIssueAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
@@ -7,7 +7,7 @@ export type NativeStatusReassessmentScope = { companyId: string; issueId?: strin
 
 export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
   await notifyStatusCardIssueChange(tx, scope.companyId);
-  await notifyAttentionRetentionWork(tx, { companyId: scope.companyId });
+  await notifyIssueAttentionRetentionWork(tx, { companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId });
   await notifyIssueRecovery(tx, scope);
   return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment,
     JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId }));

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/retention-work-notifications.ts

@@ -1,10 +1,54 @@
 import { notifyApprovalWork } from "./approval-work-notifications.js";
+import { RECOVERY_ORIGIN_KINDS, parseIssueGraphLivenessIncidentKey } from "./recovery/origins.js";
+import { sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import type { AttentionSourceKind } from "@paperclipai/shared";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export async function notifyAttentionRetentionWork(transaction: object, scope: { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string } = {}) {
-  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.attentionRetention, JSON.stringify(scope));
+export async function notifyAttentionRetentionWork(transaction: object, scope: { companyId?: string; sourceKind?: AttentionSourceKind; issueId?: string; sourceId?: string } = {}) {
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.attentionRetention, JSON.stringify({
+    companyId: scope.companyId, sourceKind: scope.sourceKind, sourceId: scope.sourceId, issueId: scope.issueId,
+  }));
+}
+
+export async function notifyIssueAttentionRetentionWork(transaction: object,
+  scope: { companyId: string; issueId?: string; parentId?: string | null; runId?: string; originKind?: string | null; originId?: string | null }) {
+  const database = transaction as Db;
+  const seeds = [scope.issueId, scope.parentId].filter((id): id is string => Boolean(id));
+  if (!seeds.length && scope.runId) {
+    const rows = await database.execute(sql`select issue_id::text as id from heartbeat_runs
+      where company_id = ${scope.companyId}::uuid and id = ${scope.runId}::uuid and issue_id is not null
+      union select native_issue_id::text from heartbeat_runs
+      where company_id = ${scope.companyId}::uuid and id = ${scope.runId}::uuid and native_issue_id is not null
+      union select coalesce(context_snapshot->>'issueId', context_snapshot->>'taskId') from heartbeat_runs
+      where company_id = ${scope.companyId}::uuid and id = ${scope.runId}::uuid
+        and coalesce(context_snapshot->>'issueId', context_snapshot->>'taskId') is not null`);
+    for (const row of rows) seeds.push(row.id as string);
+  }
+  if (!seeds.length) return notifyAttentionRetentionWork(transaction, { companyId: scope.companyId });
+  const origins = await database.execute(sql`select id::text as id, origin_kind as "originKind", origin_id as "originId" from issues
+    where company_id = ${scope.companyId}::uuid
+      and id in (${sql.join([...new Set(seeds)].map(id => sql`${id}::uuid`), sql`, `)})`);
+  const current = origins.find(source => source.id === scope.issueId);
+  origins.push({ originKind: scope.originKind ?? current?.originKind, originId: scope.originId ?? current?.originId });
+  for (const source of origins) {
+    const incident = source.originKind === RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation
+      ? parseIssueGraphLivenessIncidentKey(source.originId as string | null) : null;
+    if (incident?.companyId === scope.companyId) seeds.push(incident.issueId, incident.leafIssueId);
+  }
+  const affected = await database.execute(sql`with recursive affected(id) as (
+    select unnest(array[${sql.join([...new Set(seeds)].map(id => sql`${id}::uuid`), sql`, `)}])
+    union
+    select parent.id from affected current cross join lateral (
+      select related_issue_id as id from issue_relations where company_id = ${scope.companyId}::uuid
+        and issue_id = current.id and type = 'blocks'
+      union select parent_id from issues where company_id = ${scope.companyId}::uuid
+        and id = current.id and parent_id is not null
+    ) parent
+  ) select id::text as "issueId" from affected`);
+  for (const row of affected) await notifyAttentionRetentionWork(transaction, {
+    companyId: scope.companyId, issueId: row.issueId as string,
+  });
 }
 
 export async function notifyRetentionWork(transaction: object, kind: AttentionSourceKind, scope: { companyId?: string; sourceId?: string } = {}) {

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/status-cards.ts

@@ -158,7 +158,7 @@ export function statusCardService(
   const issuesSvc = () => deps.issuesSvc ?? issueService(db);
   const searchSvc = companySearchService(db);
   type Transaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
-  function mutate<T>(companyId: string | undefined, write: (tx: Transaction) => Promise<T>) {
+  function mutate<T>(companyId: string, write: (tx: Transaction) => Promise<T>) {
     return db.transaction(async tx => {
       await notifyStatusCardEvaluation(tx, companyId);
       return write(tx);
@@ -335,7 +335,9 @@ export function statusCardService(
   }
 
   async function remove(id: string) {
-    return mutate(undefined, tx => tx.delete(statusCards).where(eq(statusCards.id, id)).returning().then((rows) => rows[0] ?? null));
+    const card = await getById(id);
+    if (!card) return null;
+    return mutate(card.companyId, tx => tx.delete(statusCards).where(eq(statusCards.id, id)).returning().then((rows) => rows[0] ?? null));
   }
 
   async function listUpdates(cardId: string) {

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); } });