diff --git a/packages/db/src/heartbeat-context-snapshot-index-migration.test.ts b/packages/db/src/heartbeat-context-snapshot-index-migration.test.ts index e4147bc6dc..36a4ea6a76 100644 --- a/packages/db/src/heartbeat-context-snapshot-index-migration.test.ts +++ b/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); }); diff --git a/packages/db/src/migrations/0210_heartbeat_context_taskkey_index.sql b/packages/db/src/migrations/0210_heartbeat_context_taskkey_index.sql new file mode 100644 index 0000000000..0edc5eb44e --- /dev/null +++ b/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); diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 06f3456dba..9961a106f5 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/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 } ] -} \ No newline at end of file +} diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 49d3dd8f61..6965e1669f 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/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(), + ), }), ); diff --git a/server/src/index.ts b/server/src/index.ts index 86d05f1a14..7350e7d99c 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1133,18 +1133,15 @@ export async function startServer(): Promise { 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 }; diff --git a/server/src/services/attention.ts b/server/src/services/attention.ts index 6ccee4fbff..0b4b72688e 100644 --- a/server/src/services/attention.ts +++ b/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`${heartbeatRuns.contextSnapshot} ->> 'issueId'`, + runTaskId: sql`${heartbeatRuns.contextSnapshot} ->> 'taskId'`, }) .from(heartbeatRuns) .where(and( @@ -1716,7 +1719,8 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions ]); const latestRunCreatedAtByKey = new Map(); 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); diff --git a/server/src/services/dashboard.ts b/server/src/services/dashboard.ts index 20d47dbfef..70c1b52c84 100644 --- a/server/src/services/dashboard.ts +++ b/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, diff --git a/server/src/services/productivity-review.ts b/server/src/services/productivity-review.ts index e694f2e222..199b122421 100644 --- a/server/src/services/productivity-review.ts +++ b/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; costCents: number; usageSamples: Array<{ runId: string; usageJson: Record | 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(