perf(server): batch remaining attention retention page evidence

Load company-partitioned decision/recovery links, run/agent visibility and newer-run probes once per page; share one blocker graph reconstruction and exact SQL timestamps across page entries. Keep fresh per-entry archive locks and revalidation. Add full100-source query budgets and late visibility/supersession safety controls.

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

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. Ordinary execution, decision and budget mutations pass their affected company; exact decision sources are included.
+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. 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

@@ -171,6 +171,91 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
     expect(statements.filter(query => query.includes('from "agents"'))).toHaveLength(1);
   });
 
+  it("batches decision and recovery source links and rechecks a late hidden origin", async () => {
+    const origins = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId,
+      title: randomUUID(), status: "todo", updatedAt: old() }))).returning();
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded" }).returning();
+    const decisionRows = await db.insert(decisions).values(origins.map(origin => ({ 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 recoveryRows = await db.insert(issueRecoveryActions).values(origins.map(origin => ({ companyId,
+      sourceIssueId: origin.id, kind: "execution", ownerType: "board", cause: "fixture", fingerprint: "fixture",
+      nextAction: "Choose", updatedAt: old() }))).returning();
+    const service = attentionRetentionService(db);
+    const pages = [];
+    for (const sourceKind of ["decision", "recovery_action"] as const) {
+      statements.length = 0;
+      const page = await service.load({ companyId, sourceKind });
+      expect(page.rows).toHaveLength(100); expect(page.complete).toBe(false);
+      expect(statements.length).toBeLessThanOrEqual(3);
+      pages.push(page.rows!);
+    }
+    await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, origins[99].id));
+    const hiddenDecision = pages[0].find(entry => entry.sourceId === decisionRows[99].id)!;
+    const hiddenRecovery = pages[1].find(entry => entry.sourceId === recoveryRows[99].id)!;
+    expect(await service.check(hiddenDecision)).toBeNull(); expect(await service.check(hiddenRecovery)).toBeNull();
+    expect(await state("decision", hiddenDecision.sourceId)).toBeNull(); expect(await state("recovery_action", hiddenRecovery.sourceId)).toBeNull();
+    expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
+  });
+
+  it("batches 100 failed runs and rechecks late visibility and supersession before archive", async () => {
+    const origins = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId,
+      title: randomUUID(), status: "todo", updatedAt: old() }))).returning();
+    const runs = await db.insert(heartbeatRuns).values(origins.map(origin => ({ companyId, agentId,
+      status: "failed", contextSnapshot: { issueId: origin.id }, createdAt: old(), finishedAt: old() }))).returning();
+    await db.insert(heartbeatRunEvents).values(runs.map(run => ({ companyId, agentId, runId: run.id,
+      seq: 1, eventType: "lifecycle", message: "Bounded retry exhausted: batch fixture" })));
+    const service = attentionRetentionService(db), scope = { companyId, sourceKind: "failed_run" as const };
+    statements.length = 0;
+    const page = await service.load(scope);
+    expect(page.rows).toHaveLength(100); expect(page.complete).toBe(false);
+    expect(statements.length).toBeLessThanOrEqual(6);
+    const hiddenRun = runs[98], supersededRun = runs[99];
+    const hidden = page.rows!.find(entry => entry.sourceId === hiddenRun.id)!;
+    const superseded = page.rows!.find(entry => entry.sourceId === supersededRun.id)!;
+    await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, origins[98].id));
+    await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded",
+      contextSnapshot: { issueId: origins[99].id }, createdAt: new Date(old().getTime() + 60_000), finishedAt: old() });
+    expect(await service.check(hidden)).toBeNull(); expect(await service.check(superseded)).toBeNull();
+    expect(await state("failed_run", hiddenRun.id)).toBeNull(); expect(await state("failed_run", supersededRun.id)).toBeNull();
+    expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
+    expect((await service.load(scope)).complete).toBe(true);
+    const refreshed = await service.load(scope);
+    expect(refreshed.rows).toHaveLength(98);
+    expect(refreshed.rows!.some(entry => [hiddenRun.id, supersededRun.id].includes(entry.sourceId))).toBe(false);
+  });
+
+  it("batches 100 blocker graphs with exact terminal timestamps and fresh terminal visibility", async () => {
+    const roots = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId,
+      title: randomUUID(), status: "blocked", updatedAt: old(), blockedTransitionAt: old(),
+      unblockDescriptor: { owner: "board", action: "Pick an owner" } }))).returning();
+    const terminals = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId,
+      title: randomUUID(), status: "backlog", assigneeAgentId: agentId, updatedAt: old() }))).returning();
+    await db.insert(issueRelations).values(roots.map((root, index) => ({ companyId,
+      issueId: terminals[index].id, relatedIssueId: root.id, type: "blocks" })));
+    await db.execute(sql`update issues set updated_at = updated_at + interval '0.000321 seconds' where id = ${terminals[99].id}::uuid`);
+    const service = attentionRetentionService(db), scope = { companyId, sourceKind: "blocker_attention" as const };
+    statements.length = 0;
+    const page = await service.load(scope);
+    const queries = [...statements];
+    expect(page.rows).toHaveLength(200); expect(page.complete).toBe(false);
+    const retentionReads = queries.filter(query => query.includes('from "decision_retention"'));
+    expect(retentionReads).toHaveLength(200);
+    expect(queries.length - retentionReads.length).toBeLessThanOrEqual(30);
+    const terminal = page.rows!.find(entry => entry.rootId === roots[99].id && entry.variant === "terminal")!;
+    expect(terminal.sourceId).toBe(terminals[99].id);
+    const [expected] = await db.execute(sql`select updated_at::text as at,
+      ceil(extract(epoch from updated_at) * 1000 + ${delay}) as due from issues where id = ${terminals[99].id}::uuid`);
+    expect(terminal.activityAt).toBe(expected.at); expect(terminal.dueAt).toBe(Number(expected.due));
+    await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, terminals[99].id));
+    expect(await service.check(terminal)).toBeNull();
+    expect(await state("blocker_attention", terminals[99].id)).toBeNull();
+    expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
+    expect((await service.load(scope)).complete).toBe(true);
+    const refreshed = await service.load(scope);
+    expect(refreshed.rows!.some(entry => entry.sourceId === terminals[99].id)).toBe(false);
+  });
+
   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

@@ -1,4 +1,4 @@
-import { and, eq, inArray, isNull, sql, type SQL } from "drizzle-orm";
+import { and, eq, getTableColumns, inArray, isNull, sql, type SQL } from "drizzle-orm";
 import { agents, decisionRetention, decisions, heartbeatRuns,
   issueApprovals, issueRecoveryActions, issueThreadInteractions, issues, approvals, type Db } from "@paperclipai/db";
 import type { AttentionSourceKind } from "@paperclipai/shared";
@@ -23,6 +23,18 @@ type Candidate = { sourceKind: Kind; sourceId: string; companyId: string; activi
 type Entry = Candidate & { id: string; dueAt: number; rootId: string; variant?: "owner" | "terminal" };
 type Cursor = { sourceKind: string; sourceId: string };
 
+type PageIssue = typeof issues.$inferSelect & { activityAt: string; deadlineAt: number;
+  blockedActivityAt: string | null; blockedDeadlineAt: number | null };
+type PageEvidence = {
+  decisions: Map<string, { originIssueId: string }>;
+  recoveries: Map<string, typeof issueRecoveryActions.$inferSelect>;
+  runs: Map<string, typeof heartbeatRuns.$inferSelect>;
+  agents: Set<string>;
+  visibleIssues: Map<string, PageIssue>;
+  newerRuns: Set<string>;
+  blockers: Awaited<ReturnType<ReturnType<typeof issueService>["listBlockerAttention"]>>;
+};
+
 const sourceTables: Record<Kind, SQL> = {
   decision: sql`decisions`, issue_thread_interaction: sql`issue_thread_interactions`,
   recovery_action: sql`issue_recovery_actions`, blocker_attention: sql`issues`, review: sql`issues`,
@@ -137,56 +149,106 @@ export function attentionRetentionService(db: Db) {
     return result;
   }
 
-  async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false, reviewAttention?: Map<string, boolean>, interactionAttention?: Map<string, boolean>): Promise<Entry | null> {
+  async function pageEvidence(database: Db, candidates: Candidate[]) {
+    const byCompany = new Map<string, Candidate[]>();
+    for (const candidate of candidates) {
+      if (!["decision", "recovery_action", "failed_run", "blocker_attention"].includes(candidate.sourceKind)) continue;
+      const rows = byCompany.get(candidate.companyId) ?? [];
+      rows.push(candidate);
+      byCompany.set(candidate.companyId, rows);
+    }
+    const result = new Map<string, PageEvidence>();
+    for (const [companyId, rows] of byCompany) {
+      const ids = (kind: Kind) => rows.filter(row => row.sourceKind === kind).map(row => row.sourceId);
+      const decisionIds = ids("decision"), recoveryIds = ids("recovery_action"), runIds = ids("failed_run"), blockerIds = ids("blocker_attention");
+      const decisionRows = decisionIds.length ? await database.select({ id: decisions.id, originIssueId: decisions.originIssueId })
+        .from(decisions).where(and(eq(decisions.companyId, companyId), inArray(decisions.id, decisionIds), eq(decisions.status, "open"))) : [];
+      const recoveryRows = recoveryIds.length ? await database.select().from(issueRecoveryActions).where(and(
+        eq(issueRecoveryActions.companyId, companyId), inArray(issueRecoveryActions.id, recoveryIds))) : [];
+      const runRows = runIds.length ? await database.select().from(heartbeatRuns).where(and(
+        eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.id, runIds))) : [];
+      const runAgents = runRows.length ? await database.select({ id: agents.id }).from(agents).where(and(
+        eq(agents.companyId, companyId), inArray(agents.id, [...new Set(runRows.map(row => row.agentId))]),
+        sql`${agents.status} <> 'terminated'`)) : [];
+      const readIssues = async (issueIds: string[]) => issueIds.length ? database.select({ ...getTableColumns(issues),
+        activityAt: sql<string>`${issues.updatedAt}::text`,
+        deadlineAt: sql<number>`ceil(extract(epoch from ${issues.updatedAt}) * 1000 + ${archiveDelay})`.mapWith(Number),
+        blockedActivityAt: sql<string | null>`${issues.blockedTransitionAt}::text`,
+        blockedDeadlineAt: sql<number | null>`ceil(extract(epoch from ${issues.blockedTransitionAt}) * 1000 + ${archiveDelay})`.mapWith(Number),
+      }).from(issues).where(and(eq(issues.companyId, companyId), inArray(issues.id, [...new Set(issueIds)]),
+        attentionIssueCondition(false))) : [];
+      const issueIds = [...blockerIds, ...decisionRows.map(row => row.originIssueId), ...recoveryRows.map(row => row.sourceIssueId),
+        ...runRows.map(row => readRunIssueId(row.contextSnapshot))].filter((id): id is string => Boolean(id));
+      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 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);
+      result.set(companyId, {
+        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,
+      });
+    }
+    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> {
     let sourceId = candidate.sourceId;
     let activityAt = candidate.activityAt;
     let dueAt = candidate.deadlineAt;
     if (candidate.sourceKind === "decision") {
-      const [row] = await database.select({ originIssueId: decisions.originIssueId }).from(decisions)
-        .where(and(eq(decisions.companyId, candidate.companyId), eq(decisions.id, sourceId), eq(decisions.status, "open")));
-      if (!row || row.originIssueId && !await visibleIssue(database, candidate.companyId, row.originIssueId)) return null;
+      const row = evidence ? evidence.decisions.get(sourceId) : (await database.select({ originIssueId: decisions.originIssueId }).from(decisions)
+        .where(and(eq(decisions.companyId, candidate.companyId), eq(decisions.id, sourceId), eq(decisions.status, "open"))))[0];
+      if (!row || row.originIssueId && !(evidence ? evidence.visibleIssues.has(row.originIssueId)
+        : await visibleIssue(database, candidate.companyId, row.originIssueId))) return null;
     } else if (candidate.sourceKind === "issue_thread_interaction") {
       const attention = interactionAttention ?? await interactionEligibility(database, [candidate]);
       if (!attention.get(sourceId)) return null;
     } else if (candidate.sourceKind === "recovery_action") {
-      const [row] = await database.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, candidate.companyId), eq(issueRecoveryActions.id, sourceId)));
-      if (!row || !await visibleIssue(database, candidate.companyId, row.sourceIssueId)) return null;
+      const row = evidence ? evidence.recoveries.get(sourceId) : (await database.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, candidate.companyId), eq(issueRecoveryActions.id, sourceId))))[0];
+      if (!row || !(evidence ? evidence.visibleIssues.has(row.sourceIssueId)
+        : await visibleIssue(database, candidate.companyId, row.sourceIssueId))) return null;
     } else if (candidate.sourceKind === "review") {
       const attention = reviewAttention ?? await reviewEligibility(database, [candidate]);
       if (!attention.get(sourceId)) return null;
     } else if (candidate.sourceKind === "blocker_attention") {
-      const row = await visibleIssue(database, candidate.companyId, sourceId);
+      const row = evidence ? evidence.visibleIssues.get(sourceId) : await visibleIssue(database, candidate.companyId, sourceId);
       if (!row) return null;
       if (variant === "owner") {
         if (row.unblockDescriptor?.owner !== "board" || !isProspectiveBlockedTransition(row)) return null;
-        const [activity] = await database.execute(sql`select blocked_transition_at::text as at,
+        const activity = evidence ? { at: evidence.visibleIssues.get(sourceId)!.blockedActivityAt!, due: evidence.visibleIssues.get(sourceId)!.blockedDeadlineAt! } : (await database.execute(sql`select blocked_transition_at::text as at,
           ceil(extract(epoch from blocked_transition_at) * 1000 + ${archiveDelay}) as due from issues
-          where company_id = ${candidate.companyId}::uuid and id = ${row.id}::uuid`) as unknown as Array<{ at: string; due: string }>;
+          where company_id = ${candidate.companyId}::uuid and id = ${row.id}::uuid`) as unknown as Array<{ at: string; due: string }>)[0];
         activityAt = activity.at; dueAt = Number(activity.due);
       } else {
-        const attention = (await issueService(database).listBlockerAttention(candidate.companyId, [row])).get(row.id);
+        const attention = evidence ? evidence.blockers.get(row.id) : (await issueService(database).listBlockerAttention(candidate.companyId, [row])).get(row.id);
         if (!terminalBlockerRequiresAttention(attention)) return null;
         sourceId = attention.terminalBlockerIssueId ?? row.id;
         if (lock && sourceId !== candidate.sourceId) await database.execute(sql`select id from issues
           where company_id = ${candidate.companyId}::uuid and id = ${sourceId}::uuid for update`);
-        const terminal = await visibleIssue(database, candidate.companyId, sourceId);
+        const terminal = evidence ? evidence.visibleIssues.get(sourceId) : await visibleIssue(database, candidate.companyId, sourceId);
         if (!terminal) return null;
-        const [activity] = await database.execute(sql`select updated_at::text as at,
+        const activity = evidence ? { at: evidence.visibleIssues.get(sourceId)!.activityAt, due: evidence.visibleIssues.get(sourceId)!.deadlineAt } : (await database.execute(sql`select updated_at::text as at,
           ceil(extract(epoch from updated_at) * 1000 + ${archiveDelay}) as due from issues
-          where company_id = ${candidate.companyId}::uuid and id = ${sourceId}::uuid`) as unknown as Array<{ at: string; due: string }>;
+          where company_id = ${candidate.companyId}::uuid and id = ${sourceId}::uuid`) as unknown as Array<{ at: string; due: string }>)[0];
         activityAt = activity.at; dueAt = Number(activity.due);
       }
       const [retention] = await database.select().from(decisionRetention).where(and(eq(decisionRetention.companyId, candidate.companyId),
         eq(decisionRetention.sourceKind, candidate.sourceKind), eq(decisionRetention.sourceId, sourceId)));
       if (retention?.keep || retention?.archivedAt) return null;
     } else if (candidate.sourceKind === "failed_run") {
-      const [run] = await database.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, sourceId)));
+      const run = evidence ? evidence.runs.get(sourceId) : (await database.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, sourceId))))[0];
       if (!run) return null;
-      const [agent] = await database.select({ id: agents.id }).from(agents).where(and(eq(agents.companyId, candidate.companyId),
-        eq(agents.id, run.agentId), sql`${agents.status} <> 'terminated'`));
+      const agent = evidence ? evidence.agents.has(run.agentId) : (await database.select({ id: agents.id }).from(agents).where(and(eq(agents.companyId, candidate.companyId),
+        eq(agents.id, run.agentId), sql`${agents.status} <> 'terminated'`)))[0];
       const issueId = readRunIssueId(run.contextSnapshot);
-      if (!agent || issueId && !await visibleIssue(database, candidate.companyId, issueId)) return null;
-      if ((await listFailedRunsWithNewerRun(database, candidate.companyId, [{ ...run, issueId }])).has(run.id)) return null;
+      if (!agent || issueId && !(evidence ? evidence.visibleIssues.has(issueId)
+        : await visibleIssue(database, candidate.companyId, issueId))) return null;
+      if ((evidence ? evidence.newerRuns : await listFailedRunsWithNewerRun(database, candidate.companyId, [{ ...run, issueId }])).has(run.id)) return null;
     }
     return { ...candidate, sourceId, activityAt, rootId: candidate.sourceId, variant,
       id: `${candidate.sourceKind}:${candidate.sourceId}:${variant ?? "source"}`, dueAt };
@@ -198,10 +260,11 @@ export function attentionRetentionService(db: Db) {
     const entries: Entry[] = [];
     const reviews = await reviewEligibility(db, rows);
     const interactions = await interactionEligibility(db, rows);
+    const evidence = await pageEvidence(db, rows);
     for (const row of rows) {
       if (signal?.aborted) return { rows: null, complete: true };
       for (const variant of row.sourceKind === "blocker_attention" ? ["owner", "terminal"] as const : [undefined]) {
-        const entry = await eligible(db, row, variant, false, reviews, interactions);
+        const entry = await eligible(db, row, variant, false, reviews, interactions, evidence.get(row.companyId));
         if (entry) entries.push(entry);
       }
     }