diff --git a/doc/chat-milestone-query-performance.md b/doc/chat-milestone-query-performance.md new file mode 100644 index 00000000000..b9cf5be8d17 --- /dev/null +++ b/doc/chat-milestone-query-performance.md @@ -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" +``` diff --git a/scripts/benchmark-chat-milestones.ts b/scripts/benchmark-chat-milestones.ts new file mode 100644 index 00000000000..170f17dc284 --- /dev/null +++ b/scripts/benchmark-chat-milestones.ts @@ -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="); +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> { + 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) => { + const builder = target.select(...fields), from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + 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 { + 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) => { + const builder = target.select(...fields), from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + 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 { + 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> | 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 = {}; + 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); } +} diff --git a/server/src/__tests__/chat-run-milestone-candidates.test.ts b/server/src/__tests__/chat-run-milestone-candidates.test.ts new file mode 100644 index 00000000000..fafa3086742 --- /dev/null +++ b/server/src/__tests__/chat-run-milestone-candidates.test.ts @@ -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>; + let db: ReturnType; + 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()); + }); +}); diff --git a/server/src/services/chat-run-milestone-candidates.ts b/server/src/services/chat-run-milestone-candidates.ts new file mode 100644 index 00000000000..eff81e01279 --- /dev/null +++ b/server/src/services/chat-run-milestone-candidates.ts @@ -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"); +} diff --git a/server/src/services/chat-run-publications.ts b/server/src/services/chat-run-publications.ts index c3d3bc0d478..93fe111ec2d 100644 --- a/server/src/services/chat-run-publications.ts +++ b/server/src/services/chat-run-publications.ts @@ -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(