merge: retain bounded attention facts and trusted recovery origins

Co-Authored-By: Paperclip <noreply@paperclip.ing>

* commit '71d7b9f09e5a1a0c5c38bbd4084802dfd52f313d':
  fix(server): bound blocker facts and preserve origin authority
  test: model retention reads in native cleanup transaction fixture

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, 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.
+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. Opaque liveness and stranded-recovery origin references resolve through company-scoped issue rows before entering UUID queries, so valid none sentinels do not roll back writes and moved recovery tasks still invalidate their original review. 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. The shared graph fetch stores at most 2,000 nodes and 4,000 edges, and bounds each frontier query before materializing rows. An internal batch result identifies roots whose evaluation is incomplete; the legacy map API exposes no extra transport fields. Retention schedules only those roots for an immediate fresh single-root check, which disarms or saves the actual source deadline. It never archives a partial intermediary or adds a recurring incomplete-root poll. Fully evaluated roots preserve their exact classification. 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

@@ -17,7 +17,7 @@ import { buildIssueGraphLivenessIncidentKey, RECOVERY_ORIGIN_KINDS } from "../se
 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";
-import { notifyAttentionRetentionWork } from "../services/retention-work-notifications.js";
+import { notifyAttentionRetentionWork, notifyIssueAttentionRetentionWork } from "../services/retention-work-notifications.js";
 import { idleWorkSnapshot } from "../services/task-admission.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 
@@ -278,12 +278,16 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
     expect(expected.terminalBlockerIssueId).toBe(terminal.id);
     expect(limited.terminalBlockerIssueId).toBe(limited.directBlockerIssueId);
     for (const input of [[root, heavy], [heavy, root]]) {
-      const batch = await owner.listBlockerAttention(companyId, input);
-      expect(batch.get(root.id)).toEqual(expected); expect(batch.get(heavy.id)).toEqual(limited);
+      const batch = await owner.listBlockerAttentionBatch(companyId, input);
+      if (!batch.deferredRootIds.has(root.id)) expect(batch.attention.get(root.id)).toEqual(expected);
+      if (!batch.deferredRootIds.has(heavy.id)) expect(batch.attention.get(heavy.id)).toEqual(limited);
+      expect(batch.nodeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES);
+      if (limit === "depth") expect(batch.deferredRootIds.has(root.id)).toBe(false);
     }
     const page = await attentionRetentionService(db).load({ companyId, sourceKind: "blocker_attention" });
     const entry = page.rows!.find(item => item.rootId === root.id && item.variant === "terminal")!;
-    expect(entry.sourceId).toBe(terminal.id); expect(entry.dueAt).toBeLessThan(Date.now());
+    expect(entry.dueAt).toBeLessThanOrEqual(Date.now());
+    if (limit === "depth") expect(entry.sourceId).toBe(terminal.id);
     await attentionRetentionService(db).check(entry);
     expect((await state("blocker_attention", terminal.id))?.archivedAt).toBeInstanceOf(Date);
     expect(await db.select().from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.sourceId, terminal.id))).toHaveLength(1);
@@ -317,20 +321,79 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
     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
+    await db.execute(sql`insert into issues (id, company_id, parent_id, title, status, updated_at)
+      select gen_random_uuid(), ${companyId}::uuid, root.id, 'parameter-branch-' || n, 'todo', ${old().toISOString()}::timestamptz
+      from issues root cross join generate_series(1, 10) n
       where root.company_id = ${companyId}::uuid and root.status = 'blocked'`);
+    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, branch.id, 'parameter-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz
+      from issues branch cross join generate_series(1, 99) n
+      where branch.company_id = ${companyId}::uuid and branch.title like 'parameter-branch-%'`);
+    const completed = await insertIssue("blocked");
+    const shallow = await insertIssue("blocked", { updatedAt: new Date() });
+    const middle = await insertIssue("todo", { updatedAt: new Date() });
+    const overdue = await insertIssue("backlog", { assigneeAgentId: agentId });
+    await db.insert(issueRelations).values([
+      { companyId, issueId: middle.id, relatedIssueId: shallow.id, type: "blocks" },
+      { companyId, issueId: overdue.id, relatedIssueId: middle.id, type: "blocks" },
+    ]);
     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 => {
+    const result = await issueService(db).listBlockerAttentionBatch(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(result.attention.size).toBe(72);
+    expect(result.deferredRootIds.has(completed.id)).toBe(false);
+    expect(result.attention.get(completed.id)?.terminalBlockerIssueId).toBe(completed.id);
+    expect([...result.attention.values()].every(item => item.state === "needs_attention" && item.terminalBlockerIssueId)).toBe(true);
     expect(Math.max(...parameterCounts)).toBeLessThanOrEqual(1100);
+    expect(parameterCounts.length).toBeLessThanOrEqual(100);
+    expect(result.nodeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES);
+    expect(result.edgeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES * 2);
+    expect(result.fetchedRows).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES * 6);
+    expect(result.deferredRootIds.has(shallow.id)).toBe(true);
+    const service = attentionRetentionService(db);
+    const page = await service.load({ companyId, sourceKind: "blocker_attention" });
+    const immediate = page.rows!.find(entry => entry.rootId === shallow.id && entry.variant === "terminal")!;
+    expect(immediate.dueAt).toBeLessThanOrEqual(Date.now());
+    await service.check(immediate);
+    expect((await state("blocker_attention", overdue.id))?.archivedAt).toBeInstanceOf(Date);
   }, 120_000);
 
+  it("preserves observed live work when a single large root exhausts shared fetch capacity", async () => {
+    const root = await insertIssue("blocked");
+    const live = await insertIssue("in_progress", { assigneeAgentId: agentId });
+    await db.insert(issueRelations).values({ companyId, issueId: live.id, relatedIssueId: root.id, type: "blocks" });
+    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}::uuid, 'bounded-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz
+      from generate_series(1, 3000) n`);
+    const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]);
+    expect(batch.deferredRootIds.has(root.id)).toBe(true);
+    expect(batch.attention.get(root.id)?.blockingTreeLive).toBe(true);
+    const service = attentionRetentionService(db), page = await service.load({ companyId, sourceKind: "blocker_attention" });
+    const entry = page.rows!.find(item => item.rootId === root.id && item.variant === "terminal")!;
+    expect(await service.check(entry)).toBeNull();
+    expect(await state("blocker_attention", root.id)).toBeNull();
+    expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
+  });
+
+  it("disarms finite deferred-root checks when large retained graphs have no archive obligation", async () => {
+    await db.execute(sql`insert into issues (id, company_id, title, status, updated_at)
+      select gen_random_uuid(), ${companyId}::uuid, 'held-root-' || n, 'blocked', ${old().toISOString()}::timestamptz
+      from generate_series(1, 4) 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, 'held-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz
+      from issues root cross join generate_series(1, 600) n
+      where root.company_id = ${companyId}::uuid and root.status = 'blocked'`);
+    await db.execute(sql`insert into decision_retention (company_id, source_kind, source_id, source_activity_at, keep)
+      select company_id, 'blocker_attention', id::text, updated_at, true from issues
+      where company_id = ${companyId}::uuid and status = 'backlog'`);
+    const service = attentionRetentionService(db), check = vi.spyOn(service, "check");
+    await start(service); await quiet();
+    expect(check).toHaveBeenCalledTimes(4);
+    expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
+  });
+
   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 };
@@ -418,6 +481,49 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
     await archived("blocker_attention", terminal.id); await quiet();
   });
 
+  it("resolves opaque run issue references through trusted company issue rows", async () => {
+    const issue = await insertIssue();
+    const [valid, opaque] = await db.insert(heartbeatRuns).values([
+      { companyId, agentId, status: "succeeded", contextSnapshot: { issueId: issue.id } },
+      { companyId, agentId, status: "succeeded", contextSnapshot: { issueId: "none" } },
+    ]).returning();
+    const hints: string[] = [];
+    const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, values) => {
+      if (event === "settled") hints.push(...values ?? []);
+    });
+    try {
+      await db.transaction(tx => notifyIssueAttentionRetentionWork(tx, { companyId, runId: valid.id }));
+      await db.transaction(tx => notifyIssueAttentionRetentionWork(tx, { companyId, runId: opaque.id }));
+    } finally { unsubscribe(); }
+    expect(hints).toEqual([JSON.stringify({ companyId, issueId: issue.id }), JSON.stringify({ companyId })]);
+  });
+
+  it("accepts ordinary liveness incident creation and completion with a none leaf sentinel", async () => {
+    const source = await insertIssue("blocked", { responsibleUserId: "board" });
+    const service = issueService(db);
+    const incident = await service.create(companyId, { title: "Invalid participant", status: "todo", assigneeUserId: "board",
+      createdByUserId: "board", originKind: RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation,
+      originId: `harness_liveness:${companyId}:${source.id}:invalid_review_participant:none` });
+    await service.update(source.id, { blockedByIssueIds: [incident.id] });
+    await service.update(incident.id, { status: "done" });
+    expect((await service.getById(incident.id))?.status).toBe("done");
+    expect((await service.getRelationSummaries(source.id)).blockedBy).toEqual([]);
+  });
+
+  it.each(["hidden", "done"] as const)("reactivates an older review when moved stranded recovery is %s", async transition => {
+    const review = await insertIssue("in_review", { assigneeAgentId: agentId, responsibleUserId: "board" });
+    const parent = await insertIssue("todo", { responsibleUserId: "board" });
+    const recovery = await insertIssue("todo", { parentId: review.id, originKind: RECOVERY_ORIGIN_KINDS.strandedIssueRecovery,
+      originId: review.id, assigneeUserId: "board", responsibleUserId: "board" });
+    await issueService(db).update(recovery.id, { parentId: parent.id });
+    expect((await issueService(db).listReviewAttention(companyId, [review])).get(review.id)?.state).not.toBe("stalled");
+    await start(); await quiet();
+    await issueService(db).update(recovery.id, transition === "hidden" ? { hiddenAt: new Date() } : { status: "done" });
+    await settle();
+    expect((await state("review", review.id))?.archivedAt).toBeInstanceOf(Date);
+    expect(await db.select().from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.sourceId, review.id))).toHaveLength(1);
+  });
+
   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__/issues-service.test.ts

@@ -4151,7 +4151,7 @@ const dependencyTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSU
 
   beforeAll(async () => {
     if (!dependencyTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issues-blockers-");
-    db = createDb(dependencyTestDatabaseUrl ?? tempDb!.connectionString);
+    db = createDb(dependencyTestDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 });
     svc = issueService(db);
     await ensureIssueRelationsTable(db);
   }, 20_000);

server/src/__tests__/work-products.test.ts

@@ -1,3 +1,5 @@
+import { PgDialect } from "drizzle-orm/pg-core";
+import type { SQL } from "drizzle-orm";
 import { describe, expect, it, vi } from "vitest";
 import {
   enrichWorkProductMetadataWithDiff,
@@ -33,6 +35,18 @@ function createWorkProductRow(overrides: Partial<Record<string, unknown>> = {})
   };
 }
 
+function retentionReads(insideTransaction: () => boolean) {
+  return vi.fn(async (query: SQL) => {
+    const statement = new PgDialect().sqlToQuery(query);
+    expect(insideTransaction()).toBe(true);
+    expect(statement.params).toContain("company-1");
+    expect(statement.params).toContain("issue-1");
+    if (statement.sql.includes("origin_kind")) return [{ id: "issue-1", originKind: null, originId: null }];
+    if (statement.sql.includes("with recursive affected")) return [{ issueId: "issue-1" }];
+    throw new Error(`Unexpected work-product fixture query: ${statement.sql}`);
+  });
+}
+
 describe("workProductService", () => {
   it("extracts runner totals and enriches work-product metadata", () => {
     const summary = workProductDiffSummaryFromEventPayload({
@@ -158,11 +172,16 @@ describe("workProductService", () => {
     const insertValues = vi.fn(() => ({ returning: insertReturning }));
     const txInsert = vi.fn(() => ({ values: insertValues }));
 
+    let transactionOpen = false;
     const tx = {
+      execute: retentionReads(() => transactionOpen),
       update: txUpdate,
       insert: txInsert,
     };
-    const transaction = vi.fn(async (callback: (input: typeof tx) => Promise<unknown>) => await callback(tx));
+    const transaction = vi.fn(async (callback: (input: typeof tx) => Promise<unknown>) => {
+      transactionOpen = true;
+      try { return await callback(tx); } finally { transactionOpen = false; }
+    });
 
     const svc = workProductService({ transaction } as any);
     const result = await svc.createForIssue("issue-1", "company-1", {
@@ -194,11 +213,16 @@ describe("workProductService", () => {
     const updateSet = vi.fn(() => ({ where: updateWhere }));
     const txUpdate = vi.fn(() => ({ set: updateSet }));
 
+    let transactionOpen = false;
     const tx = {
+      execute: retentionReads(() => transactionOpen),
       select: txSelect,
       update: txUpdate,
     };
-    const transaction = vi.fn(async (callback: (input: typeof tx) => Promise<unknown>) => await callback(tx));
+    const transaction = vi.fn(async (callback: (input: typeof tx) => Promise<unknown>) => {
+      transactionOpen = true;
+      try { return await callback(tx); } finally { transactionOpen = false; }
+    });
 
     const svc = workProductService({ transaction } as any);
     const result = await svc.update("work-product-1", {

server/src/services/attention-retention.ts

@@ -32,6 +32,7 @@ type PageEvidence = {
   agents: Set<string>;
   visibleIssues: Map<string, PageIssue>;
   newerRuns: Set<string>;
+  deferredRoots: Set<string>;
   blockers: Awaited<ReturnType<ReturnType<typeof issueService>["listBlockerAttention"]>>;
 };
 
@@ -184,7 +185,9 @@ export function attentionRetentionService(db: Db) {
       const visibleRows = await readIssues(issueIds);
       const visible = new Map(visibleRows.map(row => [row.id, row]));
       const blockerRoots = visibleRows.filter(row => blockerIds.includes(row.id));
-      const blockers: PageEvidence["blockers"] = blockerRoots.length ? await issueService(database).listBlockerAttention(companyId, blockerRoots) : new Map();
+      const blockerBatch = blockerRoots.length ? await issueService(database).listBlockerAttentionBatch(companyId, blockerRoots)
+        : { attention: new Map(), deferredRootIds: new Set<string>() };
+      const blockers: PageEvidence["blockers"] = blockerBatch.attention;
       const terminalIds = [...blockers.values()].filter(terminalBlockerRequiresAttention)
         .map(attention => attention.terminalBlockerIssueId).filter((id): id is string => id != null && !visible.has(id));
       for (const row of await readIssues(terminalIds)) visible.set(row.id, row);
@@ -192,13 +195,17 @@ export function attentionRetentionService(db: Db) {
         decisions: new Map(decisionRows.map(row => [row.id, row])), recoveries: new Map(recoveryRows.map(row => [row.id, row])),
         runs: new Map(runRows.map(row => [row.id, row])), agents: new Set(runAgents.map(row => row.id)), visibleIssues: visible,
         newerRuns: await listFailedRunsWithNewerRun(database, companyId,
-          runRows.map(run => ({ ...run, issueId: readRunIssueId(run.contextSnapshot) }))), blockers,
+          runRows.map(run => ({ ...run, issueId: readRunIssueId(run.contextSnapshot) }))), blockers, deferredRoots: blockerBatch.deferredRootIds,
       });
     }
     return result;
   }
 
   async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false, reviewAttention?: Map<string, boolean>, interactionAttention?: Map<string, boolean>, evidence?: PageEvidence): Promise<Entry | null> {
+    if (candidate.sourceKind === "blocker_attention" && variant === "terminal" && evidence?.deferredRoots.has(candidate.sourceId)) {
+      return { ...candidate, rootId: candidate.sourceId, variant,
+        id: `${candidate.sourceKind}:${candidate.sourceId}:terminal`, dueAt: Date.now() };
+    }
     let sourceId = candidate.sourceId;
     let activityAt = candidate.activityAt;
     let dueAt = candidate.deadlineAt;

server/src/services/issue-thread-interactions.test.ts

@@ -1,5 +1,6 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
-import { getTableName } from "drizzle-orm";
+import { PgDialect } from "drizzle-orm/pg-core";
+import { getTableName, type SQL } from "drizzle-orm";
 
 const mockCreateChild = vi.fn();
 
@@ -48,8 +49,18 @@ function createFakeDb(args: {
   const toolActionRequestUpdates: Array<Record<string, unknown>> = [];
   const inserts: Array<{ table: string; values: Record<string, unknown> }> = [];
   let selectCallCount = 0;
+  let transactionDepth = 0;
 
   const db: any = {
+    execute: vi.fn(async (query: SQL) => {
+      const statement = new PgDialect().sqlToQuery(query);
+      expect(transactionDepth).toBeGreaterThan(0);
+      expect(statement.params).toContain(interactionRow.companyId);
+      expect(statement.params).toContain(interactionRow.issueId);
+      if (statement.sql.includes("origin_kind")) return [{ id: interactionRow.issueId, originKind: null, originId: null }];
+      if (statement.sql.includes("with recursive affected")) return [{ issueId: interactionRow.issueId }];
+      throw new Error(`Unexpected interaction fixture query: ${statement.sql}`);
+    }),
     select: vi.fn(() => {
       selectCallCount += 1;
       const rows = selectCallCount === 1 ? [interactionRow] : (args.parentRows ?? []);
@@ -84,7 +95,10 @@ function createFakeDb(args: {
         inserts.push({ table: getTableName(table as never), values });
       },
     })),
-    transaction: async (callback: (tx: typeof db) => Promise<void>) => callback(db),
+    transaction: async (callback: (tx: typeof db) => Promise<void>) => {
+      transactionDepth++;
+      try { return await callback(db); } finally { transactionDepth--; }
+    },
   };
 
   return {
@@ -177,6 +191,7 @@ describe("issueThreadInteractionService", () => {
     };
 
     let selectCallCount = 0;
+  let transactionDepth = 0;
     const db: any = {
       select: vi.fn(() => {
         selectCallCount += 1;

server/src/services/issues.ts

@@ -3443,6 +3443,7 @@ type IssueBlockerAttentionEdge = {
   blockerIssueId: string;
 };
 type IssueBlockerAttentionQueryRow = IssueBlockerAttentionNode & {
+  executionWorkspaceId?: string | null;
   issueId: string | null;
   blockerIssueId: string;
 };
@@ -3831,11 +3832,15 @@ async function terminalExplicitBlockersByRoot(
   return terminalByRoot;
 }
 
-async function listIssueBlockerAttentionMap(
+async function listIssueBlockerAttentionMap(dbOrTx: any, companyId: string, issueRows: IssueBlockerAttentionInputNode[]) {
+  return (await readIssueBlockerAttentionBatch(dbOrTx, companyId, issueRows)).attention;
+}
+
+async function readIssueBlockerAttentionBatch(
   dbOrTx: any,
   companyId: string,
   issueRows: IssueBlockerAttentionInputNode[],
-): Promise<Map<string, IssueBlockerAttention>> {
+) {
   const roots = issueRows.filter(
     (row) => row.companyId === companyId && row.status === "blocked",
   );
@@ -3845,37 +3850,38 @@ async function listIssueBlockerAttentionMap(
       attentionMap.set(row.id, createIssueBlockerAttention());
     }
   }
-  if (roots.length === 0) return attentionMap;
+  const deferredRootIds = new Set<string>(roots.slice(BLOCKER_ATTENTION_MAX_NODES).map(root => root.id));
+  let edgeCount = 0, fetchedRows = 0;
+  const result = () => ({ attention: attentionMap, deferredRootIds, nodeCount: nodesById.size, edgeCount, fetchedRows });
+  if (roots.length === 0) return { attention: attentionMap, deferredRootIds, nodeCount: 0, edgeCount, fetchedRows };
 
   const nodesById = new Map<string, IssueBlockerAttentionNode>();
   const edgesByIssueId = new Map<string, IssueBlockerAttentionEdge[]>();
-  for (const root of roots) nodesById.set(root.id, { ...root });
+  const admittedRoots = roots.slice(0, BLOCKER_ATTENTION_MAX_NODES);
+  for (const root of admittedRoots) nodesById.set(root.id, { ...root });
 
-  type Traversal = { nodes: Set<string>; expanded: Set<string>; frontier: string[]; truncated: boolean };
-  const traversals = new Map<string, Traversal>(roots.map(root => [root.id, {
-    nodes: new Set([root.id]), expanded: new Set<string>(), frontier: [root.id], truncated: false,
+  type Traversal = { rootId: string; nodes: Set<string>; expanded: Set<string>; frontier: string[]; truncated: boolean };
+  const traversals = new Map<string, Traversal>(admittedRoots.map(root => [root.id, {
+    rootId: root.id, nodes: new Set([root.id]), expanded: new Set<string>(), frontier: [root.id], truncated: false,
   }]));
   const expanded = new Set<string>();
+  const queried = new Set<string>();
   const pendingFinalizeBlockerIssueIds = new Set<string>();
   for (let depth = 0; depth < BLOCKER_ATTENTION_MAX_DEPTH; depth += 1) {
     const active = [...traversals.values()].filter(traversal => !traversal.truncated && traversal.frontier.length > 0);
     if (active.length === 0) break;
-    const frontier = [...new Set(active.flatMap(traversal => traversal.frontier))].filter(id => !expanded.has(id));
+    const unread = new Set<string>();
+    for (const traversal of active) for (const id of traversal.frontier) if (!expanded.has(id)) unread.add(id);
+    const frontier = [...unread];
 
     for (const chunk of chunkList(
       frontier,
       ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE,
     )) {
-      const readinessByIssueId = await listIssueDependencyReadinessMap(
-        dbOrTx,
-        companyId,
-        chunk,
-      );
-      for (const readiness of readinessByIssueId.values()) {
-        for (const blockerIssueId of readiness.pendingFinalizeBlockerIssueIds) {
-          pendingFinalizeBlockerIssueIds.add(blockerIssueId);
-        }
-      }
+      const remaining = Math.min(BLOCKER_ATTENTION_MAX_NODES - nodesById.size,
+        BLOCKER_ATTENTION_MAX_NODES * 2 - edgeCount);
+      if (remaining <= 0) break;
+      const rowLimit = remaining + 1;
       const explicitBlockerRowsPromise: Promise<
         IssueBlockerAttentionQueryRow[]
       > = dbOrTx
@@ -3889,6 +3895,7 @@ async function listIssueBlockerAttentionMap(
           title: issues.title,
           status: issues.status,
           executionRunId: issues.executionRunId,
+          executionWorkspaceId: issues.executionWorkspaceId,
           assigneeAgentId: issues.assigneeAgentId,
           assigneeUserId: issues.assigneeUserId,
         })
@@ -3901,7 +3908,7 @@ async function listIssueBlockerAttentionMap(
             inArray(issueRelations.relatedIssueId, chunk),
             eq(issues.companyId, companyId),
           ),
-        );
+        ).limit(rowLimit);
       const childRowsPromise: Promise<IssueBlockerAttentionQueryRow[]> = dbOrTx
         .select({
           issueId: issues.parentId,
@@ -3926,56 +3933,56 @@ async function listIssueBlockerAttentionMap(
               BLOCKER_ATTENTION_CHILD_TERMINAL_STATUSES,
             ),
           ),
-        );
+        ).limit(rowLimit);
       const [explicitBlockerRows, childRows] = await Promise.all([
         explicitBlockerRowsPromise,
         childRowsPromise,
       ]);
 
+      for (const id of chunk) queried.add(id);
+      fetchedRows += explicitBlockerRows.length + childRows.length;
+      const pendingFinalize = await listPendingFinalizeBlockerIssueIds(dbOrTx, companyId,
+        explicitBlockerRows.flatMap(row => row.status === "done" && row.executionWorkspaceId
+          ? [{ blockerIssueId: row.blockerIssueId, executionWorkspaceId: row.executionWorkspaceId }] : []));
+      for (const id of pendingFinalize) pendingFinalizeBlockerIssueIds.add(id);
+      let complete = explicitBlockerRows.length < rowLimit && childRows.length < rowLimit;
       const unresolvedExplicitBlockerRows = explicitBlockerRows.filter(
         (row) =>
           row.status !== "done" ||
           pendingFinalizeBlockerIssueIds.has(row.blockerIssueId),
       );
-      appendBlockerAttentionEdges(edgesByIssueId, [
-        ...unresolvedExplicitBlockerRows
-          .filter(
-            (row): row is IssueBlockerAttentionQueryRow & { issueId: string } =>
-              row.issueId !== null,
-          )
-          .map((row) => ({
-            issueId: row.issueId,
-            blockerIssueId: row.blockerIssueId,
-          })),
-        ...childRows
-          .filter(
-            (row): row is IssueBlockerAttentionQueryRow & { issueId: string } =>
-              row.issueId !== null,
-          )
-          .map((row) => ({
-            issueId: row.issueId,
-            blockerIssueId: row.blockerIssueId,
-          })),
-      ]);
-
       for (const row of [...unresolvedExplicitBlockerRows, ...childRows]) {
-        if (!row.issueId || nodesById.has(row.blockerIssueId)) continue;
-        nodesById.set(row.blockerIssueId, {
-          id: row.blockerIssueId,
-          companyId: row.companyId,
-          parentId: row.parentId,
-          identifier: row.identifier,
-          title: row.title,
-          status: row.status,
-          executionRunId: row.executionRunId,
-          assigneeAgentId: row.assigneeAgentId,
-          assigneeUserId: row.assigneeUserId,
-        });
+        if (!row.issueId) continue;
+        if (!nodesById.has(row.blockerIssueId)) {
+          if (nodesById.size >= BLOCKER_ATTENTION_MAX_NODES) { complete = false; continue; }
+          nodesById.set(row.blockerIssueId, {
+            id: row.blockerIssueId, companyId: row.companyId, parentId: row.parentId, identifier: row.identifier,
+            title: row.title, status: row.status, executionRunId: row.executionRunId,
+            assigneeAgentId: row.assigneeAgentId, assigneeUserId: row.assigneeUserId,
+          });
+        }
+        const edges = edgesByIssueId.get(row.issueId) ?? [];
+        if (edges.some(edge => edge.blockerIssueId === row.blockerIssueId)) continue;
+        if (edgeCount >= BLOCKER_ATTENTION_MAX_NODES * 2) { complete = false; continue; }
+        edges.push({ issueId: row.issueId, blockerIssueId: row.blockerIssueId });
+        edgesByIssueId.set(row.issueId, edges);
+        edgeCount++;
       }
-      for (const id of chunk) expanded.add(id);
+      if (complete) for (const id of chunk) expanded.add(id);
+
     }
 
     for (const traversal of active) {
+      if (traversal.frontier.some(id => !expanded.has(id))) {
+        for (const id of traversal.frontier) if (queried.has(id)) {
+          traversal.expanded.add(id);
+          for (const edge of edgesByIssueId.get(id) ?? []) if (nodesById.has(edge.blockerIssueId)) traversal.nodes.add(edge.blockerIssueId);
+        }
+        traversal.truncated = true;
+        traversal.frontier = [];
+        deferredRootIds.add(traversal.rootId);
+        continue;
+      }
       const nextFrontier = new Set<string>();
       for (const id of traversal.frontier) {
         traversal.expanded.add(id);
@@ -4555,7 +4562,7 @@ async function listIssueBlockerAttentionMap(
     );
   }
 
-  return attentionMap;
+  return result();
 }
 
 type IssueReviewAttentionInput = Pick<IssueRow, "id" | "companyId" | "status">;
@@ -9393,6 +9400,9 @@ export function issueService(db: Db) {
       return listIssueDependencyReadinessMap(dbOrTx, companyId, issueIds);
     },
 
+    listBlockerAttentionBatch: async (companyId: string, issueRows: IssueBlockerAttentionInputNode[], dbOrTx: any = db) =>
+      readIssueBlockerAttentionBatch(dbOrTx, companyId, issueRows),
+
     listBlockerAttention: async (
       companyId: string,
       issueRows: IssueBlockerAttentionInputNode[],

server/src/services/native-runtime/native-session-executor.test.ts

@@ -4302,6 +4302,17 @@ describe("retained native cleanup activation", () => {
       releaseCommit = release;
     });
     const db = {
+      execute: async (query: SQL) => {
+        // Retention hints read the fixture's run/issue relationship inside the
+        // same cleanup transaction; this fixture has one issue and no ancestors.
+        const statement = new PgDialect().sqlToQuery(query);
+        expect(transactionOpen).toBe(true);
+        expect(statement.params).toContain(run.companyId);
+        if (statement.sql.includes("from heartbeat_runs")) return [{ id: run.nativeIssueId }];
+        if (statement.sql.includes("origin_kind")) return [{ id: run.nativeIssueId, originKind: null, originId: null }];
+        if (statement.sql.includes("with recursive affected")) return [{ issueId: run.nativeIssueId }];
+        throw new Error(`Unexpected cleanup fixture query: ${statement.sql}`);
+      },
       select: () => ({
         from: (table: unknown) => {
           const rows =

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

@@ -20,9 +20,10 @@ export async function notifyIssueAttentionRetentionWork(transaction: object,
       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`);
+      union select issue.id::text from heartbeat_runs run inner join issues issue
+        on issue.company_id = run.company_id
+          and issue.id::text = coalesce(run.context_snapshot->>'issueId', run.context_snapshot->>'taskId')
+      where run.company_id = ${scope.companyId}::uuid and run.id = ${scope.runId}::uuid`);
     for (const row of rows) seeds.push(row.id as string);
   }
   if (!seeds.length) return notifyAttentionRetentionWork(transaction, { companyId: scope.companyId });
@@ -31,10 +32,20 @@ export async function notifyIssueAttentionRetentionWork(transaction: object,
       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 });
+  const references: string[] = [];
   for (const source of origins) {
+    if (source.originKind === RECOVERY_ORIGIN_KINDS.strandedIssueRecovery && typeof source.originId === "string") {
+      references.push(source.originId);
+    }
     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);
+    if (incident?.companyId === scope.companyId) references.push(incident.issueId, incident.leafIssueId);
+  }
+  if (references.length) {
+    const trusted = await database.execute(sql`select id::text as id from issues
+      where company_id = ${scope.companyId}::uuid
+        and id::text in (${sql.join([...new Set(references)].map(id => sql`${id}`), sql`, `)})`);
+    for (const row of trusted) seeds.push(row.id as string);
   }
   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`, `)}])