Integrate bounded attention-retention source reconciliation

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.
 

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/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/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-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/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 } = {}) {