mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
perf(server): prefilter chat milestone run candidates (#17892)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Chat publications connect a run to its authorized originating conversation. > - The milestone selector evaluates detailed lineage and final-answer predicates. > - A local plan evaluated these predicates across many unrelated run and conversation pairs. > - This pull request materializes recent runs that already match an active or waiting conversation. > - The existing query still makes each publication decision. > - The benefit is less JSON work when most run history is unrelated to active chat. ## Linked Issues or Issue Description **What happened?** The milestone candidate query repeatedly reads large run context values for unrelated conversations. A synthetic baseline discarded 60,962 run and conversation pairs through the complex join filter. **Expected behavior** Apply the existing same-company conversation binding before expensive lineage checks. Preserve every candidate column, order, publication decision and continuation rule. **Steps to reproduce** Run the two synthetic PostgreSQL comparisons below against baseline `db70c5acd92f50fa37fb4f0dea3da0e7685089a2`. They capture complete SQL from the actual service implementations. Related work: #17606 changes chat lifecycle scheduling. #15802 deferred milestone cadence. This SQL-only change leaves scheduling, startup reconciliation and recovery intact. It has no dependency on those PRs. ## What Changed - Materialize recent runs with an active or waiting same-company conversation, using the unchanged native and legacy binding expression. - Retain every outer lineage, continuation, final-response, privacy, cursor, ordering and idempotency check. - Add a real PostgreSQL historical-binding regression and selective/dense performance controls. - Document the measured gain and the dense-history tradeoff. ## Verification - Dedicated binding, publication reconciliation and app coordination suites passed all 27 tests across three files with real PostgreSQL where required. - The selective fixture has 2,615 runs, 55 conversation issues and about 192 KiB of pseudorandom context padding per run. Five alternating measured samples follow two warmups on PostgreSQL 17.11. - Final-harness selective milestone median: 1,495.85 ms before, 130.96 ms after, 11.42 times faster. Ranges: 1,483–1,716 ms and 125.59–135.85 ms. All 60 complete ordered rows match, including 15 explicit historical-binding cases. - Dense control: 736.71 ms before, 809.08 ms after, with overlapping ranges. All 200 complete ordered rows match. This is inconclusive performance evidence with possible materialization overhead. - The unchanged native progress control is neutral in both fixtures. It returns 41 equal complete ordered rows. - The added large-window control has 12,215 runs, including 12,000 recent native history rows. A caller limit of one produces the real service page size of 25. First-page medians 762.65 to 800.63 ms overlap and are inconclusive. The 90% cursor improves 234.84 to 54.36 ms; the 99.99% tail improves 352.10 to 12.88 ms; the finished cursor improves 340.00 to 58.58 ms. Complete ordered results match: 25, 25, seven and zero rows. Same-run later-conversation rows remain present. - Large-window CTE materialization itself takes about 4.6–10 ms with zero temp spill. This does not establish a materialization regression or justify a larger production query rewrite. - Reproduce the added window control with `--cursor-window` and a distinct report path. - Reproduce selective: `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-milestones.ts --baseline-ref=db70c5acd92f50fa37fb4f0dea3da0e7685089a2 --docker-postgres=17 --report=/tmp/chat-milestones.json`. - Add `--dense-chat` and a distinct report path for the dense control. The harness applies normal migrations to a disposable container and uses no hosted database. - The existing real-service test at `server/src/__tests__/chat-channels.integration.test.ts:34584` calls enqueue with limit=1 four times, crosses 30 resolver-rejected candidates, expects admissions [1,1,1,0], and asserts the exact three publication keys plus internal-run exclusion. It passed and supplies the full composition/publication-set proof requested in review. - Assembled chat integration passed 37 selected tests. Together with the 27 focused tests, 64 checks passed. The selector covers bounded draining, exact continuations, private summaries, final-answer precedence, lifecycle idempotency and native progress races. - Final head `1b3a6f50c9e7ec712dcbaf82c7fc3df141ec3f28`: [full CI passed](https://github.com/paperclipai/paperclip/actions/runs/40534934737), with 55 successful checks, two skipped and one neutral. Greptile is 5/5 on this head; zero unresolved review threads. Paperclip Review is 4/5, neutral, explicitly with no must-fix correctness or security defect. Its remaining payload-memory/dense-overhead caution is retained here as a measured residual tradeoff; an identity-only projection needs production evidence and separate comparison. - After the Drizzle metadata correction, both generated service queries and every parameter remain byte-identical to the selective and dense measured versions. The real PostgreSQL binding regression passed again. - Full repository typecheck and build passed on the integrated final source. The local full test command was stopped after native PostgreSQL initialization failed repeatedly, producing 16 failures and skipped database suites. It is incomplete, not a pass. Focused real-database suites used disposable Docker PostgreSQL. Current-head CI runs the full gates separately. ## Risks - A materialized source can add overhead when most recent runs match active conversations. The dense control does not establish a gain. - Historical JSON comparisons retain string and whitespace semantics. Test fixtures disable the normalization trigger only while inserting retained historical shapes in disposable databases. - Synthetic SQL measurements exclude provider delivery and do not establish deployed latency, actual compute sleep or fleet-wide cost reduction. - No schema migration, index, startup job, polling change or deployment is included. ## Model Used OpenAI Codex, configured model `gpt-6.1-sol` with xhigh reasoning, tool use and code execution. Independent blast-radius review also used `gpt-6-astra` and `gpt-6.1-sol`. The runtime context window was not exposed. ## 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 --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
a26d217e94
commit
475f7df2e7
5 files changed
+754
-1
No files matched your search
@@ -0,0 +1,150 @@
|
||||
# Chat milestone query performance
|
||||
|
||||
The milestone selector first materializes runs that match an active or waiting
|
||||
conversation in the same company. It then applies the existing publication
|
||||
origin, continuation, status and final-answer checks. This prevents PostgreSQL
|
||||
from evaluating expensive JSON lineage predicates across unrelated
|
||||
run/conversation pairs.
|
||||
|
||||
The change preserves `boundRunIssueIdSql()` and all publication authorization,
|
||||
idempotency, ordering and pagination logic. It adds no migration or index. The
|
||||
one-second recovery cadence, startup reconciliation, commit signals, shutdown
|
||||
and native progress selector are unchanged. The separate lifecycle scheduling
|
||||
work in [PR #17606](https://github.com/paperclipai/paperclip/pull/17606) is outside
|
||||
this change.
|
||||
|
||||
## Reproduce the measurements
|
||||
|
||||
Run from the repository root with Docker available:
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-milestones.ts --baseline-ref=db70c5acd92f50fa37fb4f0dea3da0e7685089a2 --docker-postgres=17 --report=/tmp/chat-milestones-report.json
|
||||
pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-milestones.ts --baseline-ref=db70c5acd92f50fa37fb4f0dea3da0e7685089a2 --docker-postgres=17 --dense-chat --report=/tmp/chat-milestones-dense-report.json
|
||||
pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-chat-milestones.ts --baseline-ref=db70c5acd92f50fa37fb4f0dea3da0e7685089a2 --docker-postgres=17 --cursor-window --report=/tmp/chat-milestones-large-window-report.json
|
||||
```
|
||||
|
||||
Without `--docker-postgres=17`, the harness uses disposable embedded PostgreSQL.
|
||||
Neither mode uses an existing application database. The Docker mode starts a
|
||||
container on an ephemeral localhost port, applies the repository migrations and
|
||||
removes the container after the run.
|
||||
|
||||
The harness imports the baseline service from the pinned commit and captures
|
||||
the complete SQL and parameters generated by both real service implementations.
|
||||
It stops before publication writes. The report includes full positional result
|
||||
equality, all samples, plans, database version/settings, source hashes and the
|
||||
treatment diff. A detached, uncommitted treatment is identified by its file
|
||||
hashes and diff in addition to its HEAD.
|
||||
|
||||
## Local results
|
||||
|
||||
Measured on PostgreSQL 17.11 with 128 MB shared buffers and 4 MB work memory.
|
||||
The synthetic fixture contains 2,615 runs, 55 conversation issues and contexts
|
||||
containing a base64 string made from 144 KiB of pseudorandom bytes, approximately
|
||||
192 KiB of text. Each query has two warmup passes and five measured passes;
|
||||
baseline and treatment order alternates. These are local SQL measurements,
|
||||
excluding provider delivery and application processing.
|
||||
|
||||
| Fixture and query | Baseline median, range | Treatment median, range | Complete ordered rows |
|
||||
| --- | --- | --- | --- |
|
||||
| Selective bindings: milestones | 1,496 ms, 1,483–1,716 | 131 ms, 125.59–135.85 | 60 equal |
|
||||
| Selective bindings: native progress control | 17.15 ms, 16.35–18.38 | 17.43 ms, 16.77–19.78 | 41 equal |
|
||||
| Dense bindings: milestones | 737 ms, 629–1,291 | 809 ms, 463–1,288 | 200 equal |
|
||||
| Dense bindings: native progress control | 23.93 ms, 22.72–25.47 | 24.33 ms, 20.33–29.37 | 41 equal |
|
||||
|
||||
The selective fixture improved by **11.42×**. Its original merge join discarded
|
||||
60,962 run/conversation pairs through a complex lineage filter. The treatment
|
||||
materialized 101 matching runs first and discarded no pairs through that join
|
||||
filter. Shared-hit blocks fell from 3,276,955 to 64,903. Neither plan spilled to
|
||||
temporary files.
|
||||
|
||||
The dense control binds about 2,500 formerly unrelated history rows to active
|
||||
conversations through the first-class issue column, preserving existing native
|
||||
progress receipts. It materializes 1,768 runs. Its treatment median was about
|
||||
10% slower, with heavily overlapping sample ranges. A subsequent standalone
|
||||
EXPLAIN also varied substantially. **Dense performance is inconclusive, with
|
||||
possible materialization overhead; it does not establish a speedup.** Both
|
||||
native progress controls are neutral.
|
||||
|
||||
The gain depends on binding selectivity and the database plan. These
|
||||
measurements do not establish a universal speedup, improved request latency or
|
||||
deployed database cost. Compare fresh production query-statistics windows after
|
||||
any separately authorized rollout.
|
||||
|
||||
## Large-window cursor control
|
||||
|
||||
The `--cursor-window` control uses the same service implementation with 12,215
|
||||
runs, including 12,000 recent native runs with first-class task bindings and
|
||||
approximately 8 KiB encoded contexts. It requests a limit of one; the service's
|
||||
minimum candidate page size is 25. The fixture adds a second conversation for
|
||||
the same task so both late nonempty pages must preserve the other conversation
|
||||
on the cursor's own run.
|
||||
|
||||
Every statement is captured from `enqueueChatRunMilestones`, including its
|
||||
actual second-page cursor. Capture advances the real keyset loop with a
|
||||
non-publishable placeholder page and stops before any publication write. The
|
||||
finished cursor comes from the last actual candidate on the short final page.
|
||||
All statements then run against the real PostgreSQL fixture, with two warmups
|
||||
and five alternating measurements.
|
||||
|
||||
| Service page | Baseline median, range | Treatment median, range | Complete ordered rows |
|
||||
| --- | --- | --- | --- |
|
||||
| First page | 762.65 ms, 678.43–807.93 | 800.63 ms, 597.38–838.01 | 25 equal |
|
||||
| Cursor at 90% of history | 234.84 ms, 226.42–241.19 | 54.36 ms, 53.39–63.62 | 25 equal |
|
||||
| Cursor at 99.99% of history | 352.10 ms, 321.27–377.10 | 12.88 ms, 12.13–29.92 | 7 equal |
|
||||
| Cursor after the final candidate | 340.00 ms, 324.40–381.30 | 58.58 ms, 56.55–78.92 | 0 equal |
|
||||
|
||||
The first page is neutral/inconclusive; its sample ranges overlap. Later pages
|
||||
improved by 4.32×, 27.33× and 5.80× respectively. The unchanged native progress
|
||||
control was neutral at 22.79 versus 22.72 ms, with 25 equal rows.
|
||||
|
||||
The treatment still materializes 12,101 matching runs before its outer cursor
|
||||
filter. The materialization node took 4.6–10.0 ms in these plans, and no measured
|
||||
page spilled temporary data. These observations do not measure peak memory or
|
||||
prove how every JSON value is stored or decompressed. They do show that this
|
||||
bounded, native-heavy fixture did not reproduce the feared large-window cursor
|
||||
regression. No identity-only rewrite or additional cursor predicate was needed
|
||||
for this change. The dense and first-page performance limits remain disclosed.
|
||||
|
||||
## Correctness and blast radius
|
||||
|
||||
The safety condition is that the new prefilter repeats a predicate already
|
||||
required by the existing conversation join. It uses the same company,
|
||||
conversation-state and native/legacy binding expression within the same SQL
|
||||
statement. All authoritative checks still run afterward.
|
||||
|
||||
The selective benchmark compares every returned column in exact order and
|
||||
checks 15 explicit historical binding cases. These include whitespace trimming,
|
||||
task-ID fallback, numeric/object/array JSON values, uppercase and malformed
|
||||
identifiers, native and first-class issue precedence, cross-company values and
|
||||
null/scalar/array contexts. Dense results compare the complete first 200 ordered
|
||||
rows; they do not claim that all history fits on that page.
|
||||
|
||||
Migration 0313's insert trigger normally normalizes recognized JSON identifiers
|
||||
into the first-class issue binding. The fixture disables that trigger only
|
||||
while inserting historical row shapes in its disposable database and restores
|
||||
it in `finally`. Runtime triggers are unchanged.
|
||||
|
||||
Run the dedicated regression and existing coordination suites:
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/server exec vitest run src/__tests__/chat-run-milestone-candidates.test.ts src/services/chat-publication-reconciliation.test.ts src/__tests__/app-private-hostname-gate.test.ts
|
||||
```
|
||||
|
||||
These three suites passed **27 tests** during validation, with a disposable
|
||||
PostgreSQL 17 constructor override for the database test.
|
||||
|
||||
The service integration checks cover bounded draining past rejected
|
||||
candidates, exact continuation origins, private summaries, explicit final-answer
|
||||
precedence, idempotent lifecycle updates, ownership-attention messages and native
|
||||
progress reauthorization after a question, final answer or revoked reach wins.
|
||||
The relevant cases live in `server/src/__tests__/chat-channels.integration.test.ts`.
|
||||
These checks prove service behavior using synthetic provider runtimes; they do
|
||||
not prove live provider delivery or deployed sleep/wake behavior.
|
||||
|
||||
The following selection passed **37 tests** with disposable PostgreSQL 17;
|
||||
1,529 unrelated cases were skipped:
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/server exec vitest run src/__tests__/chat-channels.integration.test.ts \
|
||||
-t "drains bounded Slack chat-origin milestones|coalesces one Slack run's lifecycle|edits a Slack working reply|publishes a Telegram failure for an exact confirmation continuation|replaces working with one safe ownership-attention|does not let a late|keeps an internal Telegram run summary private|publishes only a resolver-selected Telegram reply|settles an interrupted native run once|preserves an interrupted run's selected final|skips a locked native progress|reauthorizes skipped native progress|coalesces closed native progress|publishes native progress with a typed issue binding|enforces the native progress cadence|does not replay stale native progress|keeps native progress behind question|reuses a pre-run FIFO notice|link-only interaction continuation|successful GitHub link-question continuation|publishes a closed-choice question, settles its Slack card"
|
||||
```
|
||||
@@ -0,0 +1,459 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath, pathToFileURL } from "node:url";
|
||||
import { companies, agents, issues, applyPendingMigrations, createDb, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js";
|
||||
import * as treatment from "../server/src/services/chat-run-publications.js";
|
||||
|
||||
type Query = { sql: string; params: unknown[] };
|
||||
const dockerPostgres = process.argv.includes("--docker-postgres=17");
|
||||
const correctnessOnly = process.argv.includes("--correctness-only");
|
||||
const cursorWindow = process.argv.includes("--cursor-window");
|
||||
const historyRuns = cursorWindow ? 12_000 : 2_400;
|
||||
const denseChat = process.argv.includes("--dense-chat");
|
||||
const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15);
|
||||
assert.ok(baselineRef, "Pass a pinned pre-fix --baseline-ref=<commit>");
|
||||
const repositoryRoot = fileURLToPath(new URL("../", import.meta.url));
|
||||
const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8" }).trim();
|
||||
const ANCHOR_MS = Date.parse("2026-10-19T00:00:00.000Z");
|
||||
const SINCE_MS = ANCHOR_MS - 24 * 60 * 6e4;
|
||||
const PROMPT_KIB = cursorWindow ? 6 : 144;
|
||||
function mulberry32(seed2: number) {
|
||||
return () => {
|
||||
seed2 |= 0;
|
||||
seed2 = seed2 + 1831565813 | 0;
|
||||
let t = Math.imul(seed2 ^ seed2 >>> 15, 1 | seed2);
|
||||
t = t + Math.imul(t ^ t >>> 7, 61 | t) ^ t;
|
||||
return ((t ^ t >>> 14) >>> 0) / 4294967296;
|
||||
};
|
||||
}
|
||||
const rand = mulberry32(1346457664);
|
||||
const hex = (n: number) => Array.from({ length: n }, () => "0123456789abcdef"[Math.floor(rand() * 16)]).join("");
|
||||
const uuid = () => `${hex(8)}-${hex(4)}-4${hex(3)}-${"89ab"[Math.floor(rand() * 4)]}${hex(3)}-${hex(12)}`;
|
||||
const prompt = Buffer.from(
|
||||
Array.from({ length: PROMPT_KIB * 1024 }, () => Math.floor(rand() * 256))
|
||||
).toString("base64");
|
||||
const iso = (ms: number) => new Date(ms).toISOString();
|
||||
async function seed(db: Db) {
|
||||
const companyA = uuid();
|
||||
const companyB = uuid();
|
||||
const agentA = uuid();
|
||||
const agentB = uuid();
|
||||
const endpointA = uuid();
|
||||
const endpointB = uuid();
|
||||
await db.insert(companies).values([
|
||||
{ id: companyA, name: "Synthetic chat polls A", issuePrefix: "BENCH" },
|
||||
{ id: companyB, name: "Synthetic chat polls B", issuePrefix: "BNCH" }
|
||||
]);
|
||||
await db.insert(agents).values([
|
||||
{ id: agentA, companyId: companyA, name: "Fixture A", adapterType: "codex_local" },
|
||||
{ id: agentB, companyId: companyB, name: "Fixture B", adapterType: "codex_local" }
|
||||
]);
|
||||
const conversationIssueIds: string[] = [];
|
||||
const issueIdsA: string[] = [];
|
||||
for (let n = 0; n < 50; n++) {
|
||||
const issueId = uuid();
|
||||
if (n < 45) conversationIssueIds.push(issueId);
|
||||
issueIdsA.push(issueId);
|
||||
await db.insert(issues).values({ id: issueId, companyId: companyA, title: `Fixture issue A${n}` });
|
||||
}
|
||||
const conversationIssueIdsB: string[] = [];
|
||||
for (let n = 0; n < 10; n++) {
|
||||
const issueId = uuid();
|
||||
conversationIssueIdsB.push(issueId);
|
||||
await db.insert(issues).values({ id: issueId, companyId: companyB, title: `Fixture issue B${n}` });
|
||||
}
|
||||
for (const [companyId, endpointId, agentId, ids, label] of [
|
||||
[companyA, endpointA, agentA, conversationIssueIds, "A"],
|
||||
[companyB, endpointB, agentB, conversationIssueIdsB, "B"]
|
||||
]) {
|
||||
const applicationId = uuid();
|
||||
const connectionId = uuid();
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO tool_applications
|
||||
(id, company_id, name, type, status)
|
||||
VALUES ($1::uuid, $2::uuid, $3, 'chat', 'active')`,
|
||||
[applicationId, companyId, `fixture-app-${label}`]
|
||||
);
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO tool_connections
|
||||
(id, company_id, application_id, name, uid, connection_purpose, transport, status, enabled)
|
||||
VALUES ($1::uuid, $2::uuid, $3::uuid, $4, $5, 'channel', 'chat_sdk', 'active', true)`,
|
||||
[connectionId, companyId, applicationId, `fixture-connection-${label}`, `fixture-conn-${label}`]
|
||||
);
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO chat_endpoints
|
||||
(id, company_id, connection_id, provider, public_id, publication_mode, assigned_agent_id, status)
|
||||
VALUES ($1::uuid, $2::uuid, $3::uuid, 'slack', $4, 'automatic', $5::uuid, 'active')`,
|
||||
[endpointId, companyId, connectionId, `fixture-endpoint-${label}`, agentId]
|
||||
);
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO chat_conversations
|
||||
(id, company_id, endpoint_id, issue_id, external_conversation_id, external_label, state)
|
||||
SELECT gen_random_uuid(), $1::uuid, $2::uuid, issue::uuid, 'conv-' || n, 'Conv ${label} ' || n,
|
||||
CASE WHEN n % 9 = 0 THEN 'waiting' WHEN n % 11 = 0 THEN 'completed' ELSE 'active' END
|
||||
FROM (SELECT issue, row_number() OVER () AS n FROM unnest($3::uuid[]) AS issue) sub`,
|
||||
[companyId, endpointId, ids]
|
||||
);
|
||||
}
|
||||
const nativeRunIds: string[] = [];
|
||||
const convRowsA = conversationIssueIds.map((issue, n) => ({ issue, n }));
|
||||
for (const { issue, n } of convRowsA) {
|
||||
const nativeRunId = uuid();
|
||||
nativeRunIds.push(nativeRunId);
|
||||
const base = ANCHOR_MS - n % 40 * 6e4;
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO heartbeat_runs
|
||||
(id, company_id, agent_id, scope_kind, issue_id, runtime_mode, status, error_code, created_at, updated_at, context_snapshot, result_json)
|
||||
VALUES ($1::uuid, $2::uuid, $3::uuid, 'issue', $4::uuid, 'native', 'running', NULL, $5, $5,
|
||||
jsonb_build_object('issueId', $4::text, 'source', 'chat:slack:fixture', 'prompt', $6::text), NULL)`,
|
||||
[nativeRunId, companyA, agentA, issue, iso(base), prompt]
|
||||
);
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO heartbeat_runs
|
||||
(id, company_id, agent_id, scope_kind, runtime_mode, status, error_code, created_at, updated_at, context_snapshot, result_json)
|
||||
VALUES ($1::uuid, $2::uuid, $3::uuid, 'company', 'legacy',
|
||||
CASE WHEN $4 % 5 = 0 THEN 'failed' WHEN $4 % 7 = 0 THEN 'succeeded' ELSE 'running' END,
|
||||
CASE WHEN $4 % 5 = 0 THEN 'execution_reconciliation_required' END, $5, $5,
|
||||
jsonb_build_object('issueId', $6::text, 'source', 'chat:slack:fixture', 'prompt', $7::text),
|
||||
CASE WHEN $4 % 7 = 0 THEN jsonb_build_object('presentationDecision', jsonb_build_object('kind', 'fixture')) END)`,
|
||||
[uuid(), companyA, agentA, n, iso(base), issue, prompt]
|
||||
);
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO chat_publications
|
||||
(company_id, endpoint_id, conversation_id, issue_id, idempotency_key, payload, state, created_at)
|
||||
VALUES ($1::uuid, $2::uuid,
|
||||
(SELECT id FROM chat_conversations WHERE company_id = $1::uuid AND issue_id = $4::uuid LIMIT 1),
|
||||
$4::uuid, 'run:' || $3::text || ':working:' || $2::text,
|
||||
jsonb_build_object('classification', 'external', 'source', 'safe_milestone', 'text', 'working'),
|
||||
'streaming', $5)`,
|
||||
[companyA, endpointA, nativeRunId, issue, iso(base - 12e4)]
|
||||
);
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO heartbeat_run_events
|
||||
(company_id, run_id, agent_id, seq, event_type, message, created_at)
|
||||
VALUES ($1::uuid, $2::uuid, $3::uuid, 1, 'workspace.ready', 'ready', $4),
|
||||
($1::uuid, $2::uuid, $3::uuid, 2, 'tool.execution.started', 'tool', $5)`,
|
||||
[companyA, nativeRunId, agentA, iso(base + 3e4), iso(base + 45e3)]
|
||||
);
|
||||
}
|
||||
for (const [n, issue] of conversationIssueIdsB.entries()) {
|
||||
const runId = uuid();
|
||||
const base = ANCHOR_MS - n % 10 * 6e4;
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO heartbeat_runs
|
||||
(id, company_id, agent_id, scope_kind, issue_id, runtime_mode, status, error_code, created_at, updated_at, context_snapshot)
|
||||
VALUES ($1::uuid, $2::uuid, $3::uuid, 'issue', $4::uuid, 'native', 'running', NULL, $5, $5,
|
||||
jsonb_build_object('issueId', $4::text, 'source', 'chat:slack:fixture', 'prompt', $6::text))`,
|
||||
[runId, companyB, agentB, issue, iso(base), prompt]
|
||||
);
|
||||
}
|
||||
for (let start = 0; start < historyRuns; start += 400) {
|
||||
if (cursorWindow) {
|
||||
await db.$client.unsafe(`insert into heartbeat_runs
|
||||
(company_id, agent_id, scope_kind, issue_id, native_issue_id, runtime_mode, status, created_at, updated_at, context_snapshot)
|
||||
select $1::uuid, $2::uuid, 'issue', $3::uuid, $3::uuid, 'native', 'running',
|
||||
$4::timestamptz - (12001 - n) * interval '50 milliseconds',
|
||||
$4::timestamptz - (12001 - n) * interval '50 milliseconds',
|
||||
jsonb_build_object('issueId', $3::text, 'source', 'chat:slack:fixture', 'fixtureHistory', true, 'prompt', $5::text)
|
||||
from generate_series(${start + 1}, ${start + 400}) n`,
|
||||
[companyA, agentA, conversationIssueIds[0], iso(ANCHOR_MS), prompt]);
|
||||
continue;
|
||||
}
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO heartbeat_runs
|
||||
(company_id, agent_id, scope_kind, status, error_code, created_at, updated_at, context_snapshot)
|
||||
SELECT $1::uuid, $2::uuid, 'company',
|
||||
(ARRAY['queued','running','succeeded','interrupted','failed','timed_out','cancelled'])[(n % 7) + 1],
|
||||
NULL,
|
||||
make_interval(hours => CASE WHEN n % 3 = 0 THEN -(30 + (n % 100)) ELSE -(n % 16) END, secs => -60) + $4::timestamptz,
|
||||
make_interval(hours => CASE WHEN n % 3 = 0 THEN -(30 + (n % 100)) ELSE -(n % 16) END, secs => -60) + $4::timestamptz,
|
||||
jsonb_build_object('issueId', gen_random_uuid()::text, 'source', 'chat:slack:fixture', 'prompt', $3::text)
|
||||
FROM generate_series(${start + 1}, ${start + 400}) n`,
|
||||
[companyA, agentA, prompt, iso(ANCHOR_MS)]
|
||||
);
|
||||
}
|
||||
await db.$client.unsafe(
|
||||
`INSERT INTO heartbeat_runs
|
||||
(company_id, agent_id, scope_kind, status, error_code, created_at, updated_at, context_snapshot)
|
||||
SELECT $1::uuid, $2::uuid, 'company',
|
||||
(ARRAY['queued','running','succeeded','interrupted','failed','timed_out','cancelled'])[(n % 7) + 1],
|
||||
NULL,
|
||||
make_interval(hours => CASE WHEN n % 3 = 0 THEN -(30 + (n % 100)) ELSE -(n % 16) END, secs => -60) + $4::timestamptz,
|
||||
make_interval(hours => CASE WHEN n % 3 = 0 THEN -(30 + (n % 100)) ELSE -(n % 16) END, secs => -60) + $4::timestamptz,
|
||||
jsonb_build_object('issueId', gen_random_uuid()::text, 'source', 'chat:slack:fixture', 'prompt', $3::text)
|
||||
FROM generate_series(1, 100) n`,
|
||||
[companyB, agentB, prompt, iso(ANCHOR_MS)]
|
||||
);
|
||||
await db.$client.unsafe(
|
||||
"UPDATE heartbeat_runs SET scope_kind = 'company', issue_id = NULL WHERE id = $1::uuid",
|
||||
[nativeRunIds[0]]
|
||||
);
|
||||
await db.$client.unsafe("ANALYZE heartbeat_runs");
|
||||
await db.$client.unsafe("ANALYZE chat_conversations");
|
||||
await db.$client.unsafe("ANALYZE heartbeat_run_events");
|
||||
await db.$client.unsafe("ANALYZE chat_publications");
|
||||
return { companyA, companyB };
|
||||
}
|
||||
|
||||
async function makeDenseNativeChatHistory(db: Db) {
|
||||
// Bind the otherwise unrelated history through the first-class issue column.
|
||||
// Preserve existing bound/native runs and their progress publications so the
|
||||
// unchanged native progress query remains a nonempty control.
|
||||
await db.$client.unsafe(`with bindings as (
|
||||
select distinct on (company_id) company_id, issue_id
|
||||
from chat_conversations where state = 'active' order by company_id, id
|
||||
) update heartbeat_runs r set scope_kind = 'issue', issue_id = b.issue_id,
|
||||
context_snapshot = jsonb_set(r.context_snapshot, '{issueId}', to_jsonb(b.issue_id::text))
|
||||
from bindings b where r.company_id = b.company_id
|
||||
and r.issue_id is null and r.native_issue_id is null and r.runtime_mode = 'legacy'`);
|
||||
}
|
||||
|
||||
// Add real source comments so complete-result equality includes admitted rows,
|
||||
// and conflicting native/legacy bindings cannot be hidden by an empty result.
|
||||
async function seedSourceMessages(db: Db) {
|
||||
const conversations = await db.$client.unsafe("select id, company_id, endpoint_id, issue_id from chat_conversations");
|
||||
for (const conversation of conversations) {
|
||||
const commentId = uuid();
|
||||
await db.$client.unsafe("insert into issue_comments (id, company_id, issue_id, body) values ($1, $2, $3, 'synthetic inbound')", [commentId, conversation.company_id, conversation.issue_id]);
|
||||
await db.$client.unsafe(`insert into chat_message_links (company_id, endpoint_id, conversation_id, comment_id, provider_message_id, direction)
|
||||
values ($1, $2, $3, $4::uuid, $4::text, 'inbound')`, [conversation.company_id, conversation.endpoint_id, conversation.id, commentId]);
|
||||
await db.$client.unsafe(`update heartbeat_runs set context_snapshot = context_snapshot || jsonb_build_object('wakeCommentId', $1::text)
|
||||
|| case when $4::boolean then jsonb_build_object('wakeCommentIds', coalesce(context_snapshot->'wakeCommentIds', '[]'::jsonb) || to_jsonb($1::text)) else '{}'::jsonb end
|
||||
where company_id = $2 and (issue_id = $3::uuid or context_snapshot->>'issueId' = $3::text)`, [commentId, conversation.company_id, conversation.issue_id, cursorWindow]);
|
||||
}
|
||||
await db.$client.unsafe("analyze heartbeat_runs");
|
||||
await db.$client.unsafe("analyze chat_message_links");
|
||||
}
|
||||
async function seedBindingCases(db: Db) {
|
||||
const [conversation] = await db.$client.unsafe(`select conversation.*, link.comment_id, endpoint.assigned_agent_id
|
||||
from chat_conversations conversation
|
||||
join chat_message_links link on link.conversation_id = conversation.id
|
||||
join chat_endpoints endpoint on endpoint.id = conversation.endpoint_id
|
||||
where conversation.state = 'active' order by conversation.id limit 1`);
|
||||
const [otherIssue] = await db.$client.unsafe("select id from issues where company_id = $1 and id <> $2 order by id limit 1", [conversation.company_id, conversation.issue_id]);
|
||||
const [foreignIssue] = await db.$client.unsafe("select id from issues where company_id <> $1 order by id limit 1", [conversation.company_id]);
|
||||
const issueId = conversation.issue_id as string;
|
||||
const context = { source: "chat:slack:fixture", issueId, wakeCommentId: conversation.comment_id };
|
||||
const cases = [
|
||||
{ name: "legacy-issue", context, matches: true },
|
||||
{ name: "trimmed-issue", context: { ...context, issueId: ` ${issueId} ` }, matches: true },
|
||||
{ name: "task-fallback", context: { ...context, issueId: " ", taskId: issueId }, matches: true },
|
||||
{ name: "numeric-issue-fallback", context: { ...context, issueId: 2, taskId: issueId }, matches: true },
|
||||
{ name: "object-issue-fallback", context: { ...context, issueId: { id: issueId }, taskId: issueId }, matches: true },
|
||||
{ name: "array-issue-fallback", context: { ...context, issueId: [issueId], taskId: issueId }, matches: true },
|
||||
{ name: "uppercase-legacy", context: { ...context, issueId: issueId.toUpperCase() }, matches: false },
|
||||
{ name: "malformed-legacy", context: { ...context, issueId: "not-a-uuid", taskId: issueId }, matches: false },
|
||||
{ name: "native-over-legacy", context: { ...context, issueId: "not-a-uuid" }, nativeIssueId: issueId, matches: true },
|
||||
{ name: "issue-column-over-legacy", context: { ...context, issueId: "not-a-uuid" }, runIssueId: issueId, matches: true },
|
||||
{ name: "native-over-issue-column", context, nativeIssueId: issueId, runIssueId: otherIssue.id, matches: true },
|
||||
{ name: "foreign-native-over-local", context, nativeIssueId: foreignIssue.id, matches: false },
|
||||
{ name: "null-context", context: null, matches: false },
|
||||
{ name: "scalar-context", context: "chat:slack:fixture", matches: false },
|
||||
{ name: "array-context", context: [context], matches: false },
|
||||
];
|
||||
const expected = [];
|
||||
// Preserve stored legacy shapes: the insert trigger otherwise canonicalizes
|
||||
// uppercase JSON identifiers into the first-class UUID privacy binding.
|
||||
await db.$client.unsafe("alter table heartbeat_runs disable trigger heartbeat_runs_set_scope_kind");
|
||||
try { for (const item of cases) {
|
||||
const runId = uuid();
|
||||
await db.$client.unsafe(`insert into heartbeat_runs
|
||||
(id, company_id, agent_id, status, scope_kind, issue_id, native_issue_id, context_snapshot, created_at, updated_at)
|
||||
values ($1, $2, $3, 'queued', $4, $5, $6, $7::jsonb, $8, $8)`,
|
||||
[runId, conversation.company_id, conversation.assigned_agent_id, "runIssueId" in item ? "issue" : "company",
|
||||
"runIssueId" in item ? item.runIssueId : null, "nativeIssueId" in item ? item.nativeIssueId : null,
|
||||
JSON.stringify(item.context), iso(ANCHOR_MS - 1000)]);
|
||||
expected.push({ name: item.name, runId, matches: item.matches });
|
||||
}
|
||||
} finally { await db.$client.unsafe("alter table heartbeat_runs enable trigger heartbeat_runs_set_scope_kind"); }
|
||||
return expected;
|
||||
}
|
||||
async function capture(db: Db, module: typeof treatment): Promise<Record<string, Query>> {
|
||||
const queries: Query[] = [];
|
||||
const stop = new Error("captured");
|
||||
const captureDb = new Proxy(db, { get(target, key) {
|
||||
if (key !== "select") return Reflect.get(target, key);
|
||||
return (...fields: Parameters<Db["select"]>) => {
|
||||
const builder = target.select(...fields), from = builder.from.bind(builder);
|
||||
builder.from = ((...args: Parameters<typeof from>) => {
|
||||
const selection = from(...args);
|
||||
selection.execute = async () => {
|
||||
queries.push(selection.toSQL());
|
||||
if (queries.length === 2) throw stop;
|
||||
return [];
|
||||
};
|
||||
return selection;
|
||||
}) as typeof builder.from;
|
||||
return builder;
|
||||
};
|
||||
}});
|
||||
try { await module.enqueueChatRunMilestones(captureDb, { since: new Date(SINCE_MS), limit: cursorWindow ? 1 : 200 }); }
|
||||
catch (error) { if (error !== stop) throw error; }
|
||||
assert.equal(queries.length, 2, "capture both real candidate queries before any publication writes");
|
||||
return { milestone: queries[0], native: queries[1] };
|
||||
}
|
||||
type Cursor = { updatedAt: Date; runId: string; conversationId: string };
|
||||
async function captureCursor(db: Db, module: typeof treatment, cursor: Cursor): Promise<Query> {
|
||||
let query: Query | undefined;
|
||||
let executions = 0;
|
||||
const stop = new Error("captured cursor statement");
|
||||
const captureDb = new Proxy(db, { get(target, key) {
|
||||
if (key !== "select") return Reflect.get(target, key);
|
||||
return (...fields: Parameters<Db["select"]>) => {
|
||||
const builder = target.select(...fields), from = builder.from.bind(builder);
|
||||
builder.from = ((...args: Parameters<typeof from>) => {
|
||||
const selection = from(...args);
|
||||
selection.execute = async () => {
|
||||
if (++executions === 2) { query = selection.toSQL(); throw stop; }
|
||||
// Advance only the real service's keyset loop. This non-publishable
|
||||
// placeholder page avoids resolving origins or writing publications.
|
||||
return Array.from({ length: 25 }, () => ({ runStatus: "benchmark_skip", runUpdatedAt: cursor.updatedAt,
|
||||
runId: cursor.runId, conversationId: cursor.conversationId }));
|
||||
};
|
||||
return selection;
|
||||
}) as typeof builder.from;
|
||||
return builder;
|
||||
};
|
||||
}});
|
||||
try { await module.enqueueChatRunMilestones(captureDb, { since: new Date(SINCE_MS), limit: 1 }); }
|
||||
catch (error) { if (error !== stop) throw error; }
|
||||
assert.ok(query, "real second-page service statement was not captured");
|
||||
return query;
|
||||
}
|
||||
async function seedCursorSibling(db: Db) {
|
||||
const [original] = await db.$client.unsafe(`select conversation.* from chat_conversations conversation
|
||||
join heartbeat_runs run on run.company_id = conversation.company_id and run.issue_id = conversation.issue_id
|
||||
where run.context_snapshot->>'fixtureHistory' = 'true' order by conversation.id limit 1`);
|
||||
await db.$client.unsafe(`insert into chat_conversations
|
||||
(id, company_id, endpoint_id, issue_id, external_conversation_id, external_label, state)
|
||||
values ($1, $2, $3, $4, 'cursor-sibling', 'Cursor sibling', 'active')`,
|
||||
[uuid(), original.company_id, original.endpoint_id, original.issue_id]);
|
||||
}
|
||||
async function selectLateCursor(db: Db, fraction = 0.9): Promise<Cursor> {
|
||||
const [row] = await db.$client.unsafe(`select chosen.updated_at, chosen.id as run_id, conversation.id as conversation_id
|
||||
from (select id, company_id, issue_id, updated_at from heartbeat_runs
|
||||
where context_snapshot->>'fixtureHistory' = 'true'
|
||||
order by updated_at, id offset ${Math.floor(historyRuns * fraction)} limit 1) chosen
|
||||
join chat_conversations conversation on conversation.company_id = chosen.company_id and conversation.issue_id = chosen.issue_id
|
||||
where conversation.state = 'active' order by conversation.id limit 1`);
|
||||
assert.ok(row);
|
||||
return { updatedAt: new Date(row.updated_at), runId: row.run_id, conversationId: row.conversation_id };
|
||||
}
|
||||
const median = (values: number[]) => values.slice().sort((a, b) => a - b)[Math.floor(values.length / 2)];
|
||||
const keepAlive = setInterval(() => {}, 1000);
|
||||
let database: Awaited<ReturnType<typeof startDatabase>> | undefined;
|
||||
let db: Db | undefined;
|
||||
let baselineDir: string | undefined;
|
||||
async function startDatabase() {
|
||||
if (!dockerPostgres) return startEmbeddedPostgresTestDatabase("chat-milestone-benchmark-");
|
||||
const container = `paperclip-chat-milestones-${randomUUID()}`;
|
||||
const docker = (...args: string[]) => execFileSync("docker", args, { encoding: "utf8", stdio: ["ignore", "pipe", "pipe"] }).trim();
|
||||
docker("run", "--detach", "--rm", "--name", container, "--label", "paperclip.benchmark=chat-milestones", "--publish", "127.0.0.1::5432",
|
||||
"--env", "POSTGRES_USER=paperclip", "--env", "POSTGRES_PASSWORD=paperclip", "--env", "POSTGRES_DB=paperclip", "postgres:17");
|
||||
const cleanup = async () => { docker("rm", "--force", container); };
|
||||
try {
|
||||
let ready = false;
|
||||
for (let attempt = 0; attempt < 100; attempt++) {
|
||||
try { docker("exec", container, "pg_isready", "-h", "127.0.0.1", "-U", "paperclip"); ready = true; break; }
|
||||
catch { await new Promise(resolve => setTimeout(resolve, 100)); }
|
||||
}
|
||||
assert.ok(ready, "Disposable Docker PostgreSQL did not become ready");
|
||||
const port = docker("port", container, "5432/tcp").split(":").at(-1);
|
||||
const connectionString = `postgres://paperclip:paperclip@127.0.0.1:${port}/paperclip`;
|
||||
await applyPendingMigrations(connectionString);
|
||||
return { connectionString, cleanup };
|
||||
} catch (error) { await cleanup(); throw error; }
|
||||
}
|
||||
try {
|
||||
database = await startDatabase();
|
||||
db = createDb(database.connectionString);
|
||||
const modulePath = "server/src/services/chat-run-publications.ts";
|
||||
const moduleUrl = new URL(`../${modulePath}`, import.meta.url);
|
||||
const oldSource = execFileSync("git", ["show", `${baselineRevision}:${modulePath}`], { encoding: "utf8" });
|
||||
const rewritten = oldSource.replace(/from "(\.{1,2}\/[^\"]+)"/g,
|
||||
(_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, moduleUrl)))}`);
|
||||
baselineDir = await fs.mkdtemp(path.join(fileURLToPath(new URL("../server/scripts/", import.meta.url)), ".chat-milestone-baseline-"));
|
||||
const baselineFile = path.join(baselineDir, "chat-run-publications.ts");
|
||||
await fs.writeFile(baselineFile, rewritten);
|
||||
const baseline = await import(pathToFileURL(baselineFile).href) as typeof treatment;
|
||||
await seed(db);
|
||||
if (denseChat) await makeDenseNativeChatHistory(db);
|
||||
if (cursorWindow) await seedCursorSibling(db);
|
||||
await seedSourceMessages(db);
|
||||
const bindingCases = await seedBindingCases(db);
|
||||
const queries = { baseline: await capture(db, baseline), treatment: await capture(db, treatment) };
|
||||
const lateCursor = cursorWindow ? await selectLateCursor(db) : undefined;
|
||||
const nearTailCursor = cursorWindow ? await selectLateCursor(db, 0.9999) : undefined;
|
||||
let emptyCursor: Cursor | undefined;
|
||||
if (lateCursor) {
|
||||
queries.baseline.cursor = await captureCursor(db, baseline, lateCursor);
|
||||
queries.treatment.cursor = await captureCursor(db, treatment, lateCursor);
|
||||
}
|
||||
if (nearTailCursor) {
|
||||
queries.baseline.tail = await captureCursor(db, baseline, nearTailCursor);
|
||||
queries.treatment.tail = await captureCursor(db, treatment, nearTailCursor);
|
||||
const finalPage = await db.$client.unsafe(queries.baseline.tail.sql, queries.baseline.tail.params as never[]).values();
|
||||
assert.ok(finalPage.length > 0 && finalPage.length < 25, "tail cursor must produce a short final page");
|
||||
const last = finalPage.at(-1)!;
|
||||
emptyCursor = { updatedAt: new Date(last[3]), runId: last[0], conversationId: last[7] };
|
||||
queries.baseline.empty = await captureCursor(db, baseline, emptyCursor);
|
||||
queries.treatment.empty = await captureCursor(db, treatment, emptyCursor);
|
||||
}
|
||||
const results = [];
|
||||
for (const kind of cursorWindow ? ["milestone", "cursor", "tail", "empty", "native"] : ["milestone", "native"]) {
|
||||
const samples = { baseline: [] as number[], treatment: [] as number[] };
|
||||
const rows = { baseline: [] as unknown[], treatment: [] as unknown[] };
|
||||
for (let round = 0; round < (correctnessOnly ? 1 : 7); round++) {
|
||||
const order = round % 2 === 0 ? ["baseline", "treatment"] as const : ["treatment", "baseline"] as const;
|
||||
for (const side of order) {
|
||||
const query = queries[side][kind], start = performance.now();
|
||||
rows[side] = await db.$client.unsafe(query.sql, query.params as never[]).values();
|
||||
if (round >= 2) samples[side].push(performance.now() - start);
|
||||
}
|
||||
assert.deepEqual(rows.treatment, rows.baseline, `${kind}: complete ordered positional rows differ`);
|
||||
}
|
||||
if (kind === "milestone" && !denseChat && !cursorWindow) {
|
||||
const runIds = new Set(rows.treatment.map(row => (row as unknown[])[0]));
|
||||
for (const item of bindingCases) assert.equal(runIds.has(item.runId), item.matches, `binding semantics: ${item.name}`);
|
||||
}
|
||||
const expectedCursor = kind === "cursor" ? lateCursor : kind === "tail" ? nearTailCursor : undefined;
|
||||
if (expectedCursor) {
|
||||
const first = rows.treatment[0] as unknown[];
|
||||
assert.equal(first[0], expectedCursor.runId, "late page must preserve the other conversation for the same run");
|
||||
assert.ok(String(first[7]) > expectedCursor.conversationId);
|
||||
}
|
||||
if (kind === "empty") assert.equal(rows.treatment.length, 0, "finished cursor must produce an empty page");
|
||||
else assert.ok(rows.baseline.length > 0, `${kind}: equality must include actual candidates`);
|
||||
const plans: Record<string, unknown> = {};
|
||||
for (const side of ["baseline", "treatment"] as const) {
|
||||
const query = queries[side][kind];
|
||||
const [plan] = await db.$client.unsafe(`EXPLAIN (${correctnessOnly ? "" : "ANALYZE, BUFFERS, "}FORMAT JSON) ${query.sql}`, query.params as never[]);
|
||||
plans[side] = plan["QUERY PLAN"];
|
||||
}
|
||||
results.push({ kind, equalRows: rows.baseline.length, baselineMedianMs: median(samples.baseline), treatmentMedianMs: median(samples.treatment),
|
||||
speedup: median(samples.baseline) / median(samples.treatment), samples, plans });
|
||||
}
|
||||
const sourceHashes = Object.fromEntries(await Promise.all([
|
||||
"server/src/services/chat-run-publications.ts", "server/src/services/chat-run-milestone-candidates.ts", "scripts/benchmark-chat-milestones.ts",
|
||||
].map(async file => [file, createHash("sha256").update(await fs.readFile(path.join(repositoryRoot, file))).digest("hex")])));
|
||||
const output = { sourceHashes, baselineRevision, treatmentRevision: execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8" }).trim(),
|
||||
candidateSource: await fs.readFile(new URL("../server/src/services/chat-run-milestone-candidates.ts", import.meta.url), "utf8"),
|
||||
treatmentDiff: execFileSync("git", ["-C", repositoryRoot, "diff", "--", "server/src/services/chat-run-publications.ts"], { encoding: "utf8" }),
|
||||
postgres: await db.$client.unsafe("select version(), current_setting('shared_buffers') as shared_buffers, current_setting('work_mem') as work_mem"),
|
||||
fixture: { runs: historyRuns + 215, bindingCases, bindingCasesChecked: !denseChat && !cursorWindow, denseNativeChat: denseChat,
|
||||
cursorWindow, historyRuns, requestedLimit: cursorWindow ? 1 : 200, lateCursor, nearTailCursor, emptyCursor,
|
||||
conversationIssues: 55, promptKiB: PROMPT_KIB, windowHours: 24 }, queries, results };
|
||||
const reportPath = path.resolve(process.argv.find(arg => arg.startsWith("--report="))?.slice(9) ?? ".scratch/chat-milestones-report.json");
|
||||
await fs.mkdir(path.dirname(reportPath), { recursive: true });
|
||||
await fs.writeFile(reportPath, JSON.stringify(output, null, 2));
|
||||
console.log(JSON.stringify({ reportPath, ...output, queries: undefined, treatmentDiff: undefined, candidateSource: undefined, results: results.map(({ plans, ...result }) => result) }, null, 2));
|
||||
} finally {
|
||||
try {
|
||||
await db?.$client.end({ timeout: 0 });
|
||||
await database?.cleanup();
|
||||
if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true });
|
||||
} finally { clearInterval(keepAlive); }
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, asc, gte, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
agents, chatConversations, chatEndpoints, companies, createDb, heartbeatRuns,
|
||||
issues, toolApplications, toolConnections,
|
||||
} from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { chatRunMilestoneSource } from "../services/chat-run-milestone-candidates.js";
|
||||
import { boundRunIssueIdSql } from "../services/run-task-binding.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("chat milestone candidate prefilter", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("chat-milestone-candidates-");
|
||||
db = createDb(database.connectionString);
|
||||
}, 90_000);
|
||||
afterAll(async () => {
|
||||
await db?.$client.end({ timeout: 0 });
|
||||
await database?.cleanup();
|
||||
});
|
||||
|
||||
it("preserves complete rows, binding precedence, company scope and the recovery window", async () => {
|
||||
const companyId = randomUUID(), foreignCompanyId = randomUUID(), agentId = randomUUID();
|
||||
const issueId = randomUUID(), foreignIssueId = randomUUID(), closedIssueId = randomUUID();
|
||||
const applicationId = randomUUID(), connectionId = randomUUID(), endpointId = randomUUID();
|
||||
const since = new Date("2026-10-20T00:00:00.000Z");
|
||||
await db.insert(companies).values([
|
||||
{ id: companyId, name: "Chat candidates", issuePrefix: "CHAT" },
|
||||
{ id: foreignCompanyId, name: "Foreign", issuePrefix: "FRGN" },
|
||||
]);
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Chat agent" });
|
||||
await db.insert(issues).values([
|
||||
{ id: issueId, companyId, title: "Active" },
|
||||
{ id: closedIssueId, companyId, title: "Closed" },
|
||||
{ id: foreignIssueId, companyId: foreignCompanyId, title: "Foreign" },
|
||||
]);
|
||||
await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Slack", type: "native" });
|
||||
await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Slack", uid: connectionId, transport: "chat_sdk", connectionPurpose: "channel" });
|
||||
await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, provider: "slack", publicId: endpointId, assignedAgentId: agentId });
|
||||
await db.insert(chatConversations).values([
|
||||
{ companyId, endpointId, issueId, externalConversationId: "active", externalLabel: "Active", state: "waiting" },
|
||||
{ companyId, endpointId, issueId: closedIssueId, externalConversationId: "closed", externalLabel: "Closed", state: "completed" },
|
||||
]);
|
||||
const cases = [
|
||||
{ context: { issueId }, matches: true },
|
||||
{ context: { issueId: ` ${issueId} ` }, matches: true },
|
||||
{ context: { issueId: " ", taskId: issueId }, matches: true },
|
||||
{ context: { issueId: 12, taskId: issueId }, matches: true },
|
||||
{ context: { issueId: { id: issueId }, taskId: issueId }, matches: true },
|
||||
{ context: { issueId: [issueId], taskId: issueId }, matches: true },
|
||||
{ context: { issueId: issueId.toUpperCase() }, matches: false },
|
||||
{ context: { issueId: "malformed", taskId: issueId }, matches: false },
|
||||
{ context: { issueId: "malformed" }, nativeIssueId: issueId, matches: true },
|
||||
{ context: { issueId: "malformed" }, runIssueId: issueId, matches: true },
|
||||
{ context: { issueId }, nativeIssueId: foreignIssueId, runIssueId: issueId, matches: false },
|
||||
{ context: { issueId: foreignIssueId }, matches: false },
|
||||
{ context: { issueId: closedIssueId }, matches: false },
|
||||
{ context: null, matches: false },
|
||||
{ context: "scalar", matches: false },
|
||||
{ context: [{ issueId }], matches: false },
|
||||
{ context: { issueId }, updatedAt: new Date(since.getTime() - 1), matches: false },
|
||||
];
|
||||
const expectedIds: string[] = [];
|
||||
// Historical row shapes must survive the fixture's insert-time UUID
|
||||
// normalization so this tests legacy reads, including uppercase text.
|
||||
await db.$client.unsafe("alter table heartbeat_runs disable trigger heartbeat_runs_set_scope_kind");
|
||||
try {
|
||||
for (const item of cases) {
|
||||
const runId = randomUUID();
|
||||
await db.$client.unsafe(`insert into heartbeat_runs
|
||||
(id, company_id, agent_id, status, scope_kind, issue_id, native_issue_id, context_snapshot, updated_at)
|
||||
values ($1, $2, $3, 'running', $4, $5, $6, $7::jsonb, $8)`,
|
||||
[runId, companyId, agentId, "runIssueId" in item ? "issue" : "company",
|
||||
"runIssueId" in item ? item.runIssueId : null, "nativeIssueId" in item ? item.nativeIssueId : null,
|
||||
JSON.stringify(item.context), (item.updatedAt ?? since).toISOString()]);
|
||||
if (item.matches) expectedIds.push(runId);
|
||||
}
|
||||
} finally {
|
||||
await db.$client.unsafe("alter table heartbeat_runs enable trigger heartbeat_runs_set_scope_kind");
|
||||
}
|
||||
const fields = {
|
||||
id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, updatedAt: heartbeatRuns.updatedAt,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot, resultJson: heartbeatRuns.resultJson,
|
||||
};
|
||||
const original = await db.select(fields).from(heartbeatRuns).where(and(
|
||||
gte(heartbeatRuns.updatedAt, since),
|
||||
sql`exists (select 1 from ${chatConversations}
|
||||
where ${chatConversations.companyId} = ${heartbeatRuns.companyId}
|
||||
and ${chatConversations.issueId}::text = ${boundRunIssueIdSql()}
|
||||
and ${chatConversations.state} in ('active', 'waiting'))`,
|
||||
)).orderBy(asc(heartbeatRuns.id));
|
||||
const filtered = await db.select(fields).from(chatRunMilestoneSource(since)).orderBy(asc(heartbeatRuns.id));
|
||||
expect(filtered).toEqual(original);
|
||||
expect(filtered.map(row => row.id)).toEqual(expectedIds.sort());
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,42 @@
|
||||
import { sql, Subquery } from "drizzle-orm";
|
||||
import { chatConversations, heartbeatRuns } from "@paperclipai/db";
|
||||
import { boundRunIssueIdSql } from "./run-task-binding.js";
|
||||
|
||||
/**
|
||||
* Match the cheap chat binding before evaluating publication lineage. Without
|
||||
* this fence PostgreSQL can read/decompress a run's context many times for the
|
||||
* origin predicates even when the run has no matching conversation at all.
|
||||
* This is a prefilter only: the caller still owns every publication check.
|
||||
*/
|
||||
export function chatRunMilestoneSource(since: Date) {
|
||||
const selectedFields = {
|
||||
id: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status,
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
updatedAt: heartbeatRuns.updatedAt,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId,
|
||||
issueId: heartbeatRuns.issueId,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
};
|
||||
return new Subquery(sql`
|
||||
with chat_milestone_runs as materialized (
|
||||
select ${heartbeatRuns.id}, ${heartbeatRuns.companyId},
|
||||
${heartbeatRuns.agentId}, ${heartbeatRuns.status},
|
||||
${heartbeatRuns.errorCode}, ${heartbeatRuns.updatedAt},
|
||||
${heartbeatRuns.nativeIssueId}, ${heartbeatRuns.issueId},
|
||||
${heartbeatRuns.contextSnapshot}, ${heartbeatRuns.resultJson}
|
||||
from ${heartbeatRuns}
|
||||
where ${heartbeatRuns.updatedAt} >= ${since.toISOString()}
|
||||
and exists (
|
||||
select 1 from ${chatConversations}
|
||||
where ${chatConversations.companyId} = ${heartbeatRuns.companyId}
|
||||
and ${chatConversations.issueId}::text = ${boundRunIssueIdSql()}
|
||||
and ${chatConversations.state} in ('active', 'waiting')
|
||||
)
|
||||
)
|
||||
select * from chat_milestone_runs
|
||||
`, selectedFields, "heartbeat_runs");
|
||||
}
|
||||
@@ -36,6 +36,7 @@ import { hasChatRunOwnedProviderInteraction } from "./chat-interaction-arbitrati
|
||||
import { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON } from "./heartbeat-run-summary.js";
|
||||
import { resolveChatOriginPublicationBindings } from "./issues.js";
|
||||
import { boundRunIssueIdSql } from "./run-task-binding.js";
|
||||
import { chatRunMilestoneSource } from "./chat-run-milestone-candidates.js";
|
||||
import { authorizeNativeChatReviewPresentation } from "./native-runtime/native-chat-review-presentation.js";
|
||||
import {
|
||||
SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
|
||||
@@ -826,7 +827,7 @@ export async function enqueueChatRunMilestones(
|
||||
conversationId: chatConversations.id,
|
||||
agentName: agents.name,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.from(chatRunMilestoneSource(since))
|
||||
.innerJoin(
|
||||
chatConversations,
|
||||
and(
|
||||
|
||||
Reference in new issue
Block a user