perf(server): cut steady-state DB hot paths in dashboard, attention, and productivity sweeps (#10992)

<!-- ASD-STE100 -->

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - The server keeps fleet health with periodic sweeps and shows a
dashboard with run activity
> - The Paperclip instance became slow again after the first round of
recovery-sweep indexes landed
> - Live profiling found four steady-state hot paths that read much more
data than they use
> - This pull request bounds the dashboard recursion, adds the missing
taskKey index, and narrows two wide reads
> - The benefit is a large drop in constant database load and a
responsive server

## Linked Issues or Issue Description

**Describe the bug**
The server becomes slow while agents work. Live query sampling shows
four hot paths:
1. The dashboard run-activity recursive CTE reads every run a company
ever had on each call. One call takes 2.85 seconds. The UI calls it
after almost every fleet event through the dashboard and sidebar-badges
routes.
2. The productivity-review sweep runs each 30 seconds. Its run-scope
filter is `issueId OR taskId OR taskKey` on the run context JSONB. No
index exists for `taskKey`. The planner must detoast every run snapshot
for the agent. One query takes 444 ms and the sweep makes one for each
of ~152 candidate issues.
3. The attention failed-run section selects the full `context_snapshot`
for every run newer than the oldest exhausted run. That fetch moves 29
MB for each feed build.
4. The retention sweep pages the attention feed with a cursor. Each page
makes a full feed rebuild.

**Expected behavior**
Periodic sweeps and dashboard queries read only the data they use, and
use indexes.

**Actual behavior**
The database stays saturated. Users see a slow server.

## What Changed

- `server/src/services/dashboard.ts`: bound both arms of the
`recovered_runs` recursive CTE to the chart window. A retry is always
newer than the run it retries, so the bound cannot change visible chart
data. Live time went from 2,852 ms to 54 ms.
- `packages/db/src/migrations/0210_heartbeat_context_taskkey_index.sql`:
add the `taskKey` expression index that completes the
issueId/taskId/taskKey trio. With all three, the planner uses a
BitmapOr. Live time for the productivity run-scope query went from 444
ms to 1.9 ms.
- `packages/db/src/schema/heartbeat_runs.ts`: mirror the new index in
the Drizzle schema.
- `server/src/services/productivity-review.ts`: select only the seven
run fields the evidence code reads. Before, the query pulled full rows
with `result_json` (up to 43 kB per row, 100 rows per issue).
- `server/src/services/attention.ts`: project `issueId`/`taskId` text
fields instead of the full `context_snapshot` in the failed-run
newer-runs query (29 MB per feed build before).
- `server/src/index.ts`: the retention sweep now builds the attention
feed once per company with `all: true` instead of one full rebuild per
cursor page.
- `packages/db/src/heartbeat-context-snapshot-index-migration.test.ts`:
cover the new index and re-run migration 0210 statements to prove
idempotency.

## Verification

- `pnpm --filter @paperclipai/db typecheck` (includes migration
numbering and safety checks) — pass.
- `npx tsc --noEmit` in `server/` — pass.
- `npx vitest run
packages/db/src/heartbeat-context-snapshot-index-migration.test.ts` —
pass (embedded Postgres, full migration chain, planner assertions,
idempotent re-run of 0209 and 0210).
- `npx vitest run` on attention, dashboard, productivity-review,
decision-retention, issue-blocker-attention, and issue-review-attention
test files — 72/72 pass.
- Live EXPLAIN ANALYZE before/after numbers are in the What Changed
list.

## Risks

- Migration 0210 builds one btree index without CONCURRENTLY inside the
transactional migration runner. The table is not in the large-table
bucket. The 0209 twin built in seconds on a 100k-row live table.
- The CTE bound excludes retry ancestors that are older than the chart
window. Those rows are not visible to the chart query, so chart output
does not change.
- The attention projection changes JSONB scalar handling in one edge
case: a non-string `issueId`/`taskId` value now casts to text instead of
reading as absent. These keys are always strings in practice.
- The retention sweep now holds one full feed in memory per company. The
cursor loop already accumulated all pages into one array, so peak memory
is unchanged.

## Model Used

Claude Fable 5 (`claude-fable-5`, Anthropic, Mythos-class tier, extended
thinking + tool use) via Paperclip agent runtime.

- [x] I searched existing PRs and issues and this change is not a
duplicate.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

packages/db/src/heartbeat-context-snapshot-index-migration.test.ts

@@ -26,6 +26,7 @@ d("heartbeat context_snapshot expression index migration", () => {
     const names = idx.map((r) => r.indexname as string);
     expect(names).toContain("heartbeat_runs_company_ctx_issue_created_idx");
     expect(names).toContain("heartbeat_runs_company_ctx_task_created_idx");
+    expect(names).toContain("heartbeat_runs_company_ctx_taskkey_created_idx");
     expect(names).toContain("agent_wakeup_requests_company_payload_issue_idx");
 
     await sql.unsafe("SET enable_seqscan = off");
@@ -41,6 +42,16 @@ d("heartbeat context_snapshot expression index migration", () => {
     const taskText = taskPlan.map((r) => Object.values(r)[0]).join("\n");
     expect(taskText).toContain("heartbeat_runs_company_ctx_task_created_idx");
 
+    // The productivity-review run scope ORs issueId/taskId/taskKey; all three
+    // expression indexes must exist so the planner can BitmapOr at real row
+    // counts instead of detoasting every run snapshot for the agent. An empty
+    // table plans a single index scan, so assert the taskKey index directly.
+    const taskKeyPlan = await sql.unsafe(
+      "EXPLAIN SELECT id FROM heartbeat_runs WHERE company_id = '00000000-0000-0000-0000-000000000001' AND context_snapshot ->> 'taskKey' = 'x' ORDER BY created_at DESC, id DESC LIMIT 1",
+    );
+    const taskKeyText = taskKeyPlan.map((r) => Object.values(r)[0]).join("\n");
+    expect(taskKeyText).toContain("heartbeat_runs_company_ctx_taskkey_created_idx");
+
     const wakePlan = await sql.unsafe(
       "EXPLAIN SELECT id FROM agent_wakeup_requests WHERE company_id = '00000000-0000-0000-0000-000000000001' AND status = 'deferred_issue_execution' AND payload ->> 'issueId' = 'x' LIMIT 1",
     );
@@ -49,17 +60,22 @@ d("heartbeat context_snapshot expression index migration", () => {
 
     // Idempotency: re-running the migration statements against an already
     // migrated database must be a no-op, not an error.
-    const migrationSql = await readFile(
-      fileURLToPath(new URL("./migrations/0209_heartbeat_context_snapshot_indexes.sql", import.meta.url)),
-      "utf8",
-    );
-    const statements = migrationSql
-      .split("--> statement-breakpoint")
-      .map((s) => s.trim())
-      .filter((s) => s.length > 0);
-    expect(statements.length).toBeGreaterThan(0);
-    for (const statement of statements) {
-      await sql.unsafe(statement);
+    for (const migration of [
+      "./migrations/0209_heartbeat_context_snapshot_indexes.sql",
+      "./migrations/0210_heartbeat_context_taskkey_index.sql",
+    ]) {
+      const migrationSql = await readFile(
+        fileURLToPath(new URL(migration, import.meta.url)),
+        "utf8",
+      );
+      const statements = migrationSql
+        .split("--> statement-breakpoint")
+        .map((s) => s.trim())
+        .filter((s) => s.length > 0);
+      expect(statements.length).toBeGreaterThan(0);
+      for (const statement of statements) {
+        await sql.unsafe(statement);
+      }
     }
   }, 240_000);
 });

packages/db/src/migrations/0210_heartbeat_context_taskkey_index.sql

@@ -0,0 +1 @@
+CREATE INDEX IF NOT EXISTS "heartbeat_runs_company_ctx_taskkey_created_idx" ON "heartbeat_runs" USING btree ("company_id", ("context_snapshot" ->> 'taskKey'), "created_at" DESC);

packages/db/src/migrations/meta/_journal.json

@@ -1457,6 +1457,13 @@
       "when": 1786020026023,
       "tag": "0209_heartbeat_context_snapshot_indexes",
       "breakpoints": true
+    },
+    {
+      "idx": 210,
+      "version": "7",
+      "when": 1786032087897,
+      "tag": "0210_heartbeat_context_taskkey_index",
+      "breakpoints": true
     }
   ]
-}
+}

packages/db/src/schema/heartbeat_runs.ts

@@ -99,5 +99,10 @@ export const heartbeatRuns = pgTable(
       sql`(${table.contextSnapshot} ->> 'taskId')`,
       table.createdAt.desc(),
     ),
+    companyCtxTaskKeyCreatedIdx: index("heartbeat_runs_company_ctx_taskkey_created_idx").on(
+      table.companyId,
+      sql`(${table.contextSnapshot} ->> 'taskKey')`,
+      table.createdAt.desc(),
+    ),
   }),
 );

server/src/index.ts

@@ -1133,18 +1133,15 @@ export async function startServer(): Promise<StartedServer> {
       const activeCompanies = await db.select({ id: companies.id }).from(companies).where(eq(companies.status, "active"));
       let archived = 0;
       for (const company of activeCompanies) {
-        const items = [];
-        let cursor: string | undefined;
-        do {
-          const page = await attentionService(db as any).list(company.id, {
-            includeDismissed: true,
-            limit: 100,
-            cursor,
-          });
-          items.push(...page.items);
-          cursor = page.nextCursor ?? undefined;
-        } while (cursor);
-        archived += await retentionExecutor.autoArchive({ companyId: company.id, items });
+        // Cursor pagination rebuilds the whole feed for every page; one
+        // unscoped all-items build keeps this sweep at a single feed build
+        // per company per tick.
+        const page = await attentionService(db as any).list(company.id, {
+          includeDismissed: true,
+          all: true,
+          allowUnscopedAll: true,
+        });
+        archived += await retentionExecutor.autoArchive({ companyId: company.id, items: page.items });
       }
       const notifications = await retentionExecutor.deliverNotifications();
       return { archived, ...notifications };

server/src/services/attention.ts

@@ -1704,7 +1704,10 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
             .select({
               agentId: heartbeatRuns.agentId,
               createdAt: heartbeatRuns.createdAt,
-              contextSnapshot: heartbeatRuns.contextSnapshot,
+              // Project just the ids readRunIssueId needs; pulling the whole
+              // context_snapshot detoasts megabytes per feed build.
+              runIssueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`,
+              runTaskId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'taskId'`,
             })
             .from(heartbeatRuns)
             .where(and(
@@ -1716,7 +1719,8 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
       ]);
       const latestRunCreatedAtByKey = new Map<string, Date>();
       for (const newerRun of newerRuns) {
-        const newerRunKey = `${newerRun.agentId}:${readRunIssueId(newerRun.contextSnapshot) ?? ""}`;
+        const newerRunIssueId = readRunIssueId({ issueId: newerRun.runIssueId, taskId: newerRun.runTaskId });
+        const newerRunKey = `${newerRun.agentId}:${newerRunIssueId ?? ""}`;
         const latestCreatedAt = latestRunCreatedAtByKey.get(newerRunKey);
         if (!latestCreatedAt || newerRun.createdAt > latestCreatedAt) {
           latestRunCreatedAtByKey.set(newerRunKey, newerRun.createdAt);

server/src/services/dashboard.ts

@@ -102,6 +102,10 @@ export function dashboardService(db: Db) {
       // restart-killed run whose retry succeeded is pulled out of the headline
       // failed count. error_code is carried through so a failure spike can be
       // attributed to an error class (e.g. process_lost, provider_quota).
+      // Both recursive arms are bounded to the chart window: a retry is always
+      // created after the run it retries, so ancestors of an out-of-window
+      // child are themselves out of window and invisible to the membership
+      // test below. Unbounded, the seed walks every run the company ever had.
       const runActivityRows = (await db.execute(sql`
         WITH RECURSIVE recovered_runs(id) AS (
           SELECT parent.id
@@ -109,11 +113,13 @@ export function dashboardService(db: Db) {
           JOIN ${heartbeatRuns} AS parent ON parent.id = child.retry_of_run_id
           WHERE child.company_id = ${companyId}
             AND child.status = 'succeeded'
+            AND child.created_at >= ${runActivityStart.toISOString()}::timestamptz
           UNION
           SELECT parent.id
           FROM recovered_runs rr
           JOIN ${heartbeatRuns} AS child ON child.id = rr.id
           JOIN ${heartbeatRuns} AS parent ON parent.id = child.retry_of_run_id
+          WHERE child.created_at >= ${runActivityStart.toISOString()}::timestamptz
         )
         SELECT
           to_char(run.created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS date,

server/src/services/productivity-review.ts

@@ -44,6 +44,12 @@ export const PRODUCTIVITY_REVIEW_REFRESH_COMMENT_PREFIX = "Productivity review e
 type IssueRow = typeof issues.$inferSelect;
 type AgentRow = typeof agents.$inferSelect;
 type HeartbeatRunRow = typeof heartbeatRuns.$inferSelect;
+// Evidence only reads these run fields; selecting the full row detoasts
+// result_json/context_snapshot for up to MAX_RUNS_FOR_STREAK runs per issue.
+type ProductivityRunSample = Pick<
+  HeartbeatRunRow,
+  "id" | "agentId" | "status" | "livenessState" | "createdAt" | "nextAction" | "usageJson"
+>;
 type ProductivityReviewTrigger = "no_comment_streak" | "long_active_duration" | "high_churn";
 
 type ProductivityReviewThresholds = {
@@ -74,7 +80,7 @@ type ProductivityReviewEvidence = {
   commentCountLastHour: number;
   commentCountLastSixHours: number;
   elapsedMs: number | null;
-  latestRuns: HeartbeatRunRow[];
+  latestRuns: ProductivityRunSample[];
   latestComments: Array<typeof issueComments.$inferSelect>;
   costCents: number;
   usageSamples: Array<{ runId: string; usageJson: Record<string, unknown> | null }>;
@@ -447,7 +453,15 @@ export function productivityReviewService(db: Db, deps?: { enqueueWakeup?: Enque
     const sixHoursAgo = new Date(now.getTime() - 6 * 60 * 60 * 1000);
 
     const latestRuns = await db
-      .select()
+      .select({
+        id: heartbeatRuns.id,
+        agentId: heartbeatRuns.agentId,
+        status: heartbeatRuns.status,
+        livenessState: heartbeatRuns.livenessState,
+        createdAt: heartbeatRuns.createdAt,
+        nextAction: heartbeatRuns.nextAction,
+        usageJson: heartbeatRuns.usageJson,
+      })
       .from(heartbeatRuns)
       .where(
         and(