From 765ccd8691e5216411608a7c11c9c1e9bcc81f8a Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 10:44:51 -0600 Subject: [PATCH] perf(chat): index chat-admission checks by company and payload issue id (#20562) 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. > - Chat connections save each external message as a `chat_actions` row before the server admits it as an agent wakeup. > - The server checks if an issue holds chat input that it did not admit. Recovery skips such an issue. Manual task start refuses it. > - On a hosted instance, the recovery read with this check used the most total database time: 488 calls at about 1 s each. The manual-start check ran 155 times at about 0.9 s. > - The check matches `chat_actions.payload->>'issueId'` against the issue. No index covers this value, so each call reads the full `chat_actions` table and its payloads. > - This pull request adds one concurrent partial index for the check. The check SQL does not change. > - The benefit is less background database load, so request reads wait less. The results stay the same. ## Linked Issues or Issue Description **What existing behavior does this improve?** The chat-admission check `unadmittedChatWakeupCondition`. Recovery uses it in the per-issue candidate read. Manual task start uses it in the held-input check. **Subsystem affected** Server chat admission, issue recovery, manual task start, and the database schema. **Current behavior** The check finds `inbound_wakeup` actions by `company_id` and `payload->>'issueId'`. Two nested checks find `failed_run_retry` actions and later `inbound_wakeup` actions on the same company and payload issue ID. `chat_actions` has no index on `company_id` or on the payload issue ID. Each call reads the full table and reads each payload. On a hosted instance, the two statements used about 630 s of database time in one statistics window. **Proposed behavior** The check and its nested checks read a small set of rows from one index. The admission result does not change. **Reason and benefit** The hosted database had many concurrent active queries. Request reads waited behind background work such as this check. **Breaking changes** None. The migration adds an index only. ## What Changed - Migration `0458_chat_admission_issue_index` builds `chat_actions_company_issue_kind_idx` with `CREATE INDEX CONCURRENTLY IF NOT EXISTS`. It runs through the existing top-level statement path, which repairs an invalid index on retry. - Keys: `(company_id, payload->>'issueId', kind, status)`. - Partial: `kind IN ('inbound_wakeup', 'failed_run_retry')`. - One index serves the outer input lookup, the failed-run retry lookup, and the later-input lookup. All three match on the same company and payload issue ID. - The Drizzle schema declares the index. The snapshot chain moves to `0458` and keeps five snapshots. - New regression test `server/src/__tests__/chat-admission-condition.test.ts` pins the admission semantics on real PostgreSQL: - Pending and failed input without a receipt holds the issue. Received, processed and cancelled input does not. Other action kinds do not. - A skipped receipt, a receipt for another agent, or a receipt in another company does not admit input. A failed or cancelled run still counts as admission. - The issue ID is compared as company-scoped text. Uppercase, array, null, other-issue and other-company values do not match. - Only an admitted retry in the same conversation and issue that lists the action releases skipped input. - Only later admitted input in the same conversation, issue and session generation releases a failed input. Pending input stays held. - The check reads `chat_actions` only through the new index. - `scripts/benchmark-chat-admission-index.ts` measures both caller statements with and without the index and fails if the rows differ. - `scripts/benchmark-chat-admission-index-migration.ts` times the migration through the normal migration runner. - `doc/DATABASE.md` documents the index and the benchmark commands. ## Verification Local SQL measurements on synthetic data. PostgreSQL 18.1 (embedded), default settings. The fixture has 100,995 `chat_actions` rows (662 MB): 60,000 GitHub actions with 6 KiB incompressible payloads, 2,000 issues with 20 admitted inbound messages each (3 KiB payloads), and 1,000 failed-run retries. Each case has 2 warmups and 9 samples of `EXPLAIN ANALYZE` execution time. Runs alternate: no index, index, no index, index. | Caller statement | Case | No index, median (range) | Index, median (range) | | --- | --- | --- | --- | | Manual-start held check | quiet issue | 3,093 ms (2,150–3,519) / 3,586 ms (2,375–11,235) | 0.036 ms (0.034–0.052) / 0.040 ms (0.033–0.047) | | Manual-start held check | admitted input | 3,555 ms (3,310–8,080) / 5,205 ms (3,353–9,525) | 0.049 ms / 0.052 ms | | Manual-start held check | unadmitted input | 3,002 ms (2,864–3,221) / 4,179 ms (3,200–6,585) | 0.051 ms / 0.044 ms | | Manual-start held check | failed, later admitted | 3,586 ms / 3,722 ms | 0.066 ms / 0.054 ms | | Manual-start held check | admitted retry | 2,733 ms / 4,953 ms | 0.049 ms / 0.053 ms | | Recovery candidate read | quiet issue | 43 ms (41–48) / 96 ms (44–219) | 0.047 ms / 0.055 ms | | Recovery candidate read | admitted input | 203 ms (48–575) / 174 ms (44–314) | 0.055 ms / 0.058 ms | | Recovery candidate read | unadmitted input | 66 ms / 144 ms | 0.053 ms / 0.054 ms | - Buffers per call fall from about 40,000 shared hits to 5–23. - Locally, the recovery read without the index uses a hashed full scan of `chat_actions`. The held check uses a full scan per call. On the hosted instance both took about 1 s per call. The local fixture is smaller than the hosted table. - Correctness: complete positional rows match across all four runs for all six cases (quiet, admitted, unadmitted, failed then admitted, retried, other-company decoy). The script asserts the expected held result for each case. - Migration timing: 600,001 `chat_actions` rows (1,759 MB), 300,000 of them in the index scope. First run through `applyPendingMigrations`: 4.2 s. Rerun after the history row is removed: 0.5 s. Concurrent build alone: 3.4 s. Index size: 30 MB. Commands: ```sh pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-admission-index.ts OTHER_PAYLOAD_KIB=2 pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-admission-index-migration.ts cd server && npx vitest run src/__tests__/chat-admission-condition.test.ts # 6 passed pnpm --filter @paperclipai/db check:migrations # passed pnpm --filter @paperclipai/server typecheck # passed ``` ## Risks - The migration builds one index concurrently. It does not block writes. On the 1.76 GB local fixture it took about 4 s. A larger hosted table takes longer, but the build does not lock out reads or writes. - Writes to `inbound_wakeup` and `failed_run_retry` actions maintain one more index. Status updates on these rows were already not HOT updates, because other partial indexes use `status`. - The planner has no expression statistics for a partial index. The new test pins that the check uses the index on an analyzed table with unrelated rows. - The check SQL does not change, so admission behavior does not change. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (Anthropic), model ID `claude-opus-5-5`, used through Claude Code with tool use and code execution. ## 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 - [x] All Paperclip CI gates are green - [x] 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 | 14 ++ packages/db/src/client.ts | 1 + .../0458_chat_admission_issue_index.sql | 1 + ...{0453_snapshot.json => 0458_snapshot.json} | 185 ++++++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/chat_channels.ts | 6 + ...enchmark-chat-admission-index-migration.ts | 82 +++++++ scripts/benchmark-chat-admission-index.ts | 216 ++++++++++++++++++ .../chat-admission-condition.test.ts | 171 ++++++++++++++ 9 files changed, 681 insertions(+), 2 deletions(-) create mode 100644 packages/db/src/migrations/0458_chat_admission_issue_index.sql rename packages/db/src/migrations/meta/{0453_snapshot.json => 0458_snapshot.json} (99%) create mode 100644 scripts/benchmark-chat-admission-index-migration.ts create mode 100644 scripts/benchmark-chat-admission-index.ts create mode 100644 server/src/__tests__/chat-admission-condition.test.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 461c9ca63af..5b7804d1ee5 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -447,6 +447,20 @@ through it instead of reading the agent's whole run history. To measure it: pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-native-session-run-index.ts ``` +Chat-admission checks use `chat_actions_company_issue_kind_idx` from migration +0458, which builds it concurrently. The check decides whether an issue holds +external chat input that was never admitted: recovery skips such an issue and +manual task start refuses it. The index keys `inbound_wakeup` and +`failed_run_retry` actions by company, `payload->>'issueId'`, kind and status. +It serves the outer input lookup and the nested retry and later-input lookups, +which all match on the same company and payload issue ID. Without it, each check +reads the full `chat_actions` table. The check SQL is unchanged. Local proof: + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-admission-index.ts +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-admission-index-migration.ts +``` + The chat wake-notice sweep compares UUID columns directly against guarded, canonical lowercase UUID payload values. Malformed and noncanonical payloads still match nothing. The query can use existing primary keys without adding a diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index add40c1c478..320df7b461b 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -620,6 +620,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0455_pending_budget_cancellation_index.sql", "0456_chat_review_lookup_indexes.sql", "0457_native_session_run_index.sql", + "0458_chat_admission_issue_index.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0458_chat_admission_issue_index.sql b/packages/db/src/migrations/0458_chat_admission_issue_index.sql new file mode 100644 index 00000000000..f2b17c11d07 --- /dev/null +++ b/packages/db/src/migrations/0458_chat_admission_issue_index.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_company_issue_kind_idx" ON "chat_actions" USING btree ("company_id",("payload"->>'issueId'),"kind","status") WHERE "chat_actions"."kind" IN ('inbound_wakeup', 'failed_run_retry'); diff --git a/packages/db/src/migrations/meta/0453_snapshot.json b/packages/db/src/migrations/meta/0458_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0453_snapshot.json rename to packages/db/src/migrations/meta/0458_snapshot.json index 8082bdb3ed7..7964760c56a 100644 --- a/packages/db/src/migrations/meta/0453_snapshot.json +++ b/packages/db/src/migrations/meta/0458_snapshot.json @@ -1,6 +1,6 @@ { - "id": "d54bfeb4-dedb-4f8b-9195-a4d4aa67d71f", - "prevId": "64e8be1e-e0cb-4cd8-85e0-0fdcce5eafde", + "id": "3347b37d-b35b-4357-b08c-a78c017086ac", + "prevId": "109f5f96-1597-40da-9e08-6ed488f8ad27", "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": [ @@ -10955,6 +10989,28 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_deliveries_github_automatic_delivery_idx": { + "name": "chat_deliveries_github_automatic_delivery_idx", + "columns": [ + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"normalized_event\"->'githubAutomatic'->'context'->>'deliveryId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "(\"chat_deliveries\".\"normalized_event\"->'githubAutomatic'->'context'->>'deliveryId') is not null", + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -12393,6 +12449,44 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_message_links_company_comment_idx": { + "name": "chat_message_links_company_comment_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "comment_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_message_links\".\"comment_id\" is not null", + "concurrently": true, + "method": "btree", + "with": {} + }, + "chat_message_links_delivery_idx": { + "name": "chat_message_links_delivery_idx", + "columns": [ + { + "expression": "delivery_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_message_links\".\"delivery_id\" is not null", + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -34552,6 +34646,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "heartbeat_run_events_retry_exhausted_idx": { + "name": "heartbeat_run_events_retry_exhausted_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_run_events\".\"event_type\" = 'lifecycle' and \"heartbeat_run_events\".\"message\" like 'Bounded retry exhausted%'", + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -35393,6 +35515,22 @@ "method": "btree", "with": {} }, + "heartbeat_runs_pending_budget_cancellation_idx": { + "name": "heartbeat_runs_pending_budget_cancellation_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "jsonb_typeof(\"heartbeat_runs\".\"result_json\"->'startupCancellation'->'budgetEnforcement') = 'object'\n and \"heartbeat_runs\".\"result_json\"->'startupCancellation'->'budgetEnforcement'->>'deliveredAt' is null", + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_tree_hold_suppression_idx": { "name": "heartbeat_runs_tree_hold_suppression_idx", "columns": [ @@ -35431,6 +35569,27 @@ "method": "btree", "with": {} }, + "heartbeat_runs_company_wake_comment_idx": { + "name": "heartbeat_runs_company_wake_comment_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"context_snapshot\"->>'wakeCommentId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": true, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ @@ -35583,6 +35742,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": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 006da97123e..28cd5ff66b5 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3193,6 +3193,13 @@ "when": 1794323383634, "tag": "0457_native_session_run_index", "breakpoints": true + }, + { + "idx": 458, + "version": "7", + "when": 1794326875077, + "tag": "0458_chat_admission_issue_index", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/chat_channels.ts b/packages/db/src/schema/chat_channels.ts index 3a0a147d7b8..49ecc309ea2 100644 --- a/packages/db/src/schema/chat_channels.ts +++ b/packages/db/src/schema/chat_channels.ts @@ -700,6 +700,12 @@ export const chatActions = pgTable( table.endpointId, table.providerActionId, ), + // Chat-admission checks read an issue's inbound input and its failed-run + // retries by company and payload issue id. + index("chat_actions_company_issue_kind_idx") + .on(table.companyId, sql`(${table.payload}->>'issueId')`, table.kind, table.status) + .where(sql`${table.kind} IN ('inbound_wakeup', 'failed_run_retry')`) + .concurrently(), index("chat_actions_github_stale_head_idx").on(table.companyId, table.endpointId, sql`(${table.payload}->'event'->>'deliveryId')`).where(sql`${table.kind} = 'github_review_check' and ${table.status} = 'cancelled' and ${table.result}->>'code' = 'stale_head'`), diff --git a/scripts/benchmark-chat-admission-index-migration.ts b/scripts/benchmark-chat-admission-index-migration.ts new file mode 100644 index 00000000000..ab0c16c4804 --- /dev/null +++ b/scripts/benchmark-chat-admission-index-migration.ts @@ -0,0 +1,82 @@ +/** + * Times migration 0458 through the normal migration runner on a large + * synthetic chat_actions table, including a rerun after the history row is removed. + * + * pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-admission-index-migration.ts + * + * Optional sizes: OTHER_ACTIONS (default 300000), OTHER_PAYLOAD_KIB (4), + * WAKEUPS (300000), WAKEUP_PAYLOAD_KIB (2). + */ +import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; +import { readFileSync } from "node:fs"; +import { applyPendingMigrations, createDb, startEmbeddedPostgresTestDatabase } from "../packages/db/src/index.js"; +import { seedGitHubReviewWait } from "../server/src/__tests__/helpers/github-review-wait.js"; + +const size = (name: string, fallback: number) => Number(process.env[name] ?? fallback); +const OTHER_ACTIONS = size("OTHER_ACTIONS", 300_000), OTHER_PAYLOAD_KIB = size("OTHER_PAYLOAD_KIB", 4); +const WAKEUPS = size("WAKEUPS", 300_000), WAKEUP_PAYLOAD_KIB = size("WAKEUP_PAYLOAD_KIB", 2); +const migration = readFileSync(new URL("../packages/db/src/migrations/0458_chat_admission_issue_index.sql", import.meta.url), "utf8"); +const migrationHash = createHash("sha256").update(migration).digest("hex"); +const indexName = migration.match(/IF NOT EXISTS "([a-z_]+)"/)![1]!; + +const keepAlive = setInterval(() => {}, 1_000); +let temporary: Awaited> | undefined; +let db: ReturnType | undefined; +try { + temporary = await startEmbeddedPostgresTestDatabase("chat-admission-index-migration-"); + db = createDb(temporary.connectionString); + const sql = db.$client; + await sql.unsafe(`DROP INDEX IF EXISTS "${indexName}"`); + const { companyId, agentId, endpointId, conversationId } = await seedGitHubReviewWait(db); + await sql.unsafe("create extension if not exists pgcrypto"); + const blob = (kib: string) => `(select string_agg(encode(gen_random_bytes(768), 'base64'), '') + from generate_series(1, ${kib}) s(i) where g > 0)`; + + const loadStart = performance.now(); + await sql.unsafe(`insert into chat_actions (company_id, endpoint_id, kind, provider_action_id, status, payload) + select $1::uuid, $2::uuid, 'github_review_check', 'other-' || g, 'processed', + jsonb_build_object('event', jsonb_build_object('body', ${blob("$3::int")})) + from generate_series(1, $4::int) g`, [companyId, endpointId, OTHER_PAYLOAD_KIB, OTHER_ACTIONS]); + await sql.unsafe(`insert into chat_actions (company_id, endpoint_id, conversation_id, kind, provider_action_id, status, payload) + select $1::uuid, $2::uuid, $3::uuid, case when g % 50 = 0 then 'failed_run_retry' else 'inbound_wakeup' end, + 'wake-' || g, 'processed', jsonb_build_object('issueId', gen_random_uuid()::text, 'agentId', $4::text, + 'text', ${blob("$5::int")}) + from generate_series(1, $6::int) g`, [companyId, endpointId, conversationId, agentId, WAKEUP_PAYLOAD_KIB, WAKEUPS]); + await sql.unsafe("vacuum analyze chat_actions"); + const loadMs = performance.now() - loadStart; + const tableSizes = await sql.unsafe(`select relname as table, n_live_tup::bigint::text as rows, + pg_size_pretty(pg_total_relation_size(relid)) as total_size from pg_stat_user_tables + where relname = 'chat_actions'`); + + async function migrate() { + // Remove only this migration's history row (the runner records the file's sha256). + await sql.unsafe("delete from drizzle.__drizzle_migrations where hash = $1", [migrationHash]); + const start = performance.now(); + await applyPendingMigrations(temporary!.connectionString); + return performance.now() - start; + } + // First run builds the index; the rerun finds it valid and skips it. + const firstRunMs = await migrate(); + const valid = await sql.unsafe(`select c.relname, i.indisvalid, pg_size_pretty(pg_relation_size(c.oid)) as size + from pg_index i join pg_class c on c.oid = i.indexrelid where c.relname = $1`, [indexName]); + assert.equal(valid.length, 1); + assert.ok(valid[0]!.indisvalid); + const rerunMs = await migrate(); + await sql.unsafe(`DROP INDEX CONCURRENTLY "${indexName}"`); + const buildStart = performance.now(); + await sql.unsafe(migration.trim()); + const buildMs = performance.now() - buildStart; + console.log(JSON.stringify({ + postgresVersion: (await sql.unsafe("select version()"))[0]!.version, + fixture: { synthetic: true, otherActions: OTHER_ACTIONS, otherPayloadKiB: OTHER_PAYLOAD_KIB, wakeups: WAKEUPS, + wakeupPayloadKiB: WAKEUP_PAYLOAD_KIB, loadMs: Math.round(loadMs), tableSizes }, + migration: { firstRunMs: Math.round(firstRunMs), rerunMs: Math.round(rerunMs), index: valid[0], + concurrentBuildMs: Math.round(buildMs) }, + }, null, 2)); +} finally { + try { + await db?.$client.end({ timeout: 0 }); + await temporary?.cleanup(); + } finally { clearInterval(keepAlive); } +} diff --git a/scripts/benchmark-chat-admission-index.ts b/scripts/benchmark-chat-admission-index.ts new file mode 100644 index 00000000000..4b2de1c6813 --- /dev/null +++ b/scripts/benchmark-chat-admission-index.ts @@ -0,0 +1,216 @@ +/** + * Local proof for the chat-admission lookup index (migration 0458). + * + * `unadmittedChatWakeupCondition` decides whether an issue holds external chat + * input that was never admitted. Recovery reads one issue with the condition + * negated, and manual task start reads one issue with it as an `exists` check. + * The SQL does not change; this script runs both statements with and without + * the new index and fails unless the complete rows match. + * + * pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-admission-index.ts + * + * Optional sizes: OTHER_ACTIONS (default 60000), OTHER_PAYLOAD_KIB (6), + * ISSUES (2000), WAKEUPS_PER_ISSUE (20), WAKEUP_PAYLOAD_KIB (3), RETRIES (1000). + */ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { readFileSync } from "node:fs"; +import { createRequire } from "node:module"; +import { caseHandoffs, createDb, issues, startEmbeddedPostgresTestDatabase } from "../packages/db/src/index.js"; +import { unadmittedChatWakeupCondition } from "../server/src/services/durable-chat-wakeup.js"; +import { seedGitHubReviewWait } from "../server/src/__tests__/helpers/github-review-wait.js"; + +const { and, eq, inArray, isNull, not, notExists, or, sql: dsql } = createRequire( + new URL("../server/package.json", import.meta.url))("drizzle-orm") as typeof import("drizzle-orm"); + +const size = (name: string, fallback: number) => Number(process.env[name] ?? fallback); +const OTHER_ACTIONS = size("OTHER_ACTIONS", 60_000), OTHER_PAYLOAD_KIB = size("OTHER_PAYLOAD_KIB", 6); +const ISSUES = size("ISSUES", 2_000), WAKEUPS_PER_ISSUE = size("WAKEUPS_PER_ISSUE", 20); +const WAKEUP_PAYLOAD_KIB = size("WAKEUP_PAYLOAD_KIB", 3), RETRIES = size("RETRIES", 1_000); +const migration = readFileSync(new URL("../packages/db/src/migrations/0458_chat_admission_issue_index.sql", import.meta.url), "utf8"); +const indexStatements = migration.split("--> statement-breakpoint").map(statement => statement.trim()).filter(Boolean); +const indexNames = indexStatements.map(statement => statement.match(/IF NOT EXISTS "([a-z_]+)"/)![1]!); +const git = (...args: string[]) => execFileSync("git", args, { encoding: "utf8" }).trim(); + +const keepAlive = setInterval(() => {}, 1_000); +let temporary: Awaited> | undefined; +let db: ReturnType | undefined; +try { + temporary = await startEmbeddedPostgresTestDatabase("chat-admission-index-benchmark-"); + db = createDb(temporary.connectionString); + const sql = db.$client; + await sql.unsafe("create extension if not exists pgcrypto"); + const f = await seedGitHubReviewWait(db); + const decoy = await seedGitHubReviewWait(db); + const blob = (kib: string) => `(select string_agg(encode(gen_random_bytes(768), 'base64'), '') + from generate_series(1, ${kib}) s(i) where g > 0)`; + + const loadStart = performance.now(); + // Conversations the inbound actions belong to. + await sql.unsafe(`insert into chat_conversations (company_id, endpoint_id, resource_id, issue_id, + external_conversation_id, external_thread_id, external_label) + select $1::uuid, $2::uuid, $3::uuid, $4::uuid, 'load', 'load-' || g, 'Load' from generate_series(1, 200) g`, + [f.companyId, f.endpointId, f.resourceId, f.issueId]); + await sql.unsafe(`insert into issues (company_id, title, assignee_agent_id, status) + select $1::uuid, 'Load ' || g, $2::uuid, 'in_progress' from generate_series(1, $3::int) g`, + [f.companyId, f.agentId, ISSUES]); + // Unrelated chat history with wide incompressible payloads (GitHub checks and replies). + await sql.unsafe(`insert into chat_actions (company_id, endpoint_id, kind, provider_action_id, status, payload) + select $1::uuid, $2::uuid, case when g % 2 = 0 then 'github_review_check' else 'github_response_comment' end, + 'other-' || g, 'processed', jsonb_build_object('event', jsonb_build_object('body', ${blob("$3::int")})) + from generate_series(1, $4::int) g`, [f.companyId, f.endpointId, OTHER_PAYLOAD_KIB, OTHER_ACTIONS]); + // Inbound wakeups: every issue has a history of admitted messages. + await sql.unsafe(`with i as (select id, row_number() over (order by id) rn from issues + where company_id = $1::uuid and title like 'Load %'), + c as (select id, row_number() over (order by id) - 1 rn from chat_conversations + where company_id = $1::uuid and external_conversation_id = 'load') + insert into chat_actions (company_id, endpoint_id, conversation_id, kind, provider_action_id, status, payload, created_at) + select $1::uuid, $2::uuid, c.id, 'inbound_wakeup', 'wake-' || i.rn || '-' || g, 'processed', + jsonb_build_object('issueId', i.id::text, 'agentId', $3::text, 'sessionGeneration', 1, + 'text', ${blob("$4::int")}), now() - make_interval(secs => g) + from i join c on c.rn = i.rn % 200, generate_series(1, $5::int) g`, + [f.companyId, f.endpointId, f.agentId, WAKEUP_PAYLOAD_KIB, WAKEUPS_PER_ISSUE]); + await sql.unsafe(`insert into agent_wakeup_requests (id, company_id, agent_id, source, status) + select id, company_id, $2::uuid, 'automation', 'finished' from chat_actions + where company_id = $1::uuid and kind = 'inbound_wakeup'`, [f.companyId, f.agentId]); + // Failed-run retries that skipped some older input. + await sql.unsafe(`with w as (select id, conversation_id, payload->>'issueId' issue_id, + row_number() over (order by id) rn from chat_actions where company_id = $1::uuid and kind = 'inbound_wakeup') + insert into chat_actions (company_id, endpoint_id, conversation_id, kind, provider_action_id, status, payload) + select $1::uuid, $2::uuid, w.conversation_id, 'failed_run_retry', 'retry-' || w.rn, 'processed', + jsonb_build_object('issueId', w.issue_id, 'agentId', $3::text, + 'taskRetryInput', jsonb_build_object('skippedActionIds', jsonb_build_array(w.id::text))) + from w where w.rn <= $4::int`, [f.companyId, f.endpointId, f.agentId, RETRIES]); + await sql.unsafe(`insert into agent_wakeup_requests (id, company_id, agent_id, source, status) + select id, company_id, $2::uuid, 'automation', 'finished' from chat_actions + where company_id = $1::uuid and kind = 'failed_run_retry'`, [f.companyId, f.agentId]); + + // Cases on fresh issues. Each issue has its own conversation slice. + const issueRows = await sql.unsafe(`select id from issues where company_id = $1::uuid and title like 'Load %' + order by id limit 6`, [f.companyId]); + const [quiet, admitted, unadmitted, supersededFailure, retried, decoyOnly] = issueRows.map(row => row.id as string); + // Remove the shared history from the quiet issue so it has no chat input. + await sql.unsafe(`delete from agent_wakeup_requests where id in (select id from chat_actions + where payload->>'issueId' = $1)`, [quiet]); + await sql.unsafe(`delete from chat_actions where payload->>'issueId' = $1`, [quiet]); + const [conversation] = await sql.unsafe(`select id from chat_conversations where company_id = $1::uuid + and external_conversation_id = 'load' order by id limit 1`, [f.companyId]); + const addAction = async (issueId: string, status: string, kind = "inbound_wakeup", extra = {}, at = "now()") => + (await sql.unsafe(`insert into chat_actions (company_id, endpoint_id, conversation_id, kind, provider_action_id, + status, payload, created_at) + values ($1::uuid, $2::uuid, $3::uuid, $4, gen_random_uuid()::text, $5, $6::jsonb, ${at}) returning id`, + [f.companyId, f.endpointId, conversation!.id, kind, status, + JSON.stringify({ issueId, agentId: f.agentId, sessionGeneration: 1, ...extra })]))[0]!.id as string; + const admit = (id: string, status = "queued") => sql.unsafe(`insert into agent_wakeup_requests + (id, company_id, agent_id, source, status) values ($1::uuid, $2::uuid, $3::uuid, 'automation', $4)`, + [id, f.companyId, f.agentId, status]); + await admit(await addAction(admitted, "processing")); + await addAction(unadmitted, "preparing"); + await addAction(supersededFailure, "failed", "inbound_wakeup", {}, "now() - interval '1 minute'"); + await admit(await addAction(supersededFailure, "processing")); + const skipped = await addAction(retried, "issued"); + await admit(await addAction(retried, "processed", "failed_run_retry", + { taskRetryInput: { skippedActionIds: [skipped] } })); + // Same issue id under another company does not hold this issue. + await sql.unsafe(`insert into chat_actions (company_id, endpoint_id, kind, provider_action_id, status, payload) + values ($1::uuid, $2::uuid, 'inbound_wakeup', 'decoy', 'preparing', $3::jsonb)`, + [decoy.companyId, decoy.endpointId, JSON.stringify({ issueId: decoyOnly, agentId: decoy.agentId })]); + await sql.unsafe("vacuum analyze"); + const loadMs = performance.now() - loadStart; + const tableSizes = await sql.unsafe(`select relname as table, n_live_tup::int as rows, + pg_size_pretty(pg_total_relation_size(relid)) as total_size from pg_stat_user_tables + where relname in ('chat_actions', 'agent_wakeup_requests', 'issues') order by relname`); + + // Recovery: reconcileStrandedAssignedIssues({ companyId, issueId }) candidate read (where clause as in + // server/src/services/recovery/service.ts). Manual start: the held-input check in heartbeat/queue.ts. + const recoveryQuery = (issueId: string) => db!.select({ id: issues.id, companyId: issues.companyId, + status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues).where(and( + eq(issues.companyId, f.companyId), eq(issues.id, issueId), isNull(issues.assigneeUserId), + inArray(issues.status, ["todo", "in_progress", "in_review"]), + or(dsql`${issues.assigneeAgentId} is not null`, eq(issues.status, "in_review")), + isNull(issues.hiddenAt), not(unadmittedChatWakeupCondition(issues.id, issues.companyId)), + notExists(db!.select({ id: caseHandoffs.id }).from(caseHandoffs).where(and( + eq(caseHandoffs.companyId, issues.companyId), eq(caseHandoffs.issueId, issues.id), + eq(caseHandoffs.destinationKind, "agent"), inArray(caseHandoffs.state, ["awaiting_release", "pending_dispatch"]), + ))))).toSQL(); + const heldQuery = (issueId: string) => db!.select({ id: issues.id }).from(issues).where(and( + eq(issues.id, issueId), eq(issues.companyId, f.companyId), + unadmittedChatWakeupCondition(issues.id, issues.companyId))).limit(1).toSQL(); + const cases = { quiet, admitted, unadmitted, supersededFailure, retried, decoyOnly } as Record; + const expectedHeld: Record = { quiet: false, admitted: false, unadmitted: true, + supersededFailure: false, retried: false, decoyOnly: false }; + + const round = (value: number) => Number(value.toFixed(3)); + const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!; + const values = async (query: { sql: string; params: unknown[] }) => + (await sql.unsafe(query.sql, query.params as never[]).values()).map(row => row.map(value => value?.toString() ?? null)); + async function explain(query: { sql: string; params: unknown[] }) { + const [result] = await sql.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]); + return (result!["QUERY PLAN"] as Array<{ Plan: Record; "Execution Time": number }>)[0]!; + } + async function sample(query: { sql: string; params: unknown[] }) { + for (let n = 0; n < 2; n++) await explain(query); + const samples: number[] = []; + for (let n = 0; n < 9; n++) samples.push((await explain(query))["Execution Time"]); + return { medianMs: round(median(samples)), minMs: round(Math.min(...samples)), maxMs: round(Math.max(...samples)), + samplesMs: samples.map(round) }; + } + async function plan(query: { sql: string; params: unknown[] }) { + const root = await explain(query), nodes: string[] = []; + const walk = (node: Record) => { + if (/Scan|Join|Loop/.test(String(node["Node Type"]))) nodes.push(`${node["Node Type"]}${node["Index Name"] ? ` using ${node["Index Name"]}` : ""}${node["Relation Name"] ? ` on ${node["Relation Name"]}` : ""}`); + for (const child of (node.Plans as Array> | undefined) ?? []) walk(child); + }; + walk(root.Plan); + return { sharedHitBlocks: root.Plan["Shared Hit Blocks"], sharedReadBlocks: root.Plan["Shared Read Blocks"], nodes }; + } + + async function measure(indexes: boolean) { + const builds: Array<{ index: string; ms: number }> = []; + if (!indexes) { + for (const name of indexNames) await sql.unsafe(`DROP INDEX CONCURRENTLY IF EXISTS "${name}"`); + } else { + for (const [n, statement] of indexStatements.entries()) { + const start = performance.now(); + await sql.unsafe(statement); + builds.push({ index: indexNames[n]!, ms: round(performance.now() - start) }); + } + } + await sql.unsafe("ANALYZE chat_actions"); + const rows: Record = {}, timings: Record = {}; + for (const [name, issueId] of Object.entries(cases)) { + for (const [caller, query] of [["recovery", recoveryQuery(issueId)], ["manualStartHeld", heldQuery(issueId)]] as const) { + rows[`${caller}.${name}`] = await values(query); + timings[`${caller}.${name}`] = { serverExecution: await sample(query), ...(await plan(query)) }; + } + } + return { indexes, builds, rows, timings }; + } + + // No index, index, then no index again to expose cache-order bias. + const runs = [await measure(false), await measure(true), await measure(false), await measure(true)]; + const reference = runs[0]!; + for (const run of runs) assert.deepEqual(run.rows, reference.rows, "Complete rows must match with and without the index"); + for (const [name, held] of Object.entries(expectedHeld)) { + assert.equal((reference.rows[`manualStartHeld.${name}`] as unknown[]).length, held ? 1 : 0, `${name} held`); + assert.equal((reference.rows[`recovery.${name}`] as unknown[]).length, held ? 0 : 1, `${name} recovery candidate`); + } + + console.log(JSON.stringify({ + treatmentRevision: git("rev-parse", "HEAD"), + postgresVersion: (await sql.unsafe("select version()"))[0]!.version, + settings: await sql.unsafe("select name, setting from pg_settings where name in ('shared_buffers','work_mem','jit') order by name"), + indexDefinitions: indexStatements, + fixture: { synthetic: true, otherActions: OTHER_ACTIONS, otherPayloadKiB: OTHER_PAYLOAD_KIB, issues: ISSUES, + wakeupsPerIssue: WAKEUPS_PER_ISSUE, wakeupPayloadKiB: WAKEUP_PAYLOAD_KIB, retries: RETRIES, + loadMs: Math.round(loadMs), tableSizes }, + sql: { recovery: recoveryQuery(quiet!).sql, manualStartHeld: heldQuery(quiet!).sql }, + correctness: { rowsEqualAcrossRuns: true, expectedHeld }, + results: runs.map(({ rows: _rows, ...run }) => run), + }, null, 2)); +} finally { + try { + await db?.$client.end({ timeout: 0 }); + await temporary?.cleanup(); + } finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/chat-admission-condition.test.ts b/server/src/__tests__/chat-admission-condition.test.ts new file mode 100644 index 00000000000..e568539d9fc --- /dev/null +++ b/server/src/__tests__/chat-admission-condition.test.ts @@ -0,0 +1,171 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { and, eq } from "drizzle-orm"; +import { agentWakeupRequests, agents, chatActions, chatConversations, createDb, issues, type Db } from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { seedGitHubReviewWait } from "./helpers/github-review-wait.js"; +import { unadmittedChatWakeupCondition } from "../services/durable-chat-wakeup.js"; + +// Pins which saved chat input holds an issue as "not admitted". Recovery skips +// held issues and manual task start refuses them. +describe("unadmitted chat wakeup condition", () => { + let temporary: Awaited>; + let db: Db; + beforeAll(async () => { + if (process.env.PAPERCLIP_TEST_DATABASE_URL) { + db = createDb(process.env.PAPERCLIP_TEST_DATABASE_URL); + } else { + temporary = await startEmbeddedPostgresTestDatabase("chat-admission-condition-"); + db = createDb(temporary.connectionString); + } + }, 120_000); + afterAll(async () => { await db?.$client.end({ timeout: 0 }); await temporary?.cleanup(); }); + + async function seed() { + const f = await seedGitHubReviewWait(db); + const otherConversationId = randomUUID(); + await db.insert(chatConversations).values({ id: otherConversationId, companyId: f.companyId, + endpointId: f.endpointId, resourceId: f.resourceId, issueId: f.issueId, externalConversationId: "other", + externalThreadId: `other:${otherConversationId}`, externalLabel: "Other" }); + let tick = 0; + const action = async (input: { status: string; kind?: string; issueId?: unknown; agentId?: string; + conversationId?: string; companyId?: string; endpointId?: string; payload?: Record }) => { + const id = randomUUID(); + await db.insert(chatActions).values({ id, companyId: input.companyId ?? f.companyId, + endpointId: input.endpointId ?? f.endpointId, conversationId: input.conversationId ?? f.conversationId, + kind: input.kind ?? "inbound_wakeup", providerActionId: id, status: input.status, + createdAt: new Date(Date.UTC(2026, 0, 1, 0, 0, tick++)), + payload: { issueId: "issueId" in input ? input.issueId : f.issueId, agentId: input.agentId ?? f.agentId, + sessionGeneration: 1, ...input.payload } }); + return id; + }; + const admit = (id: string, status = "queued", agentId = f.agentId, companyId = f.companyId) => + db.insert(agentWakeupRequests).values({ id, companyId, agentId, source: "automation", status }); + const held = async () => (await db.select({ id: issues.id }).from(issues).where(and( + eq(issues.id, f.issueId), eq(issues.companyId, f.companyId), + unadmittedChatWakeupCondition(issues.id, issues.companyId))).limit(1)).length === 1; + const reset = async () => { + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + await db.delete(chatActions).where(eq(chatActions.companyId, f.companyId)); + }; + return { ...f, otherConversationId, action, admit, held, reset }; + } + + it("holds only pending or failed inbound input without an admission receipt", async () => { + const s = await seed(); + expect(await s.held()).toBe(false); + for (const status of ["preparing", "issued", "processing", "failed"]) { + await s.reset(); + await s.action({ status }); + expect(await s.held(), status).toBe(true); + } + for (const status of ["received", "processed", "cancelled"]) { + await s.reset(); + await s.action({ status }); + expect(await s.held(), status).toBe(false); + } + // Another kind with the same issue id never holds. + await s.reset(); + await s.action({ status: "preparing", kind: "failed_run_retry" }); + await s.action({ status: "processing", kind: "interaction_wakeup" }); + expect(await s.held()).toBe(false); + }); + + it("accepts only a receipt for the same company and agent that was not skipped", async () => { + const s = await seed(); + const otherAgentId = randomUUID(); + await db.insert(agents).values({ id: otherAgentId, companyId: s.companyId, name: "Other", adapterType: "process" }); + const pending = await s.action({ status: "processing" }); + await s.admit(pending, "skipped"); + expect(await s.held()).toBe(true); + await db.update(agentWakeupRequests).set({ status: "failed" }).where(eq(agentWakeupRequests.id, pending)); + // A failed or cancelled run is still historical admission. + expect(await s.held()).toBe(false); + await db.update(agentWakeupRequests).set({ agentId: otherAgentId }).where(eq(agentWakeupRequests.id, pending)); + expect(await s.held()).toBe(true); + + const other = await seedGitHubReviewWait(db); + await s.reset(); + const crossCompany = await s.action({ status: "processing" }); + await s.admit(crossCompany, "queued", other.agentId, other.companyId); + expect(await s.held()).toBe(true); + }); + + it("compares the issue id as company-scoped canonical text", async () => { + const s = await seed(); + const other = await seedGitHubReviewWait(db); + await s.action({ status: "preparing", companyId: other.companyId, endpointId: other.endpointId, + conversationId: other.conversationId }); + await s.action({ status: "preparing", issueId: s.issueId.toUpperCase() }); + await s.action({ status: "preparing", issueId: [s.issueId] }); + await s.action({ status: "preparing", issueId: null }); + await s.action({ status: "preparing", issueId: randomUUID() }); + expect(await s.held()).toBe(false); + await s.action({ status: "preparing" }); + expect(await s.held()).toBe(true); + }); + + it("releases skipped input only through an admitted retry of the same conversation and issue", async () => { + const s = await seed(); + const skipped = await s.action({ status: "issued" }); + const retry = (input: Partial[0]> = {}) => s.action({ status: "processed", + kind: "failed_run_retry", ...input, payload: { taskRetryInput: { skippedActionIds: [skipped] }, ...input.payload } }); + // Not admitted, skipped, other conversation, other issue, or not listing the action. + await retry(); + await s.admit(await retry(), "skipped"); + await s.admit(await retry({ conversationId: s.otherConversationId })); + await s.admit(await retry({ issueId: randomUUID() })); + await s.admit(await retry({ payload: { taskRetryInput: { skippedActionIds: [randomUUID()] } } })); + expect(await s.held()).toBe(true); + await s.admit(await retry()); + expect(await s.held()).toBe(false); + }); + + it("releases a failed input only through later admitted input of the same session generation", async () => { + const s = await seed(); + const failed = await s.action({ status: "failed" }); + const pending = await s.action({ status: "preparing" }); + void pending; + // Later input from another generation or conversation does not settle it. + await s.admit(await s.action({ status: "processed", payload: { sessionGeneration: 2 } })); + await s.admit(await s.action({ status: "processed", conversationId: s.otherConversationId })); + await s.admit(await s.action({ status: "processed" }), "skipped"); + await db.delete(chatActions).where(eq(chatActions.id, pending)); + expect(await s.held()).toBe(true); + await s.admit(await s.action({ status: "processed" })); + expect(await s.held()).toBe(false); + + // Earlier admitted input does not settle a later failure; nor does any later + // input settle input that is still pending. + await s.reset(); + await s.admit(await s.action({ status: "processed" })); + await s.action({ status: "failed" }); + expect(await s.held()).toBe(true); + await s.reset(); + await s.action({ status: "processing" }); + await s.admit(await s.action({ status: "processed" })); + expect(await s.held()).toBe(true); + void failed; + }); + + it("finds the issue's chat input through the company and issue index", async () => { + const s = await seed(); + await db.$client.unsafe(`insert into chat_actions (company_id, endpoint_id, kind, provider_action_id, status, payload) + select $1::uuid, $2::uuid, case when g % 3 = 0 then 'github_review_check' else 'inbound_wakeup' end, + 'noise-' || g, 'processed', jsonb_build_object('issueId', gen_random_uuid()::text, 'body', repeat('x', 200)) + from generate_series(1, 3000) g`, [s.companyId, s.endpointId]); + await db.$client.unsafe("analyze chat_actions"); + await s.action({ status: "failed" }); + const query = db.select({ id: issues.id }).from(issues).where(and(eq(issues.id, s.issueId), + eq(issues.companyId, s.companyId), unadmittedChatWakeupCondition(issues.id, issues.companyId))).limit(1).toSQL(); + const [row] = await db.$client.unsafe(`EXPLAIN (FORMAT JSON) ${query.sql}`, query.params as never[]); + const scans: string[] = []; + const walk = (node: Record) => { + if (node["Relation Name"] === "chat_actions") scans.push(`${node["Node Type"]}:${node["Index Name"] ?? ""}`); + for (const child of (node.Plans as Array> | undefined) ?? []) walk(child); + }; + walk((row!["QUERY PLAN"] as Array<{ Plan: Record }>)[0]!.Plan); + expect(scans.length).toBeGreaterThan(0); + for (const scan of scans) expect(scan).toMatch(/Index.*Scan:chat_actions_company_issue_kind_idx/); + }); +});