mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf(chat): index chat-admission checks by company and payload issue id (#20562)
## 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 <noreply@anthropic.com>
This commit is contained in:
1 parent
96ab5e3d38
commit
765ccd8691
9 files changed
+681
-2
No files matched your search
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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');
|
||||
+183
-2
@@ -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": [
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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'`),
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb> | 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); }
|
||||
}
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb> | 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<string, string>;
|
||||
const expectedHeld: Record<string, boolean> = { 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<string, unknown>; "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<string, unknown>) => {
|
||||
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<Record<string, unknown>> | 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<string, unknown> = {}, timings: Record<string, unknown> = {};
|
||||
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); }
|
||||
}
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
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<string, unknown> }) => {
|
||||
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<Parameters<typeof s.action>[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<string, unknown>) => {
|
||||
if (node["Relation Name"] === "chat_actions") scans.push(`${node["Node Type"]}:${node["Index Name"] ?? ""}`);
|
||||
for (const child of (node.Plans as Array<Record<string, unknown>> | undefined) ?? []) walk(child);
|
||||
};
|
||||
walk((row!["QUERY PLAN"] as Array<{ Plan: Record<string, unknown> }>)[0]!.Plan);
|
||||
expect(scans.length).toBeGreaterThan(0);
|
||||
for (const scan of scans) expect(scan).toMatch(/Index.*Scan:chat_actions_company_issue_kind_idx/);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user