From 99edcd28a7a1a5933b52e4737214b217885088bf Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 10:15:48 -0600 Subject: [PATCH] perf(attention): read each retention candidate kind in its own bounded branch (#20566) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The attention retention worker archives old "Needs you" items, such as open decisions, pending questions, failed runs and items in review > - To find its work, the worker reads a page of retention candidates. It reads one page per changed scope, and one single-source read for each check of a due item > - On a hosted stack this candidate read ran 2,835 times at about 1 s per call, 4,201 s of database time in one afternoon window > - The read built every candidate of every kind before it applied the keyset limit. Its company and source filters used the `param is null or column = param` form. Under a generic prepared plan these filters cannot use an index. The failed-run branch also matched retry receipts with a bound LIKE parameter, so it could not use the partial index `heartbeat_run_events_retry_exhausted_idx` > - This pull request reads each kind in its own branch, puts every filter into that branch, and limits each branch to one page > - The benefit is a bounded read for each page, and a primary-key read for each single-source check, with the same rows, order and cursors ## Linked Issues or Issue Description **What existing behavior does this improve?** The attention retention candidate scan in `server/src/services/attention-retention.ts`. **Subsystem affected** Server: attention retention background work (database reads only). **Current behavior** Each page reads all candidates of all selected kinds, then sorts them and keeps 100. Optional filters use `($n is null or column = $n)`. A generic prepared plan cannot use an index for these filters. The single-source check compares `id::text`, so it cannot use the primary key. The retry receipt LIKE pattern is a parameter, so the plan cannot use the partial receipt index. **Proposed behavior** Each kind is one branch with its company, source, issue and keyset filters inside it and `limit 100`. Kinds before the cursor are not read. A source id filter uses the uuid primary key. The receipt pattern is a SQL literal that matches the partial index predicate. The page result does not change. Refs #20386 (adds the partial receipt index that this branch now uses). ## What Changed - `server/src/services/attention-retention.ts`: replace the single union scan with `candidateQuery`. Each source kind is a branch that has its own filters, retention join and one-page limit. The outer query still orders by `(source_kind, source_id)` and keeps 100 rows. - Only add a filter when the scope sets it. There is no `param is null or ...` form now. - A single-source read compares `id = $uuid`. It returns no rows for text that is not the canonical lowercase uuid form. The former `id::text = $text` comparison also never matched such text. - Skip the kinds that come before the keyset cursor kind. `ATTENTION_RETENTION_KIND_ORDER` holds the kind order, and a test checks it against the database collation. - The failed-run branch matches `'Bounded retry exhausted%'` as a literal, so it reads `heartbeat_run_events_retry_exhausted_idx`. - Export `listAttentionRetentionCandidates` for the test and the benchmark. The service uses it without a change in behavior. - New test `server/src/__tests__/attention-retention-candidates.test.ts`. It runs on real PostgreSQL and compares every page, row and cursor with the former scan (kept in the test as an oracle). It covers: one company, all companies, one kind, the blocker kind (which ignores retention), an unknown kind, issue scopes (including uppercase and malformed ids), exact source ids (uppercase, braces, no dashes, empty and malformed text return nothing), a `null` source id from a decoded scope, other-company isolation, inactive companies, kept and archived retention, and failed runs without a lifecycle receipt. - New benchmark `scripts/benchmark-attention-retention-candidates.ts`. It loads the baseline service from git, captures the SQL that each version sends for the same scopes, and times both under custom and generic plans. Each round alternates the order. It fails if the rows differ. - No migration. The existing indexes cover each branch: company and status indexes on each source table, the issues company/status index, and the partial receipt index from #20386. ## Verification - `cd server && npx vitest run src/__tests__/attention-retention-candidates.test.ts`: 10 passed. - `cd server && npx vitest run src/__tests__/attention-retention-work.test.ts`: see the note under Risks. - `cd server && npx tsc --noEmit`: pass. - Benchmark (local SQL measurement only): `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-attention-retention-candidates.ts --baseline-ref=8d2bf78323` Embedded PostgreSQL 18.1. Fixture: 12 companies (1 large and 11 at 1/5 size), 38,400 issues, 64,000 runs, 1.29 M run events, 6,400 decisions, 9,600 interactions, 4,800 recovery actions, 7,534 retention rows. Seven samples for each scope, after two warmups, with `plan_cache_mode` forced to custom and then to generic. Rows were equal in every sample. | Plan | Scope | Rows | Before median (min–max) | After median (min–max) | Speedup | EXPLAIN shared blocks before → after | | --- | --- | --- | --- | --- | --- | --- | | generic | single source (check path) | 1 | 4.82 ms (3.81–5.88) | 0.19 ms (0.12–0.66) | 25× | 3,491 → 10 | | generic | company failed runs | 81 | 45.0 ms (40.5–51.7) | 2.9 ms (2.3–4.6) | 15× | 3,752 → 3,752 | | generic | company page 1 | 100 | 39.4 ms (30.1–63.7) | 7.6 ms (5.8–14.3) | 5.2× | 4,396 → 4,719 | | generic | company page 2 | 100 | 33.3 ms (31.2–38.1) | 6.9 ms (5.2–10.3) | 4.8× | 4,237 → 4,712 | | generic | company issue scope | 4 | 11.8 ms (9.8–24.0) | 4.1 ms (3.4–6.4) | 2.9× | 4,243 → 4,225 | | generic | all companies page 1 | 100 | 107.2 ms (100.2–116.0) | 29.3 ms (18.8–64.4) | 3.7× | 13,657 → 14,672 | | custom | single source (check path) | 1 | 2.68 ms (2.01–5.33) | 0.81 ms (0.65–1.50) | 3.3× | 3,491 → 10 | | custom | company failed runs | 81 | 5.7 ms (3.9–15.8) | 4.6 ms (3.2–7.1) | 1.2× | 3,752 → 3,752 | | custom | company page 1 | 100 | 11.7 ms (9.2–13.3) | 10.7 ms (9.4–19.5) | 1.1× (neutral) | 4,396 → 4,719 | | custom | company page 2 | 100 | 11.2 ms (7.6–14.2) | 10.1 ms (8.5–12.0) | 1.1× (neutral) | 4,237 → 4,712 | | custom | company issue scope | 4 | 10.9 ms (9.1–13.3) | 8.4 ms (6.4–10.4) | 1.3× | 4,243 → 4,225 | | custom | all companies page 1 | 100 | 21.3 ms (17.4–25.9) | 24.4 ms (20.3–72.0) | 0.9× (neutral) | 13,688 → 14,672 | - The timing connection uses named prepared statements, as the server does by default. The benchmark checks `pg_prepared_statements` before it collects samples, so the samples reuse the parsed statement. - The driver uses named prepared statements by default, so after five runs PostgreSQL can switch a statement to a generic plan. The generic rows are the relevant case for a busy server. Custom plans were already good for most scopes, and the change is neutral there. - Plan after the change (company page, generic): the failed-run branch reads the partial receipt index (13 blocks), then probes `heartbeat_runs` by primary key. The single-source check reads each primary key and the retention unique index (10 blocks in all). ## Risks - Low behavior risk. The test compares the new pages with the former query on the same data, including cursors in the middle of a kind and at kind boundaries. - Kind skipping depends on the order of kind names. Every kind name is different from the others before its first underscore, so C and linguistic collations give the same order. A test checks the order against the database. - The benchmark fixture is smaller than the hosted database. The hosted cost (about 1 s per call) is larger than the local baseline. I did not measure the hosted improvement. It needs a fresh query-statistics window after deploy. - On this loaded local machine, one full run of `attention-retention-work.test.ts` had 3 tests time out at 30 s (no-run issue finalization, no-run workspace finalization, and interaction create/withdraw). The same 3 tests pass when run alone. CI is the authority for this suite. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic. Agentic coding with tool use (shell, file edits, local PostgreSQL). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- ...enchmark-attention-retention-candidates.ts | 229 ++++++++++++++++++ .../attention-retention-candidates.test.ts | 229 ++++++++++++++++++ server/src/services/attention-retention.ts | 121 ++++++--- 3 files changed, 539 insertions(+), 40 deletions(-) create mode 100644 scripts/benchmark-attention-retention-candidates.ts create mode 100644 server/src/__tests__/attention-retention-candidates.test.ts diff --git a/scripts/benchmark-attention-retention-candidates.ts b/scripts/benchmark-attention-retention-candidates.ts new file mode 100644 index 00000000000..384cbb9ca42 --- /dev/null +++ b/scripts/benchmark-attention-retention-candidates.ts @@ -0,0 +1,229 @@ +// Measures the attention retention candidate scan before and after it read each +// source kind in its own bounded branch. +// Usage (from the repository root): +// pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-attention-retention-candidates.ts \ +// --baseline-ref= [--companies=12] [--issues=12000] [--runs=20000] \ +// [--events-per-run=20] [--samples=7] [--plans] +// The script starts a disposable embedded PostgreSQL and loads one large company +// plus smaller decoy companies. It loads the baseline service from git, captures +// the candidate SQL that each service generates for the same scopes, and times +// that SQL under custom and generic plans. Baseline and treatment alternate, and +// the script asserts identical positional rows for every scope. +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { createDb, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; + +type Query = { sql: string; params: unknown[] }; +type Scope = Record; +const arg = (name: string, fallback: string) => + process.argv.find((value) => value.startsWith(`--${name}=`))?.slice(name.length + 3) ?? fallback; +const baselineRef = arg("baseline-ref", ""); +if (!baselineRef) throw new Error("Pass --baseline-ref="); +const companyCount = Number(arg("companies", "12")); +const issueCount = Number(arg("issues", "12000")); +const runCount = Number(arg("runs", "20000")); +const eventsPerRun = Number(arg("events-per-run", "20")); +const sampleCount = Number(arg("samples", "7")); +const git = (...args: string[]) => execFileSync("git", args, { encoding: "utf8" }).trim(); +const baselineSha = git("rev-parse", baselineRef); +const revision = git("rev-parse", "HEAD"); +const servicePath = "server/src/services/attention-retention.ts"; +const baselineFile = fileURLToPath(new URL("../server/src/services/attention-retention.benchmark-baseline.ts", import.meta.url)); +const keepAlive = setInterval(() => {}, 1000); +let database: Awaited> | undefined; +let db: Db | undefined; +let timer: Db["$client"] | undefined; + +// Sizes scale with `scale`: the main company uses 1, each decoy 1/5. +async function seedCompany(client: Db["$client"], companyId: string, prefix: string, scale: number) { + const issuesN = Math.max(20, Math.floor(issueCount * scale)); + const runsN = Math.max(20, Math.floor(runCount * scale)); + await client.unsafe(`INSERT INTO companies (id, name, issue_prefix) VALUES ($1, $2, $3)`, [companyId, `Synthetic ${prefix}`, prefix]); + await client.unsafe(`INSERT INTO agents (id, company_id, name, adapter_type, status) + SELECT gen_random_uuid(), $1::uuid, 'Agent ' || a, 'codex_local', CASE WHEN a <= 3 THEN 'error' ELSE 'idle' END + FROM generate_series(1, 40) a`, [companyId]); + // Status mix: 6% blocked, 6% in review, a few hidden or harness issues among them. + await client.unsafe(`INSERT INTO issues (id, company_id, title, status, hidden_at, harness_kind, created_at, updated_at) + SELECT gen_random_uuid(), $1::uuid, 'Issue ' || n, + CASE WHEN n % 17 = 0 THEN 'blocked' WHEN n % 17 = 1 THEN 'in_review' WHEN n % 5 = 0 THEN 'done' ELSE 'todo' END, + CASE WHEN n % 97 = 0 THEN timestamptz '2026-09-01' END, + CASE WHEN n % 89 = 0 THEN 'eval' END, + timestamptz '2026-09-01' + n * interval '1 minute', timestamptz '2026-09-01' + n * interval '1 minute' + FROM generate_series(1, $2::int) n`, [companyId, issuesN]); + await client.unsafe(`CREATE TABLE IF NOT EXISTS bench_issues (company_id uuid, n int, id uuid)`); + await client.unsafe(`INSERT INTO bench_issues SELECT company_id, row_number() OVER (ORDER BY created_at), id FROM issues WHERE company_id = $1::uuid`, [companyId]); + await client.unsafe(`CREATE TABLE IF NOT EXISTS bench_runs (company_id uuid, n int, id uuid, agent_id uuid, issue_id uuid)`); + await client.unsafe(`INSERT INTO bench_runs + SELECT $1::uuid, g.n, gen_random_uuid(), + (SELECT id FROM agents WHERE company_id = $1::uuid ORDER BY name OFFSET (g.n % 40) LIMIT 1), + (SELECT b.id FROM bench_issues b WHERE b.company_id = $1::uuid AND b.n = 1 + g.n % $3::int) + FROM generate_series(1, $2::int) AS g(n)`, [companyId, runsN, issuesN]); + // About 10% failed and 3% timed out, with a wide context snapshot. + await client.unsafe(`INSERT INTO heartbeat_runs (id, company_id, agent_id, status, created_at, updated_at, finished_at, context_snapshot) + SELECT r.id, r.company_id, r.agent_id, + CASE WHEN r.n % 10 = 0 THEN 'failed' WHEN r.n % 33 = 0 THEN 'timed_out' ELSE 'succeeded' END, + timestamptz '2026-10-01' + r.n * interval '1 second', timestamptz '2026-10-01' + r.n * interval '1 second', + CASE WHEN r.n % 7 <> 0 THEN timestamptz '2026-10-01' + r.n * interval '1 second' + interval '5 minutes' END, + CASE r.n % 3 WHEN 0 THEN jsonb_build_object('issueId', r.issue_id::text, 'prompt', repeat(md5(r.n::text), 48)) + WHEN 1 THEN jsonb_build_object('taskId', r.issue_id::text) ELSE '{}'::jsonb END + FROM bench_runs r WHERE r.company_id = $1::uuid`, [companyId]); + await client.unsafe(`INSERT INTO heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, stream, message, created_at) + SELECT r.company_id, r.id, r.agent_id, s, + CASE WHEN s % 10 = 1 THEN 'lifecycle' ELSE 'item.completed' END, + CASE WHEN s % 10 = 1 THEN 'system' ELSE 'stdout' END, + CASE WHEN s % 10 = 1 THEN 'Run step ' || s ELSE md5(r.n::text || ':' || s) || md5(s::text) END, + timestamptz '2026-10-01' + r.n * interval '1 second' + s * interval '1 millisecond' + FROM bench_runs r CROSS JOIN generate_series(1, $2::int) s + WHERE r.company_id = $1::uuid ORDER BY r.n / 20, s, r.n`, [companyId, eventsPerRun]); + // Retry-exhaustion receipts on 1 in 23 runs (most are not failed) plus an output decoy. + await client.unsafe(`INSERT INTO heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, stream, message, created_at) + SELECT r.company_id, r.id, r.agent_id, $2::int + x.k, x.event_type, 'system', x.message, now() + FROM bench_runs r CROSS JOIN LATERAL (VALUES + (1, 'lifecycle', 'Bounded retry exhausted after 3 scheduled attempts'), + (2, 'stdout', 'Bounded retry exhausted quoted in output')) AS x(k, event_type, message) + WHERE r.company_id = $1::uuid AND (r.n % 23 = 0 OR (x.k = 2 AND r.n % 10 = 0))`, [companyId, eventsPerRun]); + // Decisions on 1 in 6 issues; a third are open. + await client.unsafe(`INSERT INTO decisions (company_id, origin_agent_id, origin_issue_id, origin_run_id, title, body, options, status, expires_at, signed_spec, target_snapshots, updated_at) + SELECT r.company_id, r.agent_id, r.issue_id, r.id, 'Decision', 'Choose', '[]'::jsonb, + CASE WHEN r.n % 3 = 0 THEN 'open' ELSE 'resolved' END, timestamptz '2026-12-01', 'fixture', '{}'::jsonb, + timestamptz '2026-10-01' + r.n * interval '1 second' + FROM bench_runs r WHERE r.company_id = $1::uuid AND r.n % 6 = 0 AND r.n <= $2::int`, [companyId, issuesN]); + // Thread interactions on 1 in 4 issues; 15% pending, a few addressed to a user. + await client.unsafe(`INSERT INTO issue_thread_interactions (company_id, issue_id, kind, status, addressee_user_id, payload, updated_at) + SELECT i.company_id, i.id, 'ask_user_questions', CASE WHEN i.n % 7 = 0 THEN 'pending' ELSE 'answered' END, + CASE WHEN i.n % 21 = 0 THEN 'board' END, '{"version":1,"questions":[]}'::jsonb, timestamptz '2026-10-01' + i.n * interval '1 second' + FROM bench_issues i WHERE i.company_id = $1::uuid AND i.n % 4 = 0`, [companyId]); + // Recovery actions on 1 in 8 issues; a fifth active and human-owned. + await client.unsafe(`INSERT INTO issue_recovery_actions (company_id, source_issue_id, kind, status, owner_type, cause, fingerprint, next_action, updated_at) + SELECT i.company_id, i.id, 'execution', CASE WHEN i.n % 5 = 0 THEN 'active' ELSE 'resolved' END, + CASE WHEN i.n % 3 = 0 THEN 'agent' ELSE 'board' END, 'fixture', 'fp-' || i.n, 'Choose', timestamptz '2026-10-01' + i.n * interval '1 second' + FROM bench_issues i WHERE i.company_id = $1::uuid AND i.n % 8 = 0`, [companyId]); + // Retention state for about a third of the candidates: kept, archived, or plain. + await client.unsafe(`INSERT INTO decision_retention (company_id, source_kind, source_id, source_activity_at, keep, archived_at) + SELECT company_id, kind, id, now(), n % 3 = 0, CASE WHEN n % 3 = 1 THEN now() END FROM ( + SELECT company_id, 'review' AS kind, id::text AS id, row_number() OVER () AS n FROM issues WHERE company_id = $1::uuid AND status = 'in_review' + UNION ALL SELECT company_id, 'failed_run', id::text, row_number() OVER () FROM heartbeat_runs WHERE company_id = $1::uuid AND status = 'failed' + UNION ALL SELECT company_id, 'decision', id::text, row_number() OVER () FROM decisions WHERE company_id = $1::uuid AND status = 'open' + ) x WHERE n % 2 = 0`, [companyId]); +} + +const stats = (samples: number[]) => { + const sorted = [...samples].sort((a, b) => a - b); + return { + medianMs: Math.round(sorted[Math.floor(sorted.length / 2)] * 100) / 100, + minMs: Math.round(sorted[0] * 100) / 100, + maxMs: Math.round(sorted[sorted.length - 1] * 100) / 100, + }; +}; + +try { + database = await startEmbeddedPostgresTestDatabase("attention-retention-candidates-benchmark-"); + db = createDb(database.connectionString); + const client = db.$client; + const mainCompany = randomUUID(); + const seedStart = performance.now(); + await seedCompany(client, mainCompany, "BENCH", 1); + for (let index = 1; index < companyCount; index++) await seedCompany(client, randomUUID(), `DEC${index}`, 0.2); + await client.unsafe(`DROP TABLE bench_runs; DROP TABLE bench_issues`); + await client.unsafe("VACUUM ANALYZE"); + const seedMs = performance.now() - seedStart; + const [{ server_version: version }] = await client.unsafe("SHOW server_version"); + const [counts] = await client.unsafe(`SELECT (SELECT count(*) FROM issues)::int AS issues, (SELECT count(*) FROM heartbeat_runs)::int AS runs, + (SELECT count(*) FROM heartbeat_run_events)::int AS events, (SELECT count(*) FROM decisions)::int AS decisions, + (SELECT count(*) FROM issue_thread_interactions)::int AS interactions, (SELECT count(*) FROM issue_recovery_actions)::int AS recoveries, + (SELECT count(*) FROM decision_retention)::int AS retention`); + + // Capture the candidate SQL that each service version issues for a scope. + await fs.writeFile(baselineFile, git("show", `${baselineSha}:${servicePath}`)); + const captured: Query[] = []; + const capture = createDb(database.connectionString, { maxConnections: 1 }); + capture.$client.options.debug = (_connection: number, query: string, parameters: unknown[]) => { + if (query.includes("with source_candidates")) captured.push({ sql: query, params: [...parameters] }); + }; + const failedRun = (await client.unsafe(`SELECT r.id::text AS id FROM heartbeat_runs r WHERE r.company_id = $1::uuid AND r.status = 'failed' + AND EXISTS (SELECT 1 FROM heartbeat_run_events e WHERE e.run_id = r.id AND e.event_type = 'lifecycle' AND e.message LIKE 'Bounded retry exhausted%') + ORDER BY r.id LIMIT 1`, [mainCompany]))[0].id as string; + const blockedIssues = (await client.unsafe(`SELECT id::text AS id FROM issues WHERE company_id = $1::uuid AND status = 'blocked' ORDER BY id LIMIT 3`, [mainCompany])) + .map((row) => row.id as string); + const scenarios: Array<{ name: string; scope: Scope; page?: number }> = [ + { name: "company page 1", scope: { companyId: mainCompany } }, + { name: "company page 2", scope: { companyId: mainCompany }, page: 2 }, + { name: "company issue scope", scope: { companyId: mainCompany, issueIds: blockedIssues } }, + { name: "single source (check)", scope: { companyId: mainCompany, sourceKind: "failed_run", sourceId: failedRun } }, + { name: "company failed runs", scope: { companyId: mainCompany, sourceKind: "failed_run" } }, + { name: "all companies page 1", scope: {} }, + ]; + async function queriesFor(modulePath: string) { + const module = await import(modulePath); + const result: Query[] = []; + for (const scenario of scenarios) { + const service = module.attentionRetentionService(capture); + for (let page = 1; page <= (scenario.page ?? 1); page++) { + captured.length = 0; + await service.load(scenario.scope); + } + assert.ok(captured[0], `no candidate query captured for ${scenario.name}`); + result.push(captured[0]); + } + return result; + } + const baselineQueries = await queriesFor(baselineFile); + const treatmentQueries = await queriesFor(fileURLToPath(new URL(`../${servicePath}`, import.meta.url))); + await fs.rm(baselineFile, { force: true }); + await capture.$client.end({ timeout: 0 }); + + // One connection, so the plan_cache_mode setting applies to every sample. + // Named prepared statements, as the server uses by default: samples reuse + // the parsed statement and its cached plan, like a busy server. + timer = createDb(database.connectionString, { maxConnections: 1, prepare: true }).$client; + timer.options.onnotice = () => {}; + const timeOnce = async (query: Query) => { + const start = performance.now(); + const rows = await timer!.unsafe(query.sql, query.params as never[], { prepare: true }).values(); + return { ms: performance.now() - start, rows }; + }; + const plan = async (query: Query) => { + const [row] = await timer!.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]); + const root = (row["QUERY PLAN"] as Array>)[0]; + return { executionMs: Math.round(root["Execution Time"] * 100) / 100, + sharedBlocks: root.Plan["Shared Hit Blocks"] + root.Plan["Shared Read Blocks"], plan: process.argv.includes("--plans") ? root.Plan : undefined }; + }; + const report: Record[] = []; + for (const mode of ["force_custom_plan", "force_generic_plan"]) { + await timer.unsafe(`SET plan_cache_mode = ${mode}`); + for (const [index, scenario] of scenarios.entries()) { + const baseline = baselineQueries[index], treatment = treatmentQueries[index]; + const before: number[] = [], after: number[] = []; + let rows = 0; + for (let warm = 0; warm < 2; warm++) { await timeOnce(baseline); await timeOnce(treatment); } + const prepared = await timer.unsafe(`SELECT count(*)::int AS n FROM pg_prepared_statements WHERE statement IN ($1, $2)`, [baseline.sql, treatment.sql]); + assert.equal(prepared[0].n, 2, `${scenario.name}: both queries must run as prepared statements`); + // Alternate the order in each round to expose cache bias. + for (let sample = 0; sample < sampleCount; sample++) { + const order = sample % 2 === 0 ? [baseline, treatment] : [treatment, baseline]; + const results = []; + for (const query of order) results.push(await timeOnce(query)); + const [b, t] = sample % 2 === 0 ? results : [results[1], results[0]]; + assert.deepEqual(t.rows, b.rows, `${scenario.name}: rows differ`); + rows = b.rows.length; + before.push(b.ms); after.push(t.ms); + } + const beforeStats = stats(before), afterStats = stats(after); + report.push({ mode, scenario: scenario.name, rows, before: beforeStats, after: afterStats, + speedup: Math.round(beforeStats.medianMs / afterStats.medianMs * 10) / 10, + explainBefore: await plan(baseline), explainAfter: await plan(treatment) }); + } + } + console.log(JSON.stringify({ baselineSha, revision, postgresVersion: version, fixture: { companies: companyCount, ...counts, + seedMs: Math.round(seedMs) }, samples: sampleCount, report }, null, 2)); +} finally { + try { + await fs.rm(baselineFile, { force: true }); + await timer?.end({ timeout: 0 }); + await db?.$client.end({ timeout: 0 }); + await database?.cleanup(); + } finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/attention-retention-candidates.test.ts b/server/src/__tests__/attention-retention-candidates.test.ts new file mode 100644 index 00000000000..fc2b0ffa97e --- /dev/null +++ b/server/src/__tests__/attention-retention-candidates.test.ts @@ -0,0 +1,229 @@ +import { randomUUID } from "node:crypto"; +import { sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { applyPendingMigrations, createDb } from "@paperclipai/db"; +import { ATTENTION_RETENTION_KIND_ORDER, listAttentionRetentionCandidates } from "../services/attention-retention.js"; +import { DEFAULT_DECISION_ARCHIVE_DAYS } from "../services/decision-retention.js"; +import type { AttentionRetentionScope } from "../services/retention-work-notifications.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ATTENTION_RETENTION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +type Cursor = { sourceKind: string; sourceId: string }; +type Db = ReturnType; + +// The former single scan: every candidate of every kind, filtered and ordered +// after the union. It is the oracle for the per-kind branches. +function oracle(db: Db, scope: AttentionRetentionScope, cursor?: Cursor, sourceId?: string) { + const issueIds = scope.issueIds ?? (scope.issueId ? [scope.issueId] : []); + const kinds = ["decision", "issue_thread_interaction", "recovery_action", "blocker_attention", "review", "failed_run", "budget_alert", "agent_error_alert"] + .filter(kind => (!scope.sourceKind || scope.sourceKind === kind) && (!issueIds.length || !["budget_alert", "agent_error_alert"].includes(kind))); + if (!kinds.length) return Promise.resolve([]); + const branches: Record> = { + decision: sql`select 'decision', d.id::text, d.company_id, d.updated_at, d.origin_issue_id::text from decisions d where d.status = 'open'`, + 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, r.source_issue_id::text + from issue_recovery_actions r where r.status in ('active', 'escalated') and r.owner_type in ('user', 'board')`, + 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, 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), 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 'Bounded retry exhausted%')`, + 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 >= 85))`, + 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 = 'error'`, + }; + const id = sourceId ?? scope.sourceId ?? null; + return db.execute(sql`with source_candidates(source_kind, source_id, company_id, activity_at, issue_id) as (${sql.join(kinds.map(kind => branches[kind]), sql` union all `)}) + select s.source_kind as "sourceKind", s.source_id as "sourceId", s.company_id as "companyId", s.issue_id as "issueId", + s.activity_at::text as "activityAt", ceil(extract(epoch from s.activity_at) * 1000 + ${DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000}) 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' + ${issueIds.length ? sql`and s.issue_id in (${sql.join(issueIds.map(value => sql`${value}`), sql`, `)})` : sql``} + and (${scope.companyId ?? null}::uuid is null or s.company_id = ${scope.companyId ?? null}::uuid) + and (${id}::text is null or s.source_id = ${id}) + and (s.source_kind = 'blocker_attention' or ((r.id is null or r.keep = false) and r.archived_at is null)) + ${cursor ? sql`and (s.source_kind, s.source_id) > (${cursor.sourceKind}, ${cursor.sourceId})` : sql``} + order by s.source_kind, s.source_id limit 100`) + .then(rows => (rows as unknown as Array>).map(row => ({ ...row, deadlineAt: Number(row.deadlineAt) }))); +} + +(support.supported ? describe : describe.skip)("attention retention candidates", () => { + let temporary: Awaited> | undefined; + let db: Db; + const companyId = randomUUID(), otherCompanyId = randomUUID(), inactiveCompanyId = randomUUID(); + const fixture: Record = {}; + + async function seed(company: string, prefix: string, status: string, size: number) { + await db.execute(sql`insert into companies (id, name, issue_prefix, status) values (${company}::uuid, ${prefix}, ${prefix}, ${status})`); + await db.execute(sql`insert into agents (company_id, name, status) + select ${company}::uuid, 'Agent ' || n, case when n % 3 = 0 then 'error' else 'idle' end from generate_series(1, 6) n`); + // Issues: blocked, in review, hidden, harness-owned and ordinary ones. + await db.execute(sql`insert into issues (company_id, title, status, hidden_at, harness_kind, updated_at) + select ${company}::uuid, 'Issue ' || n, + case n % 5 when 0 then 'blocked' when 1 then 'in_review' when 2 then 'blocked' else 'todo' end, + case when n % 11 = 0 then now() end, case when n % 13 = 0 then 'eval' end, now() - n * interval '1 minute' + from generate_series(1, ${size}::int) n`); + const agent = sql`(select id from agents where company_id = ${company}::uuid order by name limit 1)`; + await db.execute(sql`insert into heartbeat_runs (company_id, agent_id, status, finished_at, context_snapshot) + select ${company}::uuid, ${agent}, case when n % 4 = 0 then 'timed_out' when n % 2 = 0 then 'failed' else 'succeeded' end, + case when n % 3 = 0 then now() end, + case n % 3 when 0 then jsonb_build_object('issueId', (select id::text from issues where company_id = ${company}::uuid order by title limit 1 offset n % 5)) + when 1 then jsonb_build_object('taskId', 'task-' || n) else '{}'::jsonb end + from generate_series(1, ${size}::int) n`); + // A lifecycle receipt on most runs, an output decoy and a non-matching lifecycle event on others. + await db.execute(sql`insert into heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, message) + select r.company_id, r.id, r.agent_id, 1, + case when row_number() over (order by r.id) % 5 = 0 then 'stdout' else 'lifecycle' end, + case when row_number() over (order by r.id) % 7 = 0 then 'Run step' else 'Bounded retry exhausted after 3 attempts' end + from heartbeat_runs r where r.company_id = ${company}::uuid`); + await db.execute(sql`insert into decisions (company_id, origin_agent_id, origin_issue_id, origin_run_id, title, body, options, status, expires_at, signed_spec, target_snapshots) + select r.company_id, r.agent_id, (select id from issues where company_id = r.company_id order by title limit 1), r.id, 'Decision', 'Choose', '[]'::jsonb, + case when row_number() over (order by r.id) % 3 = 0 then 'resolved' else 'open' end, now() + interval '1 day', 'fixture', '{}'::jsonb + from heartbeat_runs r where r.company_id = ${company}::uuid`); + await db.execute(sql`insert into issue_thread_interactions (company_id, issue_id, kind, status, addressee_user_id, payload) + select i.company_id, i.id, 'ask_user_questions', case when row_number() over (order by i.id) % 4 = 0 then 'answered' else 'pending' end, + case when row_number() over (order by i.id) % 5 = 0 then 'board' end, '{"version":1,"questions":[]}'::jsonb + from issues i where i.company_id = ${company}::uuid`); + await db.execute(sql`insert into issue_recovery_actions (company_id, source_issue_id, kind, status, owner_type, cause, fingerprint, next_action) + select i.company_id, i.id, 'execution', case when n % 4 = 0 then 'resolved' when n % 4 = 1 then 'escalated' else 'active' end, + case when n % 3 = 0 then 'agent' when n % 3 = 1 then 'user' else 'board' end, 'fixture', 'fp-' || n, 'Choose' + from (select company_id, id, row_number() over (order by id) as n from issues where company_id = ${company}::uuid) i`); + const [policy] = await db.execute<{ id: string }>(sql`insert into budget_policies (company_id, scope_type, scope_id, window_kind) + values (${company}::uuid, 'company', ${company}::uuid, 'monthly') returning id::text`); + await db.execute(sql`insert into budget_incidents (company_id, policy_id, scope_type, scope_id, metric, window_kind, window_start, window_end, threshold_type, amount_limit, amount_observed, status) + select ${company}::uuid, ${policy.id}::uuid, 'company', ${company}::uuid, 'billed_cents', 'monthly', now() - w * interval '1 day', now(), t, l, o, s + from (values (1, 'hard', 100, 1, 'open'), (2, 'soft', 100, 85, 'open'), (3, 'soft', 100, 84.99, 'open'), (4, 'soft', 0, 99, 'open'), (5, 'hard', 100, 100, 'resolved')) v(w, t, l, o, s)`); + // Retention: kept, archived and plain rows across kinds, including blocked issues. + await db.execute(sql`insert into decision_retention (company_id, source_kind, source_id, source_activity_at, keep, archived_at) + select company_id, kind, id, now(), n % 3 = 0, case when n % 3 = 1 then now() end from ( + select company_id, case when status = 'blocked' then 'blocker_attention' else 'review' end as kind, id::text as id, row_number() over (order by id) as n + from issues where company_id = ${company}::uuid and status in ('blocked', 'in_review') + union all select company_id, 'failed_run', id::text, row_number() over (order by id) from heartbeat_runs where company_id = ${company}::uuid + union all select company_id, 'decision', id::text, row_number() over (order by id) from decisions where company_id = ${company}::uuid + union all select company_id, 'agent_error_alert', id::text, row_number() over (order by id) from agents where company_id = ${company}::uuid + ) x where n % 2 = 0`); + } + + beforeAll(async () => { + if (externalUrl) await applyPendingMigrations(externalUrl); + else temporary = await startEmbeddedPostgresTestDatabase("attention-retention-candidates-"); + db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 2, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + await seed(companyId, `C${companyId.slice(0, 5)}`, "active", 90); + await seed(otherCompanyId, `O${otherCompanyId.slice(0, 5)}`, "active", 40); + await seed(inactiveCompanyId, `I${inactiveCompanyId.slice(0, 5)}`, "archived", 20); + const [run] = await db.execute<{ id: string }>(sql`select r.id::text from heartbeat_runs r where r.company_id = ${companyId}::uuid + and r.status = 'failed' and exists (select 1 from heartbeat_run_events e where e.run_id = r.id and e.event_type = 'lifecycle' + and e.message like 'Bounded retry exhausted%') and not exists (select 1 from decision_retention d where d.source_id = r.id::text) + order by r.id limit 1`); + fixture.failedRun = run.id; + const blocked = await db.execute<{ id: string }>(sql`select id::text from issues where company_id = ${companyId}::uuid + and status = 'blocked' and hidden_at is null and harness_kind is null order by id limit 2`); + fixture.blocked = blocked[0].id; fixture.blocked2 = blocked[1].id; + }, 60_000); + afterAll(async () => { + // An external test database is not reset between runs; remove this suite's fixtures. + if (externalUrl && db) await db.transaction(async tx => { + await tx.execute(sql`set local client_min_messages = warning`); + await tx.execute(sql`truncate companies cascade`); + }); + await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); + }); + + async function pages(read: (cursor?: Cursor) => Promise>>) { + const pages: Array>> = []; + let cursor: Cursor | undefined; + for (;;) { + const page = await read(cursor); + pages.push(page); + if (page.length < 100) return pages; + const last = page[page.length - 1]!; + cursor = { sourceKind: last.sourceKind as string, sourceId: last.sourceId as string }; + } + } + + it("orders source kinds the same way as the database", async () => { + const rows = await db.execute<{ kind: string }>(sql`select kind from unnest(array[${sql.join( + [...ATTENTION_RETENTION_KIND_ORDER].reverse().map(kind => sql`${kind}::text`), sql`, `)}]) kind order by kind`); + expect(rows.map(row => row.kind)).toEqual([...ATTENTION_RETENTION_KIND_ORDER]); + }); + + it.each([ + ["one company", { companyId }], + ["every company", {}], + ["one kind", { companyId, sourceKind: "failed_run" as const }], + ["the blocker kind, which ignores retention", { companyId, sourceKind: "blocker_attention" as const }], + ["an unknown kind", { companyId, sourceKind: "approval" as const }], + ])("returns the former pages, rows and cursors for %s", async (_name, scope: AttentionRetentionScope) => { + const expected = await pages(cursor => oracle(db, scope, cursor)); + const actual = await pages(cursor => listAttentionRetentionCandidates(db as never, scope, cursor as never)); + expect(actual).toEqual(expected); + if (!scope.sourceKind?.startsWith("approval")) expect(expected.flat().length).toBeGreaterThan(0); + }); + + it("pages across kind boundaries without skipping or repeating a candidate", async () => { + const all = (await pages(cursor => listAttentionRetentionCandidates(db as never, {}, cursor as never))).flat(); + expect(all.length).toBeGreaterThan(200); + expect(new Set(all.map(row => `${row.sourceKind}:${row.sourceId}`)).size).toBe(all.length); + expect(new Set(all.map(row => row.sourceKind))).toEqual(new Set(ATTENTION_RETENTION_KIND_ORDER)); + expect(all.some(row => row.companyId === inactiveCompanyId)).toBe(false); + // Start a page at every kind boundary and in the middle of a kind. + for (const cursor of [{ sourceKind: "blocker_attention", sourceId: "" }, { sourceKind: "decision", sourceId: "8" }, + { sourceKind: "failed_run", sourceId: "ffffffff" }, { sourceKind: "zzz", sourceId: "" }, { sourceKind: "a", sourceId: "z" }]) { + expect(await listAttentionRetentionCandidates(db as never, { companyId }, cursor as never)).toEqual(await oracle(db, { companyId }, cursor)); + } + }); + + it("filters by issue and leaves company-wide kinds out of an issue scope", async () => { + for (const scope of [{ companyId, issueId: fixture.blocked }, { companyId, issueIds: [fixture.blocked, fixture.blocked2, "not-a-uuid"] }, + { companyId, issueIds: [fixture.blocked.toUpperCase()] }, { issueIds: [] }] as AttentionRetentionScope[]) { + const actual = await listAttentionRetentionCandidates(db as never, scope); + expect(actual).toEqual(await oracle(db, scope)); + if (scope.issueIds?.length || scope.issueId) { + expect(actual.some(row => ["budget_alert", "agent_error_alert"].includes(row.sourceKind))).toBe(false); + } + } + expect((await listAttentionRetentionCandidates(db as never, { companyId, issueId: fixture.blocked })).length).toBeGreaterThan(0); + }); + + it("matches one source only by its exact text id", async () => { + const run = fixture.failedRun; + const exact = await listAttentionRetentionCandidates(db as never, { companyId, sourceKind: "failed_run" }, undefined, run); + expect(exact.map(row => row.sourceId)).toEqual([run]); + expect(exact).toEqual(await oracle(db, { companyId, sourceKind: "failed_run" }, undefined, run)); + expect(await listAttentionRetentionCandidates(db as never, { companyId, sourceId: run })).toEqual(exact); + // Text that casts to the same uuid but is not its printed form never matched. + for (const other of [run.toUpperCase(), `{${run}}`, run.replaceAll("-", ""), "", "not-a-uuid"]) { + expect(await listAttentionRetentionCandidates(db as never, { companyId, sourceKind: "failed_run" }, undefined, other)).toEqual([]); + expect(await oracle(db, { companyId, sourceKind: "failed_run" }, undefined, other)).toEqual([]); + } + // Company isolation: the same id in another company scope is not a candidate. + expect(await listAttentionRetentionCandidates(db as never, { companyId: otherCompanyId, sourceKind: "failed_run" }, undefined, run)).toEqual([]); + // A decoded scope can carry null; it means no source filter. + const nullScope = { companyId, sourceKind: "failed_run", sourceId: null } as unknown as AttentionRetentionScope; + expect(await listAttentionRetentionCandidates(db as never, nullScope)).toEqual(await oracle(db, { companyId, sourceKind: "failed_run" })); + }); + + it("excludes kept and archived sources and failed runs without a retry receipt", async () => { + const rows = (await pages(cursor => listAttentionRetentionCandidates(db as never, { companyId }, cursor as never))).flat(); + const retained = await db.execute<{ kind: string; id: string }>(sql`select source_kind as kind, source_id as id from decision_retention + where company_id = ${companyId}::uuid and (keep or archived_at is not null) and source_kind <> 'blocker_attention'`); + const returned = new Set(rows.map(row => `${row.sourceKind}:${row.sourceId}`)); + expect(retained.length).toBeGreaterThan(0); + for (const row of retained) expect(returned.has(`${row.kind}:${row.id}`)).toBe(false); + const keptBlocked = await db.execute<{ id: string }>(sql`select r.source_id as id from decision_retention r join issues i on i.id::text = r.source_id + where r.company_id = ${companyId}::uuid and r.source_kind = 'blocker_attention' and r.keep and i.hidden_at is null and i.harness_kind is null`); + expect(keptBlocked.length).toBeGreaterThan(0); + for (const row of keptBlocked) expect(returned.has(`blocker_attention:${row.id}`)).toBe(true); + const unreceipted = await db.execute<{ id: string }>(sql`select r.id::text from heartbeat_runs r where r.company_id = ${companyId}::uuid + and r.status in ('failed', 'timed_out') and not exists (select 1 from heartbeat_run_events e where e.run_id = r.id + and e.event_type = 'lifecycle' and e.message like 'Bounded retry exhausted%')`); + expect(unreceipted.length).toBeGreaterThan(0); + for (const row of unreceipted) expect(returned.has(`failed_run:${row.id}`)).toBe(false); + }); +}); diff --git a/server/src/services/attention-retention.ts b/server/src/services/attention-retention.ts index 378302faca3..568247e329f 100644 --- a/server/src/services/attention-retention.ts +++ b/server/src/services/attention-retention.ts @@ -42,48 +42,89 @@ const sourceTables: Record = { failed_run: sql`heartbeat_runs`, budget_alert: sql`budget_incidents`, agent_error_alert: sql`agents`, }; -const sources: Record = { - 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, 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, 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, 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, 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), 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, 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, null::text from agents a where a.status = ${ATTENTION_AGENT_ERROR_STATUS}`, +// Each source is one table read. `id` is the source row's uuid primary key and +// `source_id` is its text form. `where` holds the source's own eligibility. +type Source = { from: SQL; id: SQL; companyId: SQL; activityAt: SQL; issueId: SQL; where: SQL }; +const sources: Record = { + decision: { from: sql`decisions d`, id: sql`d.id`, companyId: sql`d.company_id`, activityAt: sql`d.updated_at`, + issueId: sql`d.origin_issue_id::text`, where: sql`d.status = 'open'` }, + issue_thread_interaction: { from: sql`issue_thread_interactions i`, id: sql`i.id`, companyId: sql`i.company_id`, + activityAt: sql`i.updated_at`, issueId: sql`i.issue_id::text`, where: sql`i.status = 'pending' and i.addressee_user_id is null` }, + recovery_action: { from: sql`issue_recovery_actions r`, id: sql`r.id`, companyId: sql`r.company_id`, activityAt: sql`r.updated_at`, + issueId: sql`r.source_issue_id::text`, + where: sql`r.status in ('active', 'escalated') and r.owner_type in (${sql.join(HUMAN_RECOVERY_OWNER_TYPES.map(owner => sql`${owner}`), sql`, `)})` }, + blocker_attention: { from: sql`issues i`, id: sql`i.id`, companyId: sql`i.company_id`, activityAt: sql`i.updated_at`, + issueId: sql`i.id::text`, where: sql`i.status = 'blocked' and i.hidden_at is null and i.harness_kind is null` }, + review: { from: sql`issues i`, id: sql`i.id`, companyId: sql`i.company_id`, activityAt: sql`i.updated_at`, + issueId: sql`i.id::text`, where: sql`i.status = 'in_review' and i.hidden_at is null and i.harness_kind is null` }, + // Keep the receipt filters as SQL literals. They must imply the predicate of + // the partial index heartbeat_run_events_retry_exhausted_idx; a bound + // parameter stops a generic prepared plan from using that index. + failed_run: { from: sql`heartbeat_runs r`, id: sql`r.id`, companyId: sql`r.company_id`, + activityAt: sql`coalesce(r.finished_at, r.updated_at, r.created_at)`, + issueId: sql`coalesce(r.context_snapshot->>'issueId', r.context_snapshot->>'taskId')`, + where: sql`r.status in ('failed', 'timed_out') and exists ( + select 1 from heartbeat_run_events e where e.company_id = r.company_id and e.run_id = r.id + and e.event_type = 'lifecycle' and e.message like ${sql.raw(`'${RETRY_EXHAUSTION_MESSAGE_PREFIX}%'`)})` }, + budget_alert: { from: sql`budget_incidents b`, id: sql`b.id`, companyId: sql`b.company_id`, activityAt: sql`b.updated_at`, + issueId: sql`null::text`, where: sql`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: { from: sql`agents a`, id: sql`a.id`, companyId: sql`a.company_id`, activityAt: sql`a.updated_at`, + issueId: sql`null::text`, where: sql`a.status = ${ATTENTION_AGENT_ERROR_STATUS}` }, }; +const CANDIDATE_PAGE_SIZE = 100; +// PostgreSQL prints a uuid only in this form, so no other text equals a source_id. +const CANONICAL_UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; +// The keyset orders by (source_kind, source_id). Each kind name differs from the +// others before its first underscore, so this code-unit order is also the order +// under linguistic collations. A test pins it against the database. +export const ATTENTION_RETENTION_KIND_ORDER: readonly Kind[] = [...kinds].sort(); + +/** One page of retention candidates in (source_kind, source_id) order. + * Each kind is read in its own branch with every filter pushed into it and is + * limited to one page, so a page reads at most one page per kind instead of + * every candidate of every kind in every company. */ +function candidateQuery(scope: Scope, cursor: Cursor | undefined, sourceId: string | undefined, issueIds: string[]): SQL | null { + const selected = ATTENTION_RETENTION_KIND_ORDER.filter(kind => (!scope.sourceKind || scope.sourceKind === kind) + && (!issueIds.length || !["budget_alert", "agent_error_alert"].includes(kind)) + && (!cursor || kind >= cursor.sourceKind)); + if (!selected.length) return null; + if (sourceId !== undefined && !CANONICAL_UUID.test(sourceId)) return null; + const branches = selected.map(kind => { + const source = sources[kind]; + const conditions: SQL[] = [source.where, sql`company.status = 'active'`]; + if (scope.companyId != null) conditions.push(sql`${source.companyId} = ${scope.companyId}::uuid`); + if (sourceId !== undefined) conditions.push(sql`${source.id} = ${sourceId}::uuid`); + if (issueIds.length) conditions.push(sql`${source.issueId} in (${sql.join(issueIds.map(id => sql`${id}`), sql`, `)})`); + // Blocker attention re-reads retention for its terminal source in eligible(). + if (kind !== "blocker_attention") conditions.push(sql`(retention.id is null or retention.keep = false) and retention.archived_at is null`); + if (cursor?.sourceKind === kind) conditions.push(sql`${source.id}::text > ${cursor.sourceId}`); + const retention = kind === "blocker_attention" ? sql`` : sql`left join decision_retention retention on retention.company_id = ${source.companyId} + and retention.source_kind = ${kind} and retention.source_id = ${source.id}::text`; + return sql`(select ${kind}::text as source_kind, ${source.id}::text as source_id, ${source.companyId} as company_id, + ${source.activityAt} as activity_at, ${source.issueId} as issue_id + from ${source.from} inner join companies company on company.id = ${source.companyId} ${retention} + where ${sql.join(conditions, sql` and `)} + order by ${source.id}::text limit ${CANDIDATE_PAGE_SIZE})`; + }); + return sql`with source_candidates(source_kind, source_id, company_id, activity_at, issue_id) as (${sql.join(branches, sql` union all `)}) + select s.source_kind as "sourceKind", s.source_id as "sourceId", s.company_id as "companyId", + 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 order by s.source_kind, s.source_id limit ${CANDIDATE_PAGE_SIZE}`; +} + +export async function listAttentionRetentionCandidates(database: Db, scope: Scope = {}, cursor?: Cursor, sourceId?: string): Promise { + const issueIds = scope.issueIds ?? (scope.issueId ? [scope.issueId] : []); + const query = candidateQuery(scope, cursor, sourceId ?? scope.sourceId ?? undefined, issueIds); + if (!query) return []; + const rows = await database.execute(query) as unknown as Candidate[]; + return rows.map(row => ({ ...row, deadlineAt: Number(row.deadlineAt) })); +} + export function attentionRetentionService(db: Db) { const cursors = new Map(); - async function candidates(database: Db, scope: Scope = {}, cursor?: Cursor, sourceId?: string): Promise { - const issueIds = scope.issueIds ?? (scope.issueId ? [scope.issueId] : []); - const selected = kinds.filter(kind => (!scope.sourceKind || scope.sourceKind === kind) - && (!issueIds.length || !["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, 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.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' - ${issueIds.length ? sql`and s.issue_id in (${sql.join(issueIds.map(id => sql`${id}`), sql`, `)})` : sql``} - 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)) - ${cursor ? sql`and (s.source_kind, s.source_id) > (${cursor.sourceKind}, ${cursor.sourceId})` : sql``} - order by s.source_kind, s.source_id limit 100`) as unknown as Candidate[]; - return rows.map(row => ({ ...row, deadlineAt: Number(row.deadlineAt) })); - } + const candidates = listAttentionRetentionCandidates; async function visibleIssue(database: Db, companyId: string, issueId: string, conversations = false) { const [row] = await database.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.id, issueId), @@ -283,11 +324,11 @@ export function attentionRetentionService(db: Db) { if (entry) entries.push(entry); } } - if (rows.length === 100) { + if (rows.length === CANDIDATE_PAGE_SIZE) { const last = rows[rows.length - 1]!; cursors.set(key, { sourceKind: last.sourceKind, sourceId: last.sourceId }); } else cursors.delete(key); - return { rows: entries, complete: rows.length < 100 }; + return { rows: entries, complete: rows.length < CANDIDATE_PAGE_SIZE }; } async function check(entry: Entry) {