From 326df52673b491ac344128a2dab487f9167bc1e0 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:17:03 -0600 Subject: [PATCH] perf(api): rank Mine inbox candidates from the participation set (#20413) 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 Mine inbox reads `GET /api/companies/:companyId/inbox/mine`. Its main query is `issueService.list` with "touched by me", "not archived by me", six statuses including `done`, and `LIMIT 100` by canonical last activity. > - That query evaluates `touchedByUserCondition` (a five-way OR of EXISTS) on every status-matching issue in the company. Because `done` is included, the scan grows with company history, not with the user's work. > - The unread badge already solved the same problem with a UNION of indexed participation sources (#17973). The list path did not. > - This pull request makes the Mine list rank page ids from that same participation UNION, then load full rows (with the expensive select-list subqueries) only for the 100 selected ids. > - The benefit is that the main Mine query scales with the user's participation instead of the company's issue count. Results stay identical, and a new test proves it against the original scan. ## Linked Issues or Issue Description **What existing behavior does this improve?** The latency of the Mine inbox snapshot, `GET /api/companies/:companyId/inbox/mine`. **Subsystem affected** Server: issue list service (`server/src/services/issues.ts`) and the Mine route. **Current behavior** Measured on 2026-11-09: 280–570 ms server time on a dev instance and 1.9–3.2 s on staging. The list query scans every status-matching issue (28k of 30k in the fixture below) to test participation, then sorts the touched ones. **Proposed behavior** The list query reads only the user's participation set and computes the row projection only for the page. **Reason and benefit** Inbox opens are frequent. The old cost grows with every completed task in the company. Refs #17973 (same UNION for the unread badge). ## What Changed - Extracted the participation UNION from `unreadInboxIssueIds` into `mineParticipationIssueIds()`. The unread badge uses it unchanged. - New `IssueFilters.touchedCandidates: "participation"`. When it is set with `touchedByUserId` (and there is no search and no `oneChildPerParent`), `list`: 1. replaces `touchedByUserCondition` with `issues.id IN ()`. Each UNION member is one OR branch of the old condition, so the set is the same; 2. runs a narrow `candidates` query (`SELECT id … WHERE ORDER BY LIMIT/OFFSET`); 3. loads the full projection for only those ids (re-checking company, visibility and the read condition), and restores the phase 1 order. - Only the Mine route sets the flag. Every other `svc.list` caller keeps the old SQL. - A new Server-Timing phase, `inbox_issues_candidates`. - `server/src/__tests__/inbox-mine-candidates.test.ts`: a result-equivalence test. - `scripts/benchmark-inbox-mine.ts`: a local benchmark that seeds a large fixture, compares full payloads, and records `EXPLAIN (ANALYZE, BUFFERS)` per query. ### Not changed, with evidence - **Recomputed comment/activity maxima** (`lastActivityStatsForIssues`): 0.9 + 0.2 ms per call in the plans. Not worth a second code path. - **Touched condition inside `userCommentStatsForIssues`**: 1.4–2.1 ms. Not changed. - **`liveDescendantCountMapForIssues`**: 12 ms at 30k issues and 40 ms at 90k. It now costs about as much as the list query. It scans all live runs and hashes all company issues. It is company-wide work, outside the candidate selection, so I left it for a separate change. ## Verification ### Correctness `inbox-mine-candidates.test.ts` runs the old scan and the new path on the same data. It asserts equal ordered ids and equal full JSON payloads. Fixture: 170+ touched issues (more than one page, with tied timestamps) plus these edge cases: - included: created by me, assigned to me only, user activity, my direct comment, `done`, pending request addressed to me, archived then re-activated by a newer outside comment, read state with a later agent comment, a live child (live descendant count); - excluded: archived with no newer activity, mention-only, read audit only (`issue.read_marked`), uppercase entity id, channel-sourced comment, agent-created with my user id, `cancelled`, hidden, harness, plugin operation, private to another user (`issueReadSqlCondition`), answered request, untouched issues, another company. It also compares other shapes: no archive filter, `status=done` with the full projection, `offset`/`limit`, ascending order, unrestricted read condition, a second user, and a user with no participation. Mutation check: removing the comment source from the UNION makes the test fail. The benchmark also asserts identical full payloads for every sample of both arms. Focused suites (local): `inbox-mine-candidates`, `inbox-mine-routes`, `authorization-privacy-query`, `issues-unread-inbox`, `inbox-archive-routes`, `issues-assignee-unread`, `issues-creation-visibility` — all pass. ### Performance (local SQL measurements) Command: `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-inbox-mine.ts --baseline-ref=18c3f4ef7b --issues= [--touch-density=]`. Baseline = 18c3f4ef7b, the #20408 head with the same timing hooks. Embedded PostgreSQL 18.1, `shared_buffers` 128MB, `work_mem` 4MB, JIT on (not triggered), pool of 8. Alternating order, 2 warmups. Fixture per run: 12 activity rows and 6 comments per issue, 55% `done`, archives, read states, pending requests, 150 live runs. | Fixture | Arm | `svc.list` median (min–max) | List query phase | Main query plan | | --- | --- | --- | --- | --- | | 30k issues, user touched ~1.1k | before | 47.8 ms (41–66) | 30.0 ms | 28.1 ms exec, 33.7k buffers | | | after | 39.4 ms (36–42) | 19.4 + 4.2 ms | 16.0 ms exec, 22.1k buffers (+1.8 ms rows) | | 90k issues, user touched ~3.4k | before | 124.4 ms (113–180) | 71.1 ms | 70.8 ms exec, 103.9k buffers | | | after | 102.9 ms (88–118) | 43.1 + 3.8 ms | 41.7 ms exec, 72.7k buffers | | 30k issues, user touched ~11k (dense) | before | 135.0 ms | 119.2 ms | 112.6 ms exec, 128k buffers | | | after | 136.6 ms | 108.7 + 4.9 ms | 101.9 ms exec, 144k buffers | - The old plan scanned 27,993 status-matching issues to test participation. The new plan reads the participation UNION (1,155 ids at 30k; 3,448 at 90k) and probes `issues_pkey`. - **Neutral case:** when a user has touched most issues (dense row), the UNION does not shrink the set. The per-candidate privacy check, archive visibility and activity maxima dominate. I report this as neutral, not as a gain. - On a loaded machine (a parallel test run), the same 30k fixture took 129 ms before and 55 ms after. Planning time of the old query rose to 27 ms there. I report the quiet-machine numbers above as the result. - I did not get query evidence from staging's database. This session had no Neon or CloudOps access (no `neonctl`, no stack credentials), so all numbers are local. #20408 adds the per-phase header, which can confirm the effect on staging after deploy. ### Pre-PR checks - `tsc --noEmit -p server`: no errors in the changed files. - `pnpm verify --explain` selects 2,417 of 3,204 test files for this change, because `services/issues.ts` is imported almost everywhere. I did not repeat that full suite locally (AGENTS.md §7); required CI supplies it. ## Risks - **Equivalence depends on the UNION matching `touchedByUserCondition`.** Each branch has a UNION member: human-created by me, assigned to me, pending request addressed to me, my participation activity (canonical lowercase UUID ids only, the same as the old `= issues.id::text` join), and my direct human comments. The equivalence test pins these. A future change to `touchedByUserCondition` must also change the UNION. Both live next to each other, and the unread badge already depends on the same equality. - **Two reads instead of one.** A row can change between the candidate query and the row query. The row query re-checks company, visibility and read access, so a row that became unreadable is dropped. The page can then have fewer than 100 rows for that one response. - Scope: only the Mine route sets the flag. No migration. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits, embedded PostgreSQL benchmarks). Standard context window; extended thinking not used. ## 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 --- scripts/benchmark-inbox-mine.ts | 223 ++++++++++++++++++ .../__tests__/inbox-mine-candidates.test.ts | 139 +++++++++++ .../src/__tests__/inbox-mine-routes.test.ts | 5 +- server/src/routes/issues.ts | 2 +- server/src/services/issues.ts | 130 ++++++---- 5 files changed, 452 insertions(+), 47 deletions(-) create mode 100644 scripts/benchmark-inbox-mine.ts create mode 100644 server/src/__tests__/inbox-mine-candidates.test.ts diff --git a/scripts/benchmark-inbox-mine.ts b/scripts/benchmark-inbox-mine.ts new file mode 100644 index 00000000000..9939d8b1f83 --- /dev/null +++ b/scripts/benchmark-inbox-mine.ts @@ -0,0 +1,223 @@ +// Local SQL benchmark for the Mine inbox list (GET /api/companies/:id/inbox/mine). +// +// pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-inbox-mine.ts \ +// --baseline-ref= [--issues=30000] [--activity-per-issue=12] +// [--comments-per-issue=6] [--touch-density=1] [--samples=7] [--evidence-dir=] +// +// Seeds one disposable embedded-PostgreSQL company with tens of thousands of +// issues (most of them `done`), comments, activity, archives, read states, +// pending interactions and live runs, plus a user who touched a few hundred. +// It then runs the baseline (from --baseline-ref) and the working-tree +// `issueService.list` with the exact Mine filters, asserts the complete row +// payloads are identical, and reports median timings and per-query +// EXPLAIN (ANALYZE, BUFFERS) ranked by execution time. +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 path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { createDb, applyPendingMigrations, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; +import { INBOX_MINE_ISSUE_STATUS_FILTER } from "../packages/shared/src/index.js"; +import { issueService, type IssueFilters } from "../server/src/services/issues.js"; +import { issueReadSqlCondition } from "../server/src/services/authorization.js"; + +const repositoryRoot = fileURLToPath(new URL("../", import.meta.url)); +const git = (args: string[]) => execFileSync("git", args, { cwd: repositoryRoot, encoding: "utf8" }); +const arg = (name: string) => process.argv.find((value) => value.startsWith(`--${name}=`))?.slice(name.length + 3); +const baselineRef = arg("baseline-ref"); +assert.ok(baselineRef, "Pass a pinned pre-fix --baseline-ref="); +const baselineRevision = git(["rev-parse", `${baselineRef}^{commit}`]).trim(); +const issueCount = Number(arg("issues") ?? 30000); +const samples = Number(arg("samples") ?? 7); +const activityPerIssue = Number(arg("activity-per-issue") ?? 12); +const commentsPerIssue = Number(arg("comments-per-issue") ?? 6); +// Participation density: 1 touches ~1.4k of 30k issues; 10 touches ~11k. +const density = Number(arg("touch-density") ?? 1); +const every = (modulus: number) => Math.max(2, Math.round(modulus / density)); +const evidenceDir = arg("evidence-dir"); + +// Baseline modules: the Mine list SQL lives in these two files. Every other +// dependency is held constant at the working tree. +const modulePaths = ["server/src/services/issues.ts", "server/src/services/issue-participation.ts"]; +const baselineDir = await fs.mkdtemp(path.join(repositoryRoot, "server/scripts/.inbox-mine-baseline-")); +for (const modulePath of modulePaths) { + const source = git(["show", `${baselineRevision}:${modulePath}`]).replace(/from "(\.{1,2}\/[^"]+)"/g, (_m, specifier: string) => { + const resolved = fileURLToPath(new URL(specifier, pathToFileURL(path.join(repositoryRoot, modulePath)))); + return `from ${JSON.stringify(resolved.endsWith("/services/issue-participation.js") ? path.join(baselineDir, "issue-participation.ts") : resolved)}`; + }); + await fs.writeFile(path.join(baselineDir, path.basename(modulePath)), source); +} +const baseline = await import(pathToFileURL(path.join(baselineDir, "issues.ts")).href) as typeof import("../server/src/services/issues.js"); + +type Captured = { sql: string; params: unknown[] }; +let database: Awaited> | undefined; +let db: Db | undefined; +const keepAlive = setInterval(() => {}, 1000); +try { + database = await startEmbeddedPostgresTestDatabase("inbox-mine-benchmark-"); + await applyPendingMigrations(database.connectionString); + // A small pool, like the server's: the list issues enrichment reads in parallel. + db = createDb(database.connectionString, { maxConnections: 8 }); + const client = db.$client as any; + let capture: Captured[] | null = null; + const unsafe = client.unsafe.bind(client); + client.unsafe = (query: string, params?: unknown[], options?: unknown) => { + if (capture && /^\s*(select|with)\b/i.test(query)) capture.push({ sql: query, params: params ?? [] }); + return unsafe(query, params, options); + }; + await client.unsafe("set client_min_messages = warning"); + const [settings] = await client.unsafe("select version(), current_setting('shared_buffers') shared_buffers, current_setting('work_mem') work_mem, current_setting('jit') jit"); + + const companyId = randomUUID(), userId = "synthetic-mine-user", otherUserId = "synthetic-other-human"; + const seedStart = performance.now(); + await client.unsafe(`insert into companies(id,name,issue_prefix) values($1,'Synthetic Mine','MINE')`, [companyId]); + const agentIds = [randomUUID(), randomUUID(), randomUUID()]; + for (const agentId of agentIds) await client.unsafe(`insert into agents(id,company_id,name,adapter_type,status) values($1,$2,$3,'process','active')`, [agentId, companyId, `Agent ${agentId.slice(0, 8)}`]); + // h(n, salt) is a deterministic 0..99 bucket so the fixture is reproducible. + await client.unsafe(`create or replace function bench_h(n int, salt text) returns int language sql immutable as $$ select abs(hashtext(n::text || salt)) % 100 $$`); + await client.unsafe(`insert into issues(id,company_id,title,status,created_by_user_id,created_by_agent_id,assignee_user_id,origin_kind,hidden_at,harness_kind,completed_at,created_at,updated_at) + select gen_random_uuid(),$1::uuid,'Synthetic '||n, + case when bench_h(n,'s')<55 then 'done' when bench_h(n,'s')<60 then 'cancelled' when bench_h(n,'s')<70 then 'backlog' + when bench_h(n,'s')<82 then 'todo' when bench_h(n,'s')<90 then 'in_progress' when bench_h(n,'s')<95 then 'in_review' else 'blocked' end, + case when n%${every(75)}=0 then $2 when bench_h(n,'c')<30 then $3 else null end, + case when n%${every(75)}<>0 and bench_h(n,'c')>=30 then ($4::uuid[])[1+n%3] else null end, + case when n%${every(311)}=0 then $2 else null end, + case when n%997=0 then 'plugin:demo:operation' else 'manual' end, + case when n%1499=0 then '2026-01-01'::timestamptz else null end, + case when n%1777=0 then 'synthetic' else null end, + case when bench_h(n,'s')<55 then '2026-01-01'::timestamptz+n*interval '1 second' else null end, + '2026-01-01'::timestamptz,'2026-01-01'::timestamptz+n*interval '1 second' + from generate_series(1,$5::int) n`, [companyId, userId, otherUserId, agentIds, issueCount]); + await client.unsafe(`create table bench_fixture_issues as select id, row_number() over (order by updated_at)::int n, status from issues where company_id=$1`, [companyId]); + await client.unsafe(`update issues i set parent_id=p.id from bench_fixture_issues c join bench_fixture_issues p on p.n=c.n/2 where i.id=c.id and c.n>100 and bench_h(c.n,'p')<35`); + // Agent comments on every issue, an outside human on some, and the user's own direct comments. + await client.unsafe(`insert into issue_comments(company_id,issue_id,author_agent_id,body,created_at) + select $1::uuid,f.id,($2::uuid[])[1+k%3],'Agent progress '||f.n||'/'||k,'2026-02-01'::timestamptz+(f.n*$3::int+k)*interval '1 second' + from bench_fixture_issues f cross join generate_series(0,$3::int-1) k`, [companyId, agentIds, commentsPerIssue]); + await client.unsafe(`insert into issue_comments(company_id,issue_id,author_user_id,body,created_at) + select $1::uuid,id,$2,'Outside human note','2026-03-01'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%7=0`, [companyId, otherUserId]); + await client.unsafe(`insert into issue_comments(company_id,issue_id,author_user_id,body,created_at) + select $1::uuid,id,$2,'My note','2026-01-15'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%${every(97)}=0`, [companyId, userId]); + // Mention-only: the user is named in a comment but never participated. + await client.unsafe(`insert into issue_comments(company_id,issue_id,author_user_id,body,created_at) + select $1::uuid,id,$2,'@synthetic-mine-user can you look?','2026-03-02'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%113=0`, [companyId, otherUserId]); + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'issue',f.id::text,'agent',($2::uuid[])[1+k%3]::text, + case when k%2=0 then 'issue.updated' else 'issue.comment_added' end, + jsonb_build_object('status',f.status,'_previous',jsonb_build_object('status','todo')), + '2026-01-20'::timestamptz+(f.n*$3::int+k)*interval '1 second' + from bench_fixture_issues f cross join generate_series(0,$3::int-1) k`, [companyId, agentIds, activityPerIssue]); + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'issue',id::text,'user',$2,'issue.updated','{}'::jsonb,'2026-01-25'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%${every(89)}=0`, [companyId, userId]); + // Reads are audited but are not participation. + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'issue',id::text,'user',$2,'issue.read_marked','{}'::jsonb,'2026-01-26'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%13=0`, [companyId, userId]); + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'heartbeat_run',gen_random_uuid()::text,'agent',($2::uuid[])[1],'heartbeat.run.created','{}'::jsonb,'2026-01-01'::timestamptz+n*interval '1 second' from generate_series(1,$3::int) n`, [companyId, agentIds, issueCount * 2]); + // Archives: some stay archived, some are reactivated by later outside comments (n%7=0). + await client.unsafe(`insert into issue_inbox_archives(company_id,issue_id,user_id,archived_at) + select $1::uuid,id,$2,'2026-02-15'::timestamptz from bench_fixture_issues where n%3=0 and (n%${every(75)}=0 or n%${every(89)}=0 or n%${every(97)}=0 or n%${every(311)}=0)`, [companyId, userId]); + await client.unsafe(`insert into issue_read_states(company_id,issue_id,user_id,last_read_at) + select $1::uuid,id,$2,'2026-02-20'::timestamptz from bench_fixture_issues where n%40=0`, [companyId, userId]); + await client.unsafe(`insert into issue_thread_interactions(company_id,issue_id,kind,status,addressee_user_id,payload,created_at) + select $1::uuid,id,'request_confirmation','pending',$2,'{"version":1,"prompt":"Confirm?"}'::jsonb,'2026-03-05'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%503=0`, [companyId, userId]); + await client.unsafe(`insert into heartbeat_runs(id,company_id,agent_id,status,context_snapshot) + select gen_random_uuid(),$1::uuid,($2::uuid[])[1+n%3],case when n%2=0 then 'running' else 'queued' end, + jsonb_build_object('issueId',(select id from bench_fixture_issues where bench_fixture_issues.n=g.n*97)) + from generate_series(1,150) g(n)`, [companyId, agentIds]); + await client.unsafe(`insert into heartbeat_runs(company_id,agent_id,status) select $1::uuid,($2::uuid[])[1],'succeeded' from generate_series(1,$3::int)`, [companyId, agentIds, issueCount]); + await client.unsafe("ANALYZE"); + const [counts] = await client.unsafe(`select + (select count(*)::int from issues where company_id=$1) issues, + (select count(*)::int from issues where company_id=$1 and status='done') done, + (select count(*)::int from issue_comments where company_id=$1) comments, + (select count(*)::int from activity_log where company_id=$1) activity, + (select count(*)::int from issue_inbox_archives where company_id=$1) archives, + (select count(*)::int from issue_read_states where company_id=$1) read_states, + (select count(*)::int from issue_thread_interactions where company_id=$1) interactions, + (select count(*)::int from heartbeat_runs where company_id=$1 and status in ('queued','running')) live_runs`, [companyId]); + const fixtureMs = performance.now() - seedStart; + + const readCondition = await issueReadSqlCondition(db, { type: "board", source: "session", userId, authorizingUserId: userId } as never); + const filters: IssueFilters = { + // The route's exact filters. Baselines that predate touchedCandidates ignore it. + readCondition, touchedByUserId: userId, touchedCandidates: "participation", inboxArchivedByUserId: userId, + status: INBOX_MINE_ISSUE_STATUS_FILTER, listProjection: "index", + sortField: "updated", sortDir: "desc", limit: 100, + includeRoutineExecutions: true, includeLiveDescendantSummary: true, + }; + const arms = { baseline: baseline.issueService(db), treatment: issueService(db) }; + const times = { baseline: [] as number[], treatment: [] as number[] }; + let reference: unknown; + for (let iteration = -2; iteration < samples; iteration++) { + for (const arm of (iteration % 2 === 0 ? ["baseline", "treatment"] : ["treatment", "baseline"]) as Array) { + const started = performance.now(); + const rows = await arms[arm].list(companyId, filters); + const elapsed = performance.now() - started; + if (iteration >= 0) times[arm].push(elapsed); + const serialized = JSON.parse(JSON.stringify(rows)); + if (reference === undefined) reference = serialized; + else assert.deepEqual(serialized, reference, `${arm}: complete Mine row payloads must match`); + } + } + const resultIds = (reference as Array<{ id: string }>).map((row) => row.id); + + // Per-phase timing from the list's own phase hook (absent in old baselines). + const phaseTimes: Record> = { baseline: {}, treatment: {} }; + for (let iteration = 0; iteration < samples; iteration++) { + for (const arm of ["baseline", "treatment"] as const) { + await arms[arm].list(companyId, filters, { + timePhase: async (phase, read) => { + const started = performance.now(); + try { return await read(); } finally { (phaseTimes[arm]![phase] ??= []).push(performance.now() - started); } + }, + } as never); + } + } + assert.ok(resultIds.length === 100, `Fixture must fill the 100-row page (got ${resultIds.length})`); + + const plans: Record = {}; + for (const arm of ["baseline", "treatment"] as const) { + capture = []; + await arms[arm].list(companyId, filters); + const captured = capture; + capture = null; + const ranked = []; + for (const query of captured) { + const [row] = await client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params); + const plan = row["QUERY PLAN"][0]; + ranked.push({ executionMs: plan["Execution Time"], planningMs: plan["Planning Time"], + sharedHit: plan.Plan["Shared Hit Blocks"], sharedRead: plan.Plan["Shared Read Blocks"], + sql: query.sql, params: query.params, plan }); + } + ranked.sort((a, b) => b.executionMs - a.executionMs); + plans[arm] = ranked; + } + const stats = (values: number[]) => { + const sorted = [...values].sort((a, b) => a - b); + return { medianMs: sorted[Math.floor(sorted.length / 2)]!, minMs: sorted[0]!, maxMs: sorted[sorted.length - 1]!, samplesMs: values }; + }; + const summary = (arm: "baseline" | "treatment") => (plans[arm] as Array<{ executionMs: number; sql: string; sharedHit: number; sharedRead: number }>) + .map((query) => ({ executionMs: Number(query.executionMs.toFixed(2)), buffers: query.sharedHit + query.sharedRead, sql: query.sql.replace(/\s+/g, " ").slice(0, 160) })); + const report = { + baselineRevision, treatmentRevision: git(["rev-parse", "HEAD"]).trim(), treatmentDirty: git(["status", "--porcelain", "--", ...modulePaths]).trim() !== "", + settings, fixture: { ...counts, fixtureMs, resultRows: resultIds.length }, + baseline: stats(times.baseline), treatment: stats(times.treatment), + phases: Object.fromEntries(Object.entries(phaseTimes).map(([arm, phases]) => [arm, + Object.fromEntries(Object.entries(phases).map(([phase, values]) => [phase, Number(stats(values).medianMs.toFixed(2))]))])), + speedup: stats(times.baseline).medianMs / stats(times.treatment).medianMs, + queries: { baseline: summary("baseline"), treatment: summary("treatment") }, + }; + if (evidenceDir) { + await fs.mkdir(evidenceDir, { recursive: true }); + await fs.writeFile(path.join(evidenceDir, "summary.json"), JSON.stringify(report, null, 2)); + await fs.writeFile(path.join(evidenceDir, "plans.json"), JSON.stringify(plans, null, 2)); + } + console.log(JSON.stringify(report, null, 2)); +} finally { + await db?.$client.end({ timeout: 0 }); + await database?.cleanup(); + await fs.rm(baselineDir, { recursive: true, force: true }); + clearInterval(keepAlive); +} diff --git a/server/src/__tests__/inbox-mine-candidates.test.ts b/server/src/__tests__/inbox-mine-candidates.test.ts new file mode 100644 index 00000000000..6bcc51e48dc --- /dev/null +++ b/server/src/__tests__/inbox-mine-candidates.test.ts @@ -0,0 +1,139 @@ +import { randomUUID } from "node:crypto"; +import { sql } from "drizzle-orm"; +import { + activityLog, agents, companies, heartbeatRuns, issueComments, issueInboxArchives, issueReadStates, issues, issueThreadInteractions, +} from "@paperclipai/db"; +import { INBOX_MINE_ISSUE_STATUS_FILTER } from "@paperclipai/shared"; +import { expect, it } from "vitest"; +import { issueReadSqlCondition } from "../services/authorization.js"; +import { issueService, type IssueFilters } from "../services/issues.js"; +import { describeEmbeddedPostgres, useEmbeddedPostgres } from "./helpers/route-test-harness.js"; + +// The Mine list ranks candidates from the participation UNION +// (`touchedCandidates: "participation"`). These tests run the original +// `touchedByUserCondition` scan beside it and require identical ordered rows. +describeEmbeddedPostgres("Mine inbox participation candidates", () => { + const ctx = useEmbeddedPostgres("inbox-mine-candidates-"); + const at = (minute: number) => new Date(Date.UTC(2026, 9, 1, 0, minute)); + + async function mineFilters(userId: string, overrides: Partial = {}): Promise { + return { + readCondition: await issueReadSqlCondition(ctx.db, { type: "board", source: "session", userId, authorizingUserId: userId } as never), + touchedByUserId: userId, inboxArchivedByUserId: userId, + status: INBOX_MINE_ISSUE_STATUS_FILTER, listProjection: "index", + sortField: "updated", sortDir: "desc", limit: 100, + includeRoutineExecutions: true, includeLiveDescendantSummary: true, + ...overrides, + }; + } + + async function compare(companyId: string, filters: IssueFilters) { + const svc = issueService(ctx.db); + const scan = await svc.list(companyId, filters); + const ranked = await svc.list(companyId, { ...filters, touchedCandidates: "participation" }); + expect(ranked.map((row) => row.id)).toEqual(scan.map((row) => row.id)); + expect(JSON.parse(JSON.stringify(ranked))).toEqual(JSON.parse(JSON.stringify(scan))); + return scan.map((row) => row.id); + } + + it("selects the same ordered rows and payloads as the touched scan across participation edge cases", async () => { + const companyId = randomUUID(), foreignCompanyId = randomUUID(); + const userId = `mine-${randomUUID()}`, otherUserId = `other-${randomUUID()}`; + await ctx.db.insert(companies).values([ + { id: companyId, name: "Mine candidates", issuePrefix: `M${companyId.slice(0, 6).toUpperCase()}` }, + { id: foreignCompanyId, name: "Foreign", issuePrefix: `F${foreignCompanyId.slice(0, 6).toUpperCase()}` }, + ]); + const [agent] = await ctx.db.insert(agents).values({ companyId, name: "Worker", adapterType: "process", status: "active" }).returning(); + const named = new Map(); + const add = async (name: string, values: Partial = {}, minute = named.size) => { + const [row] = await ctx.db.insert(issues).values({ + companyId, title: name, status: "todo", createdByUserId: otherUserId, + createdAt: at(0), updatedAt: at(minute), ...values, + }).returning(); + named.set(name, row!.id); + return row!.id; + }; + const activity = (issueId: string, values: Partial) => ctx.db.insert(activityLog).values({ + companyId, entityType: "issue", entityId: issueId, actorType: "user", actorId: userId, action: "issue.updated", createdAt: at(500), ...values, + }); + const comment = (issueId: string, values: Partial) => ctx.db.insert(issueComments).values({ + companyId, issueId, body: "note", authorUserId: userId, createdAt: at(400), ...values, + }); + + // Included: every participation source, done work, and re-activated archives. + await add("created by me", { createdByUserId: userId }); + await add("assigned to me", { assigneeUserId: userId, createdByUserId: null, createdByAgentId: agent!.id }); + await activity(await add("updated by me"), {}); + await comment(await add("commented by me"), {}); + await add("done by me", { createdByUserId: userId, status: "done", completedAt: at(30) }); + const pending = await add("pending request", { createdByUserId: null, createdByAgentId: agent!.id }); + await ctx.db.insert(issueThreadInteractions).values({ + companyId, issueId: pending, kind: "request_confirmation", status: "pending", addresseeUserId: userId, + payload: { version: 1, prompt: "Confirm?" } as never, createdAt: at(600), + }); + const reactivated = await add("archived then re-activated", { createdByUserId: userId }); + await ctx.db.insert(issueInboxArchives).values({ companyId, issueId: reactivated, userId, archivedAt: at(700) }); + await comment(reactivated, { authorUserId: otherUserId, createdAt: at(800) }); + const readIssue = await add("read with later agent comment", { createdByUserId: userId }); + await ctx.db.insert(issueReadStates).values({ companyId, issueId: readIssue, userId, lastReadAt: at(900) }); + await comment(readIssue, { authorUserId: null, authorAgentId: agent!.id, createdAt: at(950) }); + const liveChild = await add("live child of mine", { createdByUserId: userId, parentId: named.get("created by me")! }); + const [run] = await ctx.db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, status: "running", contextSnapshot: { issueId: liveChild } }).returning(); + await ctx.db.update(issues).set({ executionRunId: run!.id }).where(sql`${issues.id} = ${liveChild}`); + + // Excluded by both selections. + const archived = await add("archived without new activity", { createdByUserId: userId }); + await ctx.db.insert(issueInboxArchives).values({ companyId, issueId: archived, userId, archivedAt: at(990) }); + await comment(await add("mention only"), { authorUserId: otherUserId, body: `@${userId} please look` }); + await activity(await add("read audit only"), { action: "issue.read_marked" }); + const upper = await add("uppercase entity id"); + await activity(upper, { entityId: upper.toUpperCase() }); + await comment(await add("channel comment"), { metadata: { sourceChannel: "slack" } as never }); + await add("agent created with my user id", { createdByUserId: userId, createdByAgentId: agent!.id }); + await add("cancelled by me", { createdByUserId: userId, status: "cancelled" }); + await add("hidden by me", { createdByUserId: userId, hiddenAt: at(1) }); + await add("harness by me", { createdByUserId: userId, harnessKind: "synthetic" }); + await add("plugin operation by me", { createdByUserId: userId, originKind: "plugin:demo:operation" }); + const privateId = randomUUID(); + await add("private to someone else", { + id: privateId, visibility: "private", privacyRootIssueId: privateId, responsibleUserId: otherUserId, + }); + await activity(privateId, {}); + const answered = await add("answered request", { createdByUserId: null, createdByAgentId: agent!.id }); + await ctx.db.insert(issueThreadInteractions).values({ + companyId, issueId: answered, kind: "request_confirmation", status: "answered", addresseeUserId: userId, + payload: { version: 1, prompt: "Confirm?" } as never, + }); + await ctx.db.insert(issues).values({ companyId: foreignCompanyId, title: "foreign", createdByUserId: userId }); + + // Page pressure: more than 100 touched issues, many with tied activity times. + for (let index = 0; index < 130; index++) { + await add(`bulk ${index}`, { createdByUserId: userId, status: index % 3 === 0 ? "done" : "in_progress" }, index % 4); + } + for (let index = 0; index < 40; index++) await add(`untouched ${index}`, {}, 1000 + index); + + const ids = await compare(companyId, await mineFilters(userId)); + expect(ids).toHaveLength(100); + const all = await compare(companyId, await mineFilters(userId, { limit: 1000 })); + for (const name of [ + "created by me", "assigned to me", "updated by me", "commented by me", "done by me", "pending request", + "archived then re-activated", "read with later agent comment", "live child of mine", + ]) expect(all, name).toContain(named.get(name)); + for (const name of [ + "archived without new activity", "mention only", "read audit only", "uppercase entity id", "channel comment", + "agent created with my user id", "cancelled by me", "hidden by me", "harness by me", "plugin operation by me", + "private to someone else", "answered request", "untouched 0", + ]) expect(all, name).not.toContain(named.get(name)); + expect(all.indexOf(named.get("archived then re-activated")!)).toBeLessThan(all.indexOf(named.get("created by me")!)); + + // Other selection shapes stay equivalent: no archive filter, other statuses, + // offsets, ascending order, unrestricted read access, and another user. + await compare(companyId, await mineFilters(userId, { inboxArchivedByUserId: undefined })); + await compare(companyId, await mineFilters(userId, { status: "done", listProjection: undefined, includeLiveDescendantSummary: false })); + await compare(companyId, await mineFilters(userId, { offset: 50, limit: 30 })); + await compare(companyId, await mineFilters(userId, { sortDir: "asc" })); + await compare(companyId, { ...await mineFilters(userId), readCondition: sql`true` }); + expect(await compare(companyId, await mineFilters(otherUserId, { limit: 1000 }))).toContain(named.get("archived then re-activated")); + expect(await compare(companyId, await mineFilters(`nobody-${randomUUID()}`))).toEqual([]); + }); +}); diff --git a/server/src/__tests__/inbox-mine-routes.test.ts b/server/src/__tests__/inbox-mine-routes.test.ts index 28cf3be36b3..485abe3977f 100644 --- a/server/src/__tests__/inbox-mine-routes.test.ts +++ b/server/src/__tests__/inbox-mine-routes.test.ts @@ -112,9 +112,12 @@ describeEmbeddedPostgres("Mine inbox snapshot routes", () => { })); for (const phase of [ "paperclip_inbox", "inbox_access", "inbox_issues", "inbox_parents", "inbox_failures", "inbox_failure_context", - "inbox_issues_query", "inbox_issues_labels", "inbox_issues_user_stats", "inbox_issues_read_stats", + "inbox_issues_candidates", "inbox_issues_query", "inbox_issues_labels", "inbox_issues_user_stats", "inbox_issues_read_stats", "inbox_issues_last_activity", "inbox_issues_archives", "inbox_issues_live_descendants", ]) expect(timingPhases.get(phase), phase).toBeGreaterThanOrEqual(0); + // Attribution: list sub-phases run inside inbox_issues, and every phase inside the request. + expect(timingPhases.get("inbox_issues")!).toBeGreaterThanOrEqual(timingPhases.get("inbox_issues_candidates")! + timingPhases.get("inbox_issues_query")! - 0.2); + expect(timingPhases.get("paperclip_inbox")!).toBeGreaterThanOrEqual(timingPhases.get("inbox_issues")! - 0.1); expect(snapshot).not.toHaveProperty("failureError"); expect(snapshot).not.toHaveProperty("contextError"); }); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 14cad9981a9..8928c30958c 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -8831,7 +8831,7 @@ export function issueRoutes( if (!userId) throw new HttpError(401, "A signed-in board user is required"); const readCondition = await timing.time("access", () => issueReadSqlCondition(db, req.actor)); const listFilters: IssueFilters = { - readCondition, touchedByUserId: userId, inboxArchivedByUserId: userId, + readCondition, touchedByUserId: userId, touchedCandidates: "participation", inboxArchivedByUserId: userId, status: INBOX_MINE_ISSUE_STATUS_FILTER, listProjection: "index", sortField: "updated", sortDir: "desc", limit: 100, includeRoutineExecutions: true, includeLiveDescendantSummary: true, diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index cec7d7a5090..3ad5ebad329 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1867,7 +1867,7 @@ export function parseStatusFilter( /** Closed, content-free phases of `issueService.list`, for response timing headers. */ export type IssueListPhase = - | "query" | "labels" | "active_runs" | "user_stats" | "read_stats" | "last_activity" + | "candidates" | "query" | "labels" | "active_runs" | "user_stats" | "read_stats" | "last_activity" | "archives" | "blocked_by" | "live_descendants" | "attention"; export interface IssueListOptions { @@ -1898,6 +1898,14 @@ export interface IssueFilters { oneChildPerParent?: boolean; assigneeUserId?: string; touchedByUserId?: string; + /** + * `participation` ranks touched issues from the indexed participation + * UNION (see `mineParticipationIssueIds`) and loads full rows only for the + * selected page, instead of scanning every status-matching issue with the + * five-way `touchedByUserCondition`. Same rows and order. Requires + * `touchedByUserId`; ignored with search or `oneChildPerParent`. + */ + touchedCandidates?: "participation"; inboxArchivedByUserId?: string; unreadForUserId?: string; projectId?: string; @@ -3028,6 +3036,36 @@ function unreadForUserCondition(companyId: string, userId: string) { `; } +/** + * Ids of every issue the user participated in, built from indexed sources. + * Each UNION member is one branch of `touchedByUserCondition`, so membership + * equals that condition. The result is aliased `mine_participants` (`id`). + */ +function mineParticipationIssueIds(db: Db, companyId: string, userId: string) { + const directlyParticipated = or( + and(eq(issues.createdByUserId, userId), humanCreatedIssueCondition()), + eq(issues.assigneeUserId, userId), + ); + const participationSources = unionAll( + db.select({ id: sql`${issues.id}`.as("id") }).from(issues).where(and( + eq(issues.companyId, companyId), + directlyParticipated, + )), + db.select({ id: sql`case when length(${activityLog.entityId}) = 36 and ${activityLog.entityId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then ${activityLog.entityId}::uuid end`.as("id") }).from(activityLog).where(and( + userParticipationActivityCondition(companyId, userId), + notInArray(activityLog.entityId, db.select({ id: sql`${issues.id}::text` }).from(issues) + .where(and(eq(issues.companyId, companyId), directlyParticipated))), + )), + db.select({ id: sql`${issueComments.issueId}`.as("id") }).from(issueComments).where(and( + eq(issueComments.companyId, companyId), eq(issueComments.authorUserId, userId), directHumanCommentCondition(), + )), + db.select({ id: sql`${issues.id}`.as("id") }).from(issueThreadInteractions) + .innerJoin(issues, and(eq(issues.id, issueThreadInteractions.issueId), eq(issues.companyId, companyId))) + .where(pendingUserRequestRowCondition(companyId, userId)), + ).as("mine_participation_sources"); + return db.selectDistinct({ id: participationSources.id }).from(participationSources).as("mine_participants"); +} + function inboxVisibleForUserCondition(companyId: string, userId: string) { return sql` NOT EXISTS ( @@ -8239,28 +8277,7 @@ export function issueService(db: Db) { unreadInboxIssueIds: async (companyId: string, userId: string, readCondition: SQL) => { // Rank the visible Mine set before counting unread, just like the inbox. // This projection never reads descriptions, run payloads, or recovery details. - const directlyParticipated = or( - and(eq(issues.createdByUserId, userId), humanCreatedIssueCondition()), - eq(issues.assigneeUserId, userId), - ); - const participationSources = unionAll( - db.select({ id: sql`${issues.id}`.as("id") }).from(issues).where(and( - eq(issues.companyId, companyId), - directlyParticipated, - )), - db.select({ id: sql`case when length(${activityLog.entityId}) = 36 and ${activityLog.entityId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then ${activityLog.entityId}::uuid end`.as("id") }).from(activityLog).where(and( - userParticipationActivityCondition(companyId, userId), - notInArray(activityLog.entityId, db.select({ id: sql`${issues.id}::text` }).from(issues) - .where(and(eq(issues.companyId, companyId), directlyParticipated))), - )), - db.select({ id: sql`${issueComments.issueId}`.as("id") }).from(issueComments).where(and( - eq(issueComments.companyId, companyId), eq(issueComments.authorUserId, userId), directHumanCommentCondition(), - )), - db.select({ id: sql`${issues.id}`.as("id") }).from(issueThreadInteractions) - .innerJoin(issues, and(eq(issues.id, issueThreadInteractions.issueId), eq(issues.companyId, companyId))) - .where(pendingUserRequestRowCondition(companyId, userId)), - ).as("mine_participation_sources"); - const participated = db.selectDistinct({ id: participationSources.id }).from(participationSources).as("mine_participants"); + const participated = mineParticipationIssueIds(db, companyId, userId); const eligible = db.select({ id: issues.id, updatedAt: issues.updatedAt }) .from(participated).innerJoin(issues, eq(issues.id, sql`"mine_participants"."id"`)).where(and( eq(issues.companyId, companyId), @@ -8354,6 +8371,10 @@ export function issueService(db: Db) { const rawSearch = filters?.q?.trim() ?? ""; const hasSearch = rawSearch.length > 0; const taskSearch = parseTaskSearch(rawSearch); + // Rank page ids from the user's participation set, then load full rows + // only for that page (see IssueFilters.touchedCandidates). + const participationCandidates = filters?.touchedCandidates === "participation" + && touchedByUserId !== undefined && !hasSearch && !filters?.oneChildPerParent; if (filters?.createdFromIssueId) { conditions.push(createdFromIssueCondition(companyId, filters.createdFromIssueId)); } @@ -8402,7 +8423,9 @@ export function issueService(db: Db) { conditions.push(eq(issues.assigneeUserId, filters.assigneeUserId)); } if (touchedByUserId) { - conditions.push(touchedByUserCondition(companyId, touchedByUserId)); + conditions.push(participationCandidates + ? sql`${issues.id} IN (SELECT "mine_participants"."id" FROM ${mineParticipationIssueIds(db, companyId, touchedByUserId)})` + : touchedByUserCondition(companyId, touchedByUserId)); } if (inboxArchivedByUserId) { conditions.push( @@ -8483,6 +8506,28 @@ export function issueService(db: Db) { description: sql`NULL`, descriptionTruncated: sql`false`, } : issueListSelect; + const orderBy = [ + ...(filters?.oneChildPerParent ? [asc(issues.parentId)] : []), + ...issueListOrderBy(companyId, { + hasSearch, + priorityOrder, + searchOrder, + sortField: filters?.sortField, + sortDir: filters?.sortDir, + }), + ]; + const paginate = (query: Q) => + offset > 0 + ? limit === undefined ? query.offset(offset) : query.limit(limit).offset(offset) + : limit === undefined ? query : query.limit(limit); + // Phase 1 of the participation path: the page's ids in final order. The + // select-list subqueries below then run only for these rows. + const pageIds = participationCandidates + ? (await timePhase("candidates", async () => paginate( + db.select({ id: issues.id }).from(issues).where(and(...conditions)).orderBy(...orderBy).$dynamic(), + ))).map((row) => row.id) + : null; + if (pageIds?.length === 0) return []; const issueSource = (filters?.oneChildPerParent ? db.selectDistinctOn([issues.parentId], selectedFields) : db.select(selectedFields)).from(issues); @@ -8492,27 +8537,22 @@ export function issueService(db: Db) { SELECT m.id, ${taskSearchScore(taskSearch)} AS score FROM matched m ) task_search`, sql`task_search.id = ${issues.id}`) : issueSource; - const baseQuery = searchedSource - .where(and(...conditions)) - .orderBy( - ...(filters?.oneChildPerParent ? [asc(issues.parentId)] : []), - ...issueListOrderBy(companyId, { - hasSearch, - priorityOrder, - searchOrder, - sortField: filters?.sortField, - sortDir: filters?.sortDir, - }), - ); - const pageQuery = - offset > 0 - ? limit === undefined - ? baseQuery.offset(offset) - : baseQuery.limit(limit).offset(offset) - : limit === undefined - ? baseQuery - : baseQuery.limit(limit); - const rows = (await timePhase("query", async () => pageQuery)).map((row) => ({ + const pageQuery = pageIds + // Re-check company, visibility and read access for the selected ids; + // the rows are put back into the phase 1 order below. + ? searchedSource.where(and( + eq(issues.companyId, companyId), + inArray(issues.id, pageIds), + visibleIssueCondition(), + filters?.readCondition, + )) + : paginate(searchedSource.where(and(...conditions)).orderBy(...orderBy).$dynamic()); + const pageRows = await timePhase("query", async () => pageQuery); + if (pageIds) { + const position = new Map(pageIds.map((id, index) => [id, index])); + pageRows.sort((a, b) => position.get(a.id)! - position.get(b.id)!); + } + const rows = pageRows.map((row) => ({ ...row, description: decodeDatabaseTextPreview( row.description,