From 8e4c5e81b991c52cf0834e2c236de61616d80145 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 12:38:51 -0600 Subject: [PATCH] perf(issues): read created-from tasks by index and inbox review waits by page (#20573) 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 task page shows "Created from this task". The inbox shows blocked-inbox attention for a page of tasks. > - On a hosted database, the created-from list ran about 4,000 times in one afternoon at about 499 ms each. The task page sends it each time a task opens. > - Its filter was an `OR` of three provenance paths over the list's own rows. PostgreSQL could not use an index for it. It read every task of the company and tested each one, with the list's other per-row conditions (such as the chat conversation state subquery). > - The blocked-inbox attention map also read external review waits for every in-review task of the company, not only for the tasks of the page. > - This pull request reads created-from tasks from three index probes, and reads review waits only for the in-review tasks that can change a finding of the page. > - The benefit is less database work on each task open and each inbox page. The responses stay the same. ## Linked Issues or Issue Description Refs #20400 (shares review-wait reads between the attention maps of one request). Refs #20551 (authorizes the review waits of a page with one read; this change narrows the waits that the page asks for, and adds a test that pins the batched read). Refs #13241 (adds the created-from list). **What happened** - `GET /api/companies/:id/issues?createdFromIssueId=…` used `createdFromIssueCondition`. Its filter was `origin_issue_id = source OR (origin_issue_id IS NULL AND origin_run_id IN source runs) OR (… AND EXISTS creation activity of a source run)`. The `OR` stops index use, so the statement read every task of the company. The list's other conditions, which include the chat conversation state subquery, ran for each task before the task was excluded. - `listIssueBlockedInboxAttentionMap` read review waits for every in-review task of the company graph. The caller uses the findings of the page rows only. **Expected behavior** The created-from list reads only the created tasks. The blocked-inbox map reads review waits only for in-review tasks that the liveness classifier can read for a page row. The rows, the order, and the attention values do not change. **Steps to reproduce** 1. Use a company with tens of thousands of tasks, runs, and activity rows. 2. Open a task page. Look at the created-from list statement: it reads the full task set of the company. 3. Open an inbox page with blocked-inbox attention. Look at the review-wait read: it gets the ID of every in-review task of the company. ## What Changed - `server/src/services/issue-creation-origin.ts`: `createdFromIssueCondition` is now `issues.id <> source AND issues.id IN (three branches)`. Each branch starts from an index: - recorded origin task: `issues_company_origin_issue_idx`; - legacy origin run: the source task's runs (`heartbeat_runs_company_origin_issue_idx`), then the new partial index on `issues (company_id, origin_run_id)`; - creation activity: `activity_log_run_id_idx` for the source runs, then the task by primary key. The precedence does not change: a recorded origin task wins, the run path applies only without one, and the activity path applies only without both. The activity path matches only an `entity_id` that is the canonical text of the task ID, as before (it also compares `id::text = entity_id`). - `packages/db`: add `issues_company_legacy_origin_run_idx` on `issues (company_id, origin_run_id) WHERE origin_issue_id IS NULL AND origin_run_id IS NOT NULL`. Migration `0461` creates it with `CREATE INDEX CONCURRENTLY IF NOT EXISTS` and is registered with the other concurrent migrations. `doc/DATABASE.md` documents the index and the benchmark command. - `server/src/services/issues.ts`: add `blockedInboxReviewWaitIssueIds`. It returns the in-review tasks in the blocker closure of the page rows. The blocked-inbox map reads review waits only for these tasks. The liveness classifier reads a wait only for a row or for a blocker that it reaches from a row, and the caller reads only the findings of the rows. The classifier does not change. - Tests: - `issue-created-from-routes.test.ts`: new cases for precedence in both directions, a source-run comment, an activity row whose task has another origin run, a non-canonical and a non-UUID activity entity ID, the source's own creation activity, and other-company rows. - `blocked-inbox-review-wait-scope.test.ts`: the narrowed set for chains, cycles, shared blockers and off-page tasks. The rows get the same findings with the narrowed waits as with the waits of every in-review task, and each narrowed wait is needed. - `issue-list-review-wait-authorization.integration.test.ts`: on real PostgreSQL with two authorized waits, `listExternalReviewWaits` and one issue-list request with blocked-inbox attention each send one bound-run message-link statement. With the per-wait authorization from before #20551, both tests fail (2 statements). - `scripts/benchmark-created-from-issues.ts`: local benchmark. It loads the baseline `issues.ts` and `issue-creation-origin.ts` from a pinned commit, asserts identical complete row payloads, and reports alternating timings and `EXPLAIN (ANALYZE, BUFFERS)`. ## Verification - `pnpm --filter @paperclipai/server exec vitest run src/__tests__/issue-created-from-routes.test.ts src/__tests__/blocked-inbox-review-wait-scope.test.ts src/__tests__/issue-list-review-wait-authorization.integration.test.ts src/__tests__/agent-task-hierarchy-depth.test.ts src/__tests__/issue-blocker-attention.test.ts src/__tests__/issue-list-assignee-filter-routes.test.ts src/__tests__/issues-parent-id-alias.test.ts src/__tests__/batched-chat-authorization.integration.test.ts src/__tests__/issue-liveness.test.ts` passes (126 tests). One timing test in `issue-list-assignee-filter-routes` failed one time under machine load and passed on rerun. - `pnpm --filter @paperclipai/server typecheck` and `pnpm --filter @paperclipai/db check:migrations` pass. - `pnpm verify` asks for broader verification because the change adds a file under `scripts/`. Required CI supplies that coverage. - Benchmark: `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-created-from-issues.ts --baseline-ref=bd30ef8510` Fixture (synthetic, local PostgreSQL 18.1): one company with 30,000 tasks (10,628 with a recorded origin task, 10,445 with only a legacy origin run), 60,030 runs with wide context snapshots (145 MB), and 261,079 activity rows; a second company with 10,000 tasks and 20,000 runs. The busy source task has 63 created tasks through all three paths, plus decoys. The quiet source task has 3. The UI's page: `sortField=id`, `asc`, `limit=500`, routine executions included. Both versions return identical complete payloads. 9 alternating pairs after 2 warm-up pairs: | Case | List statement before | List statement after | List call before, median (min to max) | List call after, median (min to max) | | --- | --- | --- | --- | --- | | Busy source (63 tasks) | 234.9 ms, 47,437 buffers | 0.90 ms, 437 buffers | 156.0 ms (114.5 to 480.7) | 31.6 ms (19.9 to 62.7) | | Quiet source (3 tasks) | 90.7 ms, 47,413 buffers | 0.25 ms, 46 buffers | 126.1 ms (90.6 to 300.4) | 10.5 ms (7.2 to 26.1) | The list call time includes the enrichment reads (labels, runs, attention), which do not change. Inbox page (50 open tasks, blocked-inbox attention): the review-wait read goes from 2 statements with 1,215 parameters (every in-review task of the company) to 1 statement with 31 parameters. The page time does not change in this fixture (median 801 ms before, 873 ms after, both with a wide spread), because the fixture has no review waits and the other graph reads of the map are most of the time. With real waits, the server does an inherited GitHub review read for each wait it finds, so the gain grows with the number of off-page waits. Migration timing (local PostgreSQL 18.1): `0461` (measured as the same statement under its earlier number) on 500,000 tasks (846 MB, 20 companies, one third with a legacy origin run) takes 1.6 s. The index is 12 MB. ## Risks - The new index adds a small write cost to tasks with only an origin run. New tasks with a source task record `origin_issue_id`, so the index holds mostly older tasks. The migration does not lock writes. - The created-from rewrite must keep the same rows. The new tests and the benchmark payload check cover precedence, company scope, and ID text forms. Visibility and read filters are the list's other conditions and do not change. - The review-wait narrowing depends on the liveness classifier reading waits only for rows and their blocker chains. A unit test compares the findings of the rows with full and narrowed waits. A future classifier change that reads waits of other tasks must update `blockedInboxReviewWaitIssueIds`. The function comment says this. - The local timings are synthetic. Hosted acceptance needs a fresh query-statistics window after deployment. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use (shell, file edits, local PostgreSQL) in Claude Code. ## 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 --- doc/DATABASE.md | 16 ++ packages/db/src/client.ts | 1 + ...1_created_from_legacy_origin_run_index.sql | 1 + ...{0456_snapshot.json => 0461_snapshot.json} | 120 ++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/issues.ts | 7 + scripts/benchmark-created-from-issues.ts | 253 ++++++++++++++++++ .../blocked-inbox-review-wait-scope.test.ts | 81 ++++++ .../issue-created-from-routes.test.ts | 22 ++ ...iew-wait-authorization.integration.test.ts | 98 +++++++ server/src/services/issue-creation-origin.ts | 48 +++- server/src/services/issues.ts | 36 ++- 12 files changed, 673 insertions(+), 17 deletions(-) create mode 100644 packages/db/src/migrations/0461_created_from_legacy_origin_run_index.sql rename packages/db/src/migrations/meta/{0456_snapshot.json => 0461_snapshot.json} (99%) create mode 100644 scripts/benchmark-created-from-issues.ts create mode 100644 server/src/__tests__/blocked-inbox-review-wait-scope.test.ts create mode 100644 server/src/__tests__/issue-list-review-wait-authorization.integration.test.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index fa124cd343a..bcd663b1803 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -439,6 +439,22 @@ To compare both plans on a disposable local fixture: pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-workspace-activity.ts --baseline-ref= --runs-per-workspace=3000 ``` +The task page's created-from list (`createdFromIssueCondition` in +`server/src/services/issue-creation-origin.ts`) reads created tasks through +three index probes: the recorded origin task (`issues_company_origin_issue_idx`), +the source task's runs (`heartbeat_runs_company_origin_issue_idx`) and then the +tasks whose only origin is one of those runs, and the creation activity of those +runs (`activity_log_run_id_idx`). Migration 0461 builds the partial index +`issues_company_legacy_origin_run_idx` concurrently for the run step. It keys +tasks by company and `origin_run_id`, only where `origin_issue_id` is null and +`origin_run_id` is not null. The run branch must repeat that predicate textually. +A recorded origin task still wins over a run, and a run over activity. To compare +both plans on a disposable local fixture: + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-created-from-issues.ts --baseline-ref= +``` + GitHub review lookups use four indexes from migration 0456, which builds them concurrently. `chat_message_links_company_comment_idx` finds a source comment's link for inherited review waits and run-origin checks. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index c5fdc77313c..4b574436ef5 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -623,6 +623,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0458_chat_admission_issue_index.sql", "0459_environment_lease_backup_scope_index.sql", "0460_workspace_activity_context_index.sql", + "0461_created_from_legacy_origin_run_index.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0461_created_from_legacy_origin_run_index.sql b/packages/db/src/migrations/0461_created_from_legacy_origin_run_index.sql new file mode 100644 index 00000000000..cffb794b61d --- /dev/null +++ b/packages/db/src/migrations/0461_created_from_legacy_origin_run_index.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "issues_company_legacy_origin_run_idx" ON "issues" USING btree ("company_id","origin_run_id") WHERE "issues"."origin_issue_id" is null and "issues"."origin_run_id" is not null; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0456_snapshot.json b/packages/db/src/migrations/meta/0461_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0456_snapshot.json rename to packages/db/src/migrations/meta/0461_snapshot.json index ea96ea87ab7..c68baee017d 100644 --- a/packages/db/src/migrations/meta/0456_snapshot.json +++ b/packages/db/src/migrations/meta/0461_snapshot.json @@ -1,6 +1,6 @@ { - "id": "a2606b41-36c2-4eb5-a59d-28cc730bc455", - "prevId": "2be53987-902a-41b6-8cb4-6e89a768f8cf", + "id": "f409684f-c629-4648-9123-a5e1064308ef", + "prevId": "f93ccf18-7e86-4788-be77-b17430c86b2f", "version": "7", "dialect": "postgresql", "tables": { @@ -10188,6 +10188,40 @@ "method": "btree", "with": {} }, + "chat_actions_company_issue_kind_idx": { + "name": "chat_actions_company_issue_kind_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"payload\"->>'issueId')", + "asc": true, + "isExpression": true, + "nulls": "last" + }, + { + "expression": "kind", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" IN ('inbound_wakeup', 'failed_run_retry')", + "concurrently": true, + "method": "btree", + "with": {} + }, "chat_actions_github_stale_head_idx": { "name": "chat_actions_github_stale_head_idx", "columns": [ @@ -29133,6 +29167,22 @@ "concurrently": false, "method": "btree", "with": {} + }, + "environment_leases_native_backup_scope_idx": { + "name": "environment_leases_native_backup_scope_idx", + "columns": [ + { + "expression": "(\"metadata\"->'nativeHarnessBackup'->>'sessionScopeSha256')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "(\"environment_leases\".\"metadata\"->'nativeHarnessBackup'->>'sessionScopeSha256') is not null", + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -35708,6 +35758,28 @@ "method": "btree", "with": {} }, + "heartbeat_runs_company_native_session_idx": { + "name": "heartbeat_runs_company_native_session_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "native_session_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"native_session_id\" is not null", + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_company_issue_created_idx": { "name": "heartbeat_runs_company_issue_created_idx", "columns": [ @@ -35945,6 +36017,28 @@ "method": "btree", "with": {} }, + "heartbeat_runs_company_ctx_execution_workspace_idx": { + "name": "heartbeat_runs_company_ctx_execution_workspace_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"context_snapshot\" ->> 'executionWorkspaceId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"context_snapshot\" ? 'executionWorkspaceId'", + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_company_ctx_paperclip_issue_created_idx": { "name": "heartbeat_runs_company_ctx_paperclip_issue_created_idx", "columns": [ @@ -43486,6 +43580,28 @@ "method": "btree", "with": {} }, + "issues_company_legacy_origin_run_idx": { + "name": "issues_company_legacy_origin_run_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "origin_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"issues\".\"origin_issue_id\" is null and \"issues\".\"origin_run_id\" is not null", + "concurrently": true, + "method": "btree", + "with": {} + }, "issues_company_project_workspace_idx": { "name": "issues_company_project_workspace_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index a3932712d56..dde23060e66 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3214,6 +3214,13 @@ "when": 1794332988056, "tag": "0460_workspace_activity_context_index", "breakpoints": true + }, + { + "idx": 461, + "version": "7", + "when": 1794334724911, + "tag": "0461_created_from_legacy_origin_run_index", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/issues.ts b/packages/db/src/schema/issues.ts index 11f9c134984..60e7b2408d7 100644 --- a/packages/db/src/schema/issues.ts +++ b/packages/db/src/schema/issues.ts @@ -132,6 +132,13 @@ export const issues = pgTable( projectIdx: index("issues_company_project_idx").on(table.companyId, table.projectId), originIdx: index("issues_company_origin_idx").on(table.companyId, table.originKind, table.originId), originIssueIdx: index("issues_company_origin_issue_idx").on(table.companyId, table.originIssueId), + // Tasks whose creation provenance is only a run. The created-from list + // probes this set by the source task's runs. The predicate must stay + // identical to the legacy run branch of `createdFromIssueCondition`. + legacyOriginRunIdx: index("issues_company_legacy_origin_run_idx") + .on(table.companyId, table.originRunId) + .where(sql`${table.originIssueId} is null and ${table.originRunId} is not null`) + .concurrently(), projectWorkspaceIdx: index("issues_company_project_workspace_idx").on(table.companyId, table.projectWorkspaceId), executionWorkspaceIdx: index("issues_company_execution_workspace_idx").on(table.companyId, table.executionWorkspaceId), dueMonitorIdx: index("issues_company_monitor_due_idx").on(table.companyId, table.monitorNextCheckAt), diff --git a/scripts/benchmark-created-from-issues.ts b/scripts/benchmark-created-from-issues.ts new file mode 100644 index 00000000000..ed693a2c9da --- /dev/null +++ b/scripts/benchmark-created-from-issues.ts @@ -0,0 +1,253 @@ +// Local SQL benchmark for the task page "Created from this task" list +// (GET /api/companies/:id/issues?createdFromIssueId=…&includeRoutineExecutions=true, +// which the UI pages with sortField=id, sortDir=asc and limit=500). +// +// pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-created-from-issues.ts \ +// --baseline-ref= [--issues=30000] [--runs=60000] [--samples=9] +// [--evidence-dir=] +// +// Seeds one disposable embedded-PostgreSQL company with tens of thousands of +// tasks: about a third with a recorded origin task, a third with only a legacy +// origin run, and a third manual. Runs carry a realistic context snapshot and +// every agent-created task has creation activity. The source task has +// created tasks through each of the three provenance paths, plus decoys. +// The script runs the baseline (from --baseline-ref) and the working-tree +// `issueService.list` with the route's filters, for several source tasks, +// asserts identical complete row payloads, and reports alternating timings and +// EXPLAIN (ANALYZE, BUFFERS) of the list statement. An inbox case lists a page +// with blocked-inbox attention and explains its external review-wait read. +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 { 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 runCount = Number(arg("runs") ?? 60000); +const samples = Number(arg("samples") ?? 9); +const evidenceDir = arg("evidence-dir"); + +// Baseline modules: the created-from condition and the list that embeds it. +// Every other dependency is held constant at the working tree. +const modulePaths = ["server/src/services/issues.ts", "server/src/services/issue-creation-origin.ts"]; +type Captured = { sql: string; params: unknown[] }; +let database: Awaited> | undefined; +let db: Db | undefined; +const keepAlive = setInterval(() => {}, 1000); +const baselineDir = await fs.mkdtemp(path.join(repositoryRoot, "server/scripts/.created-from-baseline-")); +// Everything after the copy directory exists runs inside the cleanup scope, +// so a baseline that cannot be read or loaded leaves no copied sources. +try { + 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-creation-origin.js") ? path.join(baselineDir, "issue-creation-origin.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"); + + database = await startEmbeddedPostgresTestDatabase("created-from-benchmark-"); + await applyPendingMigrations(database.connectionString); + 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 seedStart = performance.now(); + const companyId = randomUUID(), otherCompanyId = randomUUID(); + await client.unsafe(`insert into companies(id,name,issue_prefix) values($1,'Synthetic Origin','ORG'),($2,'Synthetic Other','OTH')`, [companyId, otherCompanyId]); + 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)}`]); + const otherAgentId = randomUUID(); + await client.unsafe(`insert into agents(id,company_id,name,adapter_type,status) values($1,$2,'Other','process','active')`, [otherAgentId, otherCompanyId]); + 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 $$`); + // Tasks first (no origins), then runs on those tasks, then origins. + await client.unsafe(`insert into issues(id,company_id,title,description,status,identifier,issue_number,created_by_agent_id,origin_kind,created_at,updated_at) + select gen_random_uuid(),$1::uuid,'Synthetic '||n,repeat('Task description. ',40), + case when bench_h(n,'s')<55 then 'done' when bench_h(n,'s')<62 then 'cancelled' when bench_h(n,'s')<72 then 'backlog' + when bench_h(n,'s')<84 then 'todo' when bench_h(n,'s')<92 then 'in_progress' when bench_h(n,'s')<96 then 'in_review' else 'blocked' end, + 'ORG-'||n,n, + case when bench_h(n,'o')<70 then ($2::uuid[])[1+n%3] else null end, + case when n%41=0 then 'routine_execution' else 'manual' end, + '2026-01-01'::timestamptz+n*interval '1 minute','2026-01-01'::timestamptz+n*interval '1 minute' + from generate_series(1,$3::int) n`, [companyId, agentIds, issueCount]); + await client.unsafe(`insert into issues(company_id,title,status,identifier,issue_number) + select $1::uuid,'Other '||n,'todo','OTH-'||n,n from generate_series(1,$2::int) n`, [otherCompanyId, Math.round(issueCount / 3)]); + await client.unsafe(`create table bench_fixture_issues as select id, issue_number n, identifier from issues where company_id=$1`, [companyId]); + await client.unsafe(`create unique index on bench_fixture_issues(n)`); + // Runs: legacy context snapshots (issueId / taskId / taskKey) and native runs, + // with a wide snapshot like real wakes. + await client.unsafe(`insert into heartbeat_runs(id,company_id,agent_id,status,runtime_mode,native_issue_id,context_snapshot,created_at) + select gen_random_uuid(),$1::uuid,($2::uuid[])[1+g%3],'succeeded', + case when g%4=0 then 'native' else 'legacy' end, + case when g%4=0 then f.id else null end, + jsonb_build_object( + case when g%7=0 then 'taskKey' when g%5=0 then 'taskId' else 'issueId' end, + case when g%7=0 then f.identifier else f.id::text end, + 'wakeReason','issue_assigned','source','assignment','prompt',repeat('context ',120)), + '2026-01-01'::timestamptz+g*interval '30 seconds' + from generate_series(1,$3::int) g join bench_fixture_issues f on f.n = 1 + (g * 7919) % $4::int`, [companyId, agentIds, runCount, issueCount]); + await client.unsafe(`insert into heartbeat_runs(company_id,agent_id,status,context_snapshot) + select $1::uuid,$2::uuid,'succeeded',jsonb_build_object('issueId',gen_random_uuid()::text) from generate_series(1,$3::int)`, [otherCompanyId, otherAgentId, Math.round(runCount / 3)]); + // Origins for agent-created tasks: an earlier task (recorded) or a run (legacy). + await client.unsafe(`update issues i set + origin_issue_id = case when bench_h(f.n,'p')<50 then p.id else null end, + origin_run_id = (select r.id::text from heartbeat_runs r where r.company_id=$1 and r.created_at <= '2026-01-01'::timestamptz+f.n*interval '1 minute' order by r.created_at desc limit 1) + from bench_fixture_issues f join bench_fixture_issues p on p.n = greatest(1, f.n - 1 - (f.n * 31) % 500) + where i.id=f.id and i.created_by_agent_id is not null and f.n > 1`, [companyId]); + // Creation activity for every agent-created task, plus ordinary activity. + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,run_id,details,created_at) + select $1::uuid,'issue',i.id::text,'agent',i.created_by_agent_id::text, + case when i.parent_id is null then 'issue.created' else 'issue.child_created' end, + case when i.origin_run_id is not null then i.origin_run_id::uuid else null end,'{}'::jsonb,i.created_at + from issues i where i.company_id=$1 and i.created_by_agent_id is not null`, [companyId]); + 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,'{"status":"todo"}'::jsonb, + '2026-01-20'::timestamptz+(f.n*8+k)*interval '1 second' + from bench_fixture_issues f cross join generate_series(0,7) k`, [companyId, agentIds]); + + // The source tasks: a busy one with every provenance path, and a quiet one + // with only recorded origins. + const [busy] = await client.unsafe(`select id, identifier from bench_fixture_issues where n = $1`, [Math.floor(issueCount / 2)]); + const [quiet] = await client.unsafe(`select id from bench_fixture_issues where n = $1`, [Math.floor(issueCount / 3)]); + const sourceRuns: string[] = []; + for (let k = 0; k < 30; k++) { + const runId = randomUUID(); + sourceRuns.push(runId); + await client.unsafe(`insert into heartbeat_runs(id,company_id,agent_id,status,runtime_mode,native_issue_id,context_snapshot) + values($1,$2,$3,'succeeded',$4,$5,$6::jsonb)`, [runId, companyId, agentIds[k % 3], k % 3 === 0 ? "native" : "legacy", k % 3 === 0 ? busy.id : null, + JSON.stringify(k % 3 === 1 ? { taskKey: busy.identifier, prompt: "x".repeat(800) } : { issueId: busy.id, prompt: "x".repeat(800) })]); + } + const pick = async (limit: number, offset: number) => (await client.unsafe( + `select id from bench_fixture_issues where n > 100 order by n limit $1 offset $2`, [limit, offset])).map((row: { id: string }) => row.id); + const recorded = await pick(40, 0); + await client.unsafe(`update issues set origin_issue_id=$1 where id = any($2::uuid[])`, [busy.id, recorded]); + const legacyRun = await pick(15, 40); + await client.unsafe(`update issues set origin_issue_id=null, origin_run_id=$1 where id = any($2::uuid[])`, [sourceRuns[1], legacyRun]); + const activityOnly = await pick(8, 55); + await client.unsafe(`update issues set origin_issue_id=null, origin_run_id=null where id = any($1::uuid[])`, [activityOnly]); + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,run_id) + select $1::uuid,'issue',id::text,'agent','agent','issue.child_created',$2::uuid from unnest($3::uuid[]) id`, [companyId, sourceRuns[2], activityOnly]); + // Decoys: a recorded origin elsewhere wins over a source run; a source-run + // comment is not creation; another task's run. + const decoys = await pick(6, 70); + await client.unsafe(`update issues set origin_issue_id=$1, origin_run_id=$2 where id = any($3::uuid[])`, [quiet.id, sourceRuns[3], decoys.slice(0, 3)]); + await client.unsafe(`update issues set origin_issue_id=null, origin_run_id=null where id = any($1::uuid[])`, [decoys.slice(3)]); + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,run_id) + select $1::uuid,'issue',id::text,'agent','agent','issue.comment_added',$2::uuid from unnest($3::uuid[]) id`, [companyId, sourceRuns[4], decoys.slice(3)]); + 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 origin_issue_id is not null) recorded_origin, + (select count(*)::int from issues where company_id=$1 and origin_issue_id is null and origin_run_id is not null) legacy_run_origin, + (select count(*)::int from heartbeat_runs where company_id=$1) runs, + (select count(*)::int from activity_log where company_id=$1) activity, + (select pg_size_pretty(pg_total_relation_size('heartbeat_runs'))) runs_size, + (select pg_size_pretty(pg_total_relation_size('issues'))) issues_size`, [companyId]); + const fixtureMs = performance.now() - seedStart; + + const readCondition = await issueReadSqlCondition(db, { type: "board", source: "session", userId: "board-user", isInstanceAdmin: true } as never); + // The UI's listAll page: route filters with sortField=id, asc, limit 500. + const filtersFor = (sourceIssueId: string): IssueFilters => ({ + readCondition, createdFromIssueId: sourceIssueId, includeRoutineExecutions: true, + sortField: "id", sortDir: "asc", limit: 500, + }); + // The inbox case is a page of issues with blocked-inbox attention: its + // review-wait read used to cover every in-review task of the company. + const cases: Record = { + busy: { filters: filtersFor(busy.id), statement: /"creation_source_issue"/ }, + quiet: { filters: filtersFor(quiet.id), statement: /"creation_source_issue"/ }, + inbox: { filters: { readCondition, status: "todo,in_progress,in_review,blocked", includeBlockedInboxAttention: true, + sortField: "updated", sortDir: "desc", limit: 50 }, statement: /"status_decisions"."reason_code"/ }, + }; + const arms = { baseline: baseline.issueService(db), treatment: issueService(db) }; + const results: Record = {}; + const stats = (values: number[]) => { + const sorted = [...values].sort((a, b) => a - b); + return { medianMs: Number(sorted[Math.floor(sorted.length / 2)]!.toFixed(2)), minMs: Number(sorted[0]!.toFixed(2)), + maxMs: Number(sorted[sorted.length - 1]!.toFixed(2)), samplesMs: values.map((value) => Number(value.toFixed(2))) }; + }; + for (const [name, { filters, statement }] of Object.entries(cases)) { + 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, `${name}/${arm}: complete row payloads must match`); + } + } + const resultIds = new Set((reference as Array<{ id: string }>).map((row) => row.id)); + if (name === "busy") { + for (const id of [...recorded, ...legacyRun, ...activityOnly]) assert.ok(resultIds.has(id), `busy source must list ${id}`); + for (const id of decoys) assert.ok(!resultIds.has(id), `busy source must not list decoy ${id}`); + } + const plans: Record = {}; + for (const arm of ["baseline", "treatment"] as const) { + capture = []; + await arms[arm].list(companyId, filters); + const matching = capture.filter((query) => statement.test(query.sql)); + capture = null; + assert.ok(matching.length > 0, `${arm}: list statement captured`); + let executionMs = 0, planningMs = 0, sharedBuffers = 0; + const statementPlans = []; + for (const query of matching) { + const [row] = await client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params); + const plan = row["QUERY PLAN"][0]; + executionMs += plan["Execution Time"]; + planningMs += plan["Planning Time"]; + sharedBuffers += plan.Plan["Shared Hit Blocks"] + plan.Plan["Shared Read Blocks"]; + statementPlans.push(plan); + } + plans[arm] = { statements: matching.length, params: matching.reduce((sum, query) => sum + query.params.length, 0), + executionMs, planningMs, sharedBuffers, plans: statementPlans }; + } + results[name] = { + resultRows: resultIds.size, + baseline: stats(times.baseline), treatment: stats(times.treatment), + speedup: Number((stats(times.baseline).medianMs / stats(times.treatment).medianMs).toFixed(1)), + listStatement: Object.fromEntries(Object.entries(plans).map(([arm, value]) => { + const { statements, params, executionMs, planningMs, sharedBuffers } = value as + { statements: number; params: number; executionMs: number; planningMs: number; sharedBuffers: number }; + return [arm, { statements, params, executionMs: Number(executionMs.toFixed(3)), planningMs: Number(planningMs.toFixed(3)), sharedBuffers }]; + })), + }; + if (evidenceDir) { + await fs.mkdir(evidenceDir, { recursive: true }); + await fs.writeFile(path.join(evidenceDir, `plans-${name}.json`), JSON.stringify(plans, null, 2)); + } + } + const report = { + baselineRevision, treatmentRevision: git(["rev-parse", "HEAD"]).trim(), treatmentDirty: git(["status", "--porcelain", "--", ...modulePaths]).trim() !== "", + settings, fixture: { ...counts, fixtureMs: Math.round(fixtureMs) }, cases: results, + }; + if (evidenceDir) await fs.writeFile(path.join(evidenceDir, "summary.json"), JSON.stringify(report, 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__/blocked-inbox-review-wait-scope.test.ts b/server/src/__tests__/blocked-inbox-review-wait-scope.test.ts new file mode 100644 index 00000000000..824e22088a0 --- /dev/null +++ b/server/src/__tests__/blocked-inbox-review-wait-scope.test.ts @@ -0,0 +1,81 @@ +import { describe, expect, it } from "vitest"; +import { blockedInboxReviewWaitIssueIds } from "../services/issues.js"; +import { + classifyIssueGraphLiveness, + type IssueGraphLivenessInput, + type IssueLivenessIssueInput, +} from "../services/recovery/issue-graph-liveness.js"; + +const companyId = "company-1"; +const agentId = "agent-1"; + +function issue(id: string, status: string): IssueLivenessIssueInput { + return { id, companyId, identifier: id.toUpperCase(), title: id, status, assigneeAgentId: agentId }; +} + +// Rows are the page. Off-page review tasks also wait for an external reply. +const issues = [ + issue("row-blocked", "blocked"), + issue("review-blocker", "in_review"), + issue("row-review", "in_review"), + issue("row-chain", "blocked"), + issue("middle", "blocked"), + issue("deep-review", "in_review"), + issue("row-cycle", "blocked"), + issue("cycle-peer", "blocked"), + issue("row-todo", "todo"), + issue("off-page-review", "in_review"), + issue("off-page-blocked", "blocked"), + issue("off-page-blocker", "in_review"), +]; +const relation = (blockerIssueId: string, blockedIssueId: string) => ({ companyId, blockerIssueId, blockedIssueId }); +const relations = [ + relation("review-blocker", "row-blocked"), + relation("middle", "row-chain"), + relation("deep-review", "middle"), + relation("cycle-peer", "row-cycle"), + relation("row-cycle", "cycle-peer"), + relation("off-page-blocker", "off-page-blocked"), + // A blocker shared with an off-page task stays in the rows' closure. + relation("review-blocker", "off-page-blocked"), +]; +const rowIds = ["row-blocked", "row-review", "row-chain", "row-cycle", "row-todo"]; +const allWaits = issues.filter((entry) => entry.status === "in_review") + .map((entry) => ({ id: `wait-${entry.id}`, companyId, issueId: entry.id, status: "waiting" })); + +function rowFindings(externalResponseWaits: IssueGraphLivenessInput["externalResponseWaits"]) { + const findings = classifyIssueGraphLiveness({ + issues, + relations, + agents: [{ id: agentId, companyId, name: "Coder", role: "engineer", status: "active", reportsTo: null }], + externalResponseWaits, + now: new Date("2026-11-10T00:00:00Z"), + }); + return findings.filter((finding) => rowIds.includes(finding.issueId)) + .map(({ issueId, state, recoveryIssueId }) => ({ issueId, state, recoveryIssueId })) + .sort((left, right) => left.issueId.localeCompare(right.issueId)); +} + +describe("blocked inbox review wait scope", () => { + it("reads waits only for in-review rows and in-review blockers that the rows reach", () => { + expect(blockedInboxReviewWaitIssueIds(rowIds, issues, relations).sort()) + .toEqual(["deep-review", "review-blocker", "row-review"]); + }); + + it("gives the rows the same findings as waits for every in-review task", () => { + const narrowedIds = new Set(blockedInboxReviewWaitIssueIds(rowIds, issues, relations)); + const narrowed = allWaits.filter((wait) => narrowedIds.has(wait.issueId)); + expect(narrowed.length).toBeLessThan(allWaits.length); + expect(rowFindings(narrowed)).toEqual(rowFindings(allWaits)); + // The waits matter: without them, the review rows and blockers have no action path. + expect(rowFindings([])).not.toEqual(rowFindings(allWaits)); + for (const wait of narrowed) { + expect(rowFindings(narrowed.filter((other) => other !== wait))).not.toEqual(rowFindings(allWaits)); + } + }); + + it("returns nothing for rows without in-review tasks in their blocker closure", () => { + expect(blockedInboxReviewWaitIssueIds(["row-todo", "missing"], issues, relations)).toEqual([]); + expect(blockedInboxReviewWaitIssueIds([], issues, relations)).toEqual([]); + }); +}); diff --git a/server/src/__tests__/issue-created-from-routes.test.ts b/server/src/__tests__/issue-created-from-routes.test.ts index c694460a3c1..e6efa99f5f7 100644 --- a/server/src/__tests__/issue-created-from-routes.test.ts +++ b/server/src/__tests__/issue-created-from-routes.test.ts @@ -79,9 +79,31 @@ describePostgres("tasks created from an issue", () => { { companyId, title: "Wrong run company", originRunId: foreignRunId }, { companyId: otherCompanyId, title: "Wrong issue company", originRunId: legacyRunId }, ]); + // Recorded origins win over runs and activity in both directions. + const recordedId = randomUUID(); + const recordedElsewhereId = randomUUID(); + const recordedActivityElsewhereId = randomUUID(); + const otherRunActivityId = randomUUID(); + const nonCanonicalActivityId = randomUUID(); + expected.add(recordedId); + await db.insert(issues).values([ + { id: recordedId, companyId, title: "Recorded origin, unrelated run", originIssueId: sourceId, originRunId: unrelatedRunId }, + { id: recordedElsewhereId, companyId, title: "Recorded elsewhere, source run", originIssueId: otherSourceId, originRunId: legacyRunId }, + { id: recordedActivityElsewhereId, companyId, title: "Recorded elsewhere, source activity", originIssueId: otherSourceId }, + { id: otherRunActivityId, companyId, title: "Unrelated origin run, source activity", originRunId: unrelatedRunId }, + { id: nonCanonicalActivityId, companyId, title: "Source activity with a non-canonical ID" }, + { companyId: otherCompanyId, title: "Foreign recorded origin", originIssueId: foreignSourceId }, + ]); await db.insert(activityLog).values([ { companyId, actorType: "agent", actorId: agentId, runId: legacyRunId, action: "issue.child_created", entityType: "issue", entityId: historicalId }, { companyId, actorType: "agent", actorId: agentId, runId: legacyRunId, action: "issue.comment_added", entityType: "issue", entityId: commentedId }, + { companyId, actorType: "agent", actorId: agentId, runId: legacyRunId, action: "issue.created", entityType: "issue", entityId: recordedActivityElsewhereId }, + { companyId, actorType: "agent", actorId: agentId, runId: legacyRunId, action: "issue.created", entityType: "issue", entityId: otherRunActivityId }, + { companyId, actorType: "agent", actorId: agentId, runId: legacyRunId, action: "issue.created", entityType: "issue", entityId: nonCanonicalActivityId.toUpperCase() }, + { companyId, actorType: "agent", actorId: agentId, runId: legacyRunId, action: "issue.created", entityType: "issue", entityId: "not-a-task-id" }, + // The source's own creation by its run is not a created task. + { companyId, actorType: "agent", actorId: agentId, runId: legacyRunId, action: "issue.created", entityType: "issue", entityId: sourceId }, + { companyId: otherCompanyId, actorType: "agent", actorId: foreignAgentId, runId: legacyRunId, action: "issue.created", entityType: "issue", entityId: commentedId }, ]); }, 30_000); afterAll(async () => { await database?.cleanup(); }); diff --git a/server/src/__tests__/issue-list-review-wait-authorization.integration.test.ts b/server/src/__tests__/issue-list-review-wait-authorization.integration.test.ts new file mode 100644 index 00000000000..9f330422514 --- /dev/null +++ b/server/src/__tests__/issue-list-review-wait-authorization.integration.test.ts @@ -0,0 +1,98 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { chatConversations, chatDeliveries, chatMessageLinks, createDb, heartbeatRuns, + issueComments, issues, statusDecisions, type Db } from "@paperclipai/db"; +import { seedBoundChatAuthorization } from "./helpers/bound-chat-authorization.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { issueService } from "../services/issues.js"; +import { listExternalReviewWaits } from "../services/native-runtime/external-review-wait.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const describePostgres = support.supported ? describe : describe.skip; + +// The bound-run message read of chat authorization. The inherited GitHub review +// read also starts from chat_message_links, but it does not join deliveries. +const boundRunLinkStatement = /from "chat_message_links" inner join "chat_deliveries"/i; + +describePostgres("issue list review-wait authorization", () => { + let temporary: Awaited>; + let db: Db; + let f: Awaited>; + let statements: string[] | null = null; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("review-wait-authorization-"); + db = createDb(temporary.connectionString); + const client = db.$client as unknown as { unsafe: (query: string, ...rest: unknown[]) => unknown }; + const unsafe = client.unsafe.bind(client); + client.unsafe = (query: string, ...rest: unknown[]) => { + statements?.push(query); + return unsafe(query, ...rest); + }; + f = await seedBoundChatAuthorization(db); + // A second waiting conversation, on the other in-review issue, from the + // same linked person. + const otherConversationId = randomUUID(), otherCommentId = randomUUID(), deliveryId = randomUUID(); + await db.insert(chatConversations).values({ id: otherConversationId, companyId: f.companyId, + endpointId: f.endpointId, resourceId: f.resourceId, issueId: f.otherIssueId, + externalConversationId: "channel-1", externalThreadId: "thread-2", externalLabel: "Other thread", state: "active" }); + await db.insert(issueComments).values({ id: otherCommentId, companyId: f.companyId, issueId: f.otherIssueId, + authorType: "user", authorUserId: f.userId, body: "Another request" }); + await db.insert(chatDeliveries).values({ id: deliveryId, companyId: f.companyId, endpointId: f.endpointId, + conversationId: otherConversationId, principalId: f.principalId, providerEventId: deliveryId, + deduplicationKey: deliveryId, eventKind: "message", normalizedEvent: {}, state: "processed", processedAt: new Date() }); + await db.insert(chatMessageLinks).values({ companyId: f.companyId, endpointId: f.endpointId, + conversationId: otherConversationId, deliveryId, commentId: otherCommentId, + providerMessageId: otherCommentId, direction: "inbound" }); + // Each issue waits on the reply that its succeeded native run asked for. + const otherRunId = randomUUID(); + await db.insert(heartbeatRuns).values([ + { id: f.runId, companyId: f.companyId, agentId: f.agentId, status: "succeeded", runtimeMode: "native", + nativeIssueId: f.issueId, contextSnapshot: f.context }, + { id: otherRunId, companyId: f.companyId, agentId: f.agentId, status: "succeeded", runtimeMode: "native", + nativeIssueId: f.otherIssueId, contextSnapshot: { ...f.context, wakeCommentIds: [otherCommentId] } }, + ]); + for (const [issueId, runId] of [[f.issueId, f.runId], [f.otherIssueId, otherRunId]] as const) { + const decisionId = randomUUID(); + // The decision's assessment chain is not part of this read. + await db.transaction(async (tx) => { + await tx.execute(sql`set local session_replication_role = replica`); + await tx.insert(statusDecisions).values({ id: decisionId, companyId: f.companyId, issueId, runId, + assessmentId: randomUUID(), decisionVersion: 1, policyVersion: "test", fromStatus: "in_progress", + toStatus: "in_review", reasonCode: "external_chat_response_waiting", decisionJson: {}, + decisionDigest: decisionId, applicationState: "applied", appliedAt: new Date() }); + }); + await db.update(issues).set({ lastStatusDecisionId: decisionId }).where(eq(issues.id, issueId)); + } + }, 60_000); + + afterAll(async () => { + try { await db?.$client.end({ timeout: 1 }); } + finally { await temporary?.cleanup(); } + }); + + async function boundRunLinkReads(read: () => Promise) { + statements = []; + try { + const result = await read(); + return { result, reads: statements.filter((statement) => boundRunLinkStatement.test(statement)).length }; + } finally { + statements = null; + } + } + + it("authorizes every wait of a page with one message read", async () => { + const { result, reads } = await boundRunLinkReads(() => + listExternalReviewWaits(db, f.companyId, [f.issueId, f.otherIssueId])); + expect(result.map((wait) => wait.issueId).sort()).toEqual([f.issueId, f.otherIssueId].sort()); + expect(reads).toBe(1); + }); + + it("reads the waits of an issue-list request once, for both attention maps", async () => { + const { result, reads } = await boundRunLinkReads(() => + issueService(db).list(f.companyId, { includeBlockedInboxAttention: true })); + expect(result.map((row) => row.id).sort()).toEqual([f.issueId, f.otherIssueId].sort()); + expect(reads).toBe(1); + }); +}); diff --git a/server/src/services/issue-creation-origin.ts b/server/src/services/issue-creation-origin.ts index 89da22e76c1..82f170a4a97 100644 --- a/server/src/services/issue-creation-origin.ts +++ b/server/src/services/issue-creation-origin.ts @@ -4,16 +4,19 @@ import { activityLog, heartbeatRuns, issues } from "@paperclipai/db"; import { isUuidLike } from "@paperclipai/shared"; import { unprocessable } from "../errors.js"; +// The canonical text of a UUID. Only this form can equal `issues.id::text`. +const CANONICAL_UUID_PATTERN = "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"; + /** Creation provenance is independent of the task's current parent or project. */ export function createdFromIssueCondition(companyId: string, sourceIssueId: string) { if (!isUuidLike(sourceIssueId)) { throw unprocessable("createdFromIssueId must be a UUID"); } const source = alias(issues, "creation_source_issue"); - // Resolve source runs independently of the candidate tasks so the database - // can reuse this set, rather than scan all company runs for each task. + const created = alias(issues, "creation_candidate_issue"); + // The source issue's runs, through heartbeat_runs_company_origin_issue_idx. const originatingRuns = sql` - SELECT ${heartbeatRuns.id}::text FROM ${heartbeatRuns} + SELECT ${heartbeatRuns.id} FROM ${heartbeatRuns} INNER JOIN ${issues} AS ${source} ON ${source.id} = ${sourceIssueId} AND ${source.companyId} = ${companyId} WHERE ${heartbeatRuns.companyId} = ${companyId} @@ -26,19 +29,36 @@ export function createdFromIssueCondition(companyId: string, sourceIssueId: stri `; // A recorded origin task wins; run and activity fallbacks apply only to rows // without one, so the list agrees with the detail's createdFrom. + // + // Each branch starts from an index (origin task, source runs, run activity) + // so the list reads only the created tasks. An OR over the list's own rows + // would test every task of the company, with the list's other per-row + // conditions, before it could exclude one. return sql` ${issues.id} <> ${sourceIssueId} - AND ( - ${issues.originIssueId} = ${sourceIssueId} - OR (${issues.originIssueId} IS NULL AND ${issues.originRunId} IN (${originatingRuns})) - OR (${issues.originIssueId} IS NULL AND ${issues.originRunId} IS NULL AND EXISTS ( - SELECT 1 FROM ${activityLog} - WHERE ${activityLog.companyId} = ${companyId} - AND ${activityLog.entityType} = 'issue' - AND ${activityLog.entityId} = ${issues.id}::text - AND ${activityLog.action} IN ('issue.created', 'issue.child_created') - AND ${activityLog.runId}::text IN (${originatingRuns}) - )) + AND ${issues.id} IN ( + SELECT ${created.id} FROM ${issues} AS ${created} + WHERE ${created.companyId} = ${companyId} + AND ${created.originIssueId} = ${sourceIssueId} + UNION ALL + SELECT ${created.id} FROM ${issues} AS ${created} + WHERE ${created.companyId} = ${companyId} + -- The predicate of issues_company_legacy_origin_run_idx. + AND ${created.originIssueId} is null and ${created.originRunId} is not null + AND ${created.originRunId} IN (SELECT source_run.id::text FROM (${originatingRuns}) AS source_run) + UNION ALL + SELECT ${created.id} FROM ${activityLog} + INNER JOIN ${issues} AS ${created} + ON ${created.id} = CASE WHEN ${activityLog.entityId} ~ ${CANONICAL_UUID_PATTERN} + THEN ${activityLog.entityId}::uuid END + AND ${created.id}::text = ${activityLog.entityId} + WHERE ${activityLog.companyId} = ${companyId} + AND ${activityLog.entityType} = 'issue' + AND ${activityLog.action} IN ('issue.created', 'issue.child_created') + AND ${activityLog.runId} IN (SELECT source_run.id FROM (${originatingRuns}) AS source_run) + AND ${created.companyId} = ${companyId} + AND ${created.originIssueId} IS NULL + AND ${created.originRunId} IS NULL ) `; } diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index a24ce2e3c2a..565da8152da 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -5776,6 +5776,38 @@ function compareBlockedInboxRows( return right.id.localeCompare(left.id); } +/** + * The in-review issues whose external response waits can change a finding + * for the given rows. The caller reads only the rows' findings, and the + * liveness classifier reads a wait only for a row, or for a blocker that it + * reaches from a row through blocker relations. This returns every in-review + * issue in that blocker closure, which includes all of those issues. + */ +export function blockedInboxReviewWaitIssueIds( + rowIssueIds: readonly string[], + graphIssues: ReadonlyArray<{ id: string; status: string }>, + graphRelations: ReadonlyArray<{ blockerIssueId: string; blockedIssueId: string }>, +): string[] { + const statusById = new Map(graphIssues.map((issue) => [issue.id, issue.status])); + const blockersByBlockedId = new Map(); + for (const relation of graphRelations) { + const blockers = blockersByBlockedId.get(relation.blockedIssueId) ?? []; + blockers.push(relation.blockerIssueId); + blockersByBlockedId.set(relation.blockedIssueId, blockers); + } + const reached = new Set(); + const pending = [...rowIssueIds]; + while (pending.length > 0) { + const issueId = pending.pop()!; + if (reached.has(issueId)) continue; + reached.add(issueId); + for (const blockerId of blockersByBlockedId.get(issueId) ?? []) { + if (!reached.has(blockerId)) pending.push(blockerId); + } + } + return [...reached].filter((issueId) => statusById.get(issueId) === "in_review"); +} + async function listIssueBlockedInboxAttentionMap( dbOrTx: any, companyId: string, @@ -6040,7 +6072,9 @@ async function listIssueBlockedInboxAttentionMap( }); const findings = classifyIssueGraphLiveness({ - externalResponseWaits: await readExternalReviewWaits(graphIssues.filter(issue => issue.status === "in_review").map(issue => issue.id)), + externalResponseWaits: await readExternalReviewWaits( + blockedInboxReviewWaitIssueIds(rowIssueIds, graphIssues, graphRelations), + ), issues: graphIssues.map((issue) => ({ id: issue.id, companyId: issue.companyId,