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,