From a91f78e7e08928d8da572b747f435bd3d23de94f Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 07:19:22 -0500 Subject: [PATCH] perf(chat): reduce inbound queue discovery cost (#17971) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip helps people manage companies of AI agents. > - External chat messages enter a durable delivery queue. > - Queue reads must preserve pending wake ownership and conversation order. > - Retained processed history increases the estimated cost of a correlated pending-action test. > - PostgreSQL 17 enables JIT for these small selectors and reads extra JSON in deadline discovery. > - This pull request uses a scoped pending-action set for open-work discovery. > - The benefit is less database work with the same selected deliveries and deadlines. ## Linked Issues or Issue Description **What existing behavior does this improve?** Inbound delivery deadline discovery and conversation-head selection. **Subsystem affected** Server chat queue discovery. **Current behavior** A correlated `EXISTS` inside an `OR` has a high cost estimate on retained completed history. PostgreSQL 17 already hashes its pending-action read, but the estimate still enables JIT. Deadline discovery also reads JSON from completed deliveries that have no pending wake. **Proposed behavior** Use `(company_id, endpoint_id, delivery_id)` membership in the pending-action set for the two open-work selectors. Keep ready-only evaluation on its original scoped `EXISTS`. Keep retry, lease, ordering, lifecycle and endpoint rules. **Reason and benefit** The local retained-history fixture reduces deadline selection time. The small control and the final head selection are neutral. Ready-only selection remains an unchanged control. **Breaking changes** None. No API, schema, migration, stored history, worker, timer or notification changes. Related work: #17647 and #15927 change inbound wake-notice scheduling and selection. This PR changes the separate conversation queue ownership predicate. ## What Changed - Use a fully scoped pending-action set for open-work discovery. Exclude null delivery IDs to preserve exact boolean results. Keep duplicate actions from multiplying deliveries. - Preserve the original ready-only query. - Add five real PostgreSQL tests and a reproducible baseline/treatment benchmark. Capture the complete real service SQL, parameters, positional rows and execution plans. - Document the query cause, rejected hypotheses, test results and limits in `doc/plans/2026-10-27-inbound-queue-query-proof.md`. ## Verification Pinned pre-fix baseline: `09ad9adce2cf4742944353a3707a2f908e9bb35e`. Local fixture: Linux ARM64 PostgreSQL 17.11, Node 26.4.0. It has 1500 completed deliveries and actions, two open deliveries, and 32 KiB of synthetic incompressible JSON per historical delivery. A separate control has 20 completed rows. The normal settings use JIT on with a 100000 cost threshold, 128 MiB shared buffers and 4 MiB work_mem. The final quiet retained-history pair measured these local SQL medians. Each arm has two warmups and five alternating samples. The small control is neutral. | Actual selector | Baseline | Treatment | Complete positional output | | --- | ---: | ---: | --- | | Next inbound deadline | 18.78 ms | 1.58 ms | Equal | | Conversation head | 5.71 ms | 6.18 ms | Equal | | Ordinary ready control | 9.23 ms | 8.33 ms | Equal ; unchanged | The deadline plan uses fewer shared buffers. Head physical reads remain unchanged. Deadline discovery avoids JIT on this fixture. The pristine head estimate is below the JIT threshold in both arms, and its timing is neutral. With JIT disabled, deadline discovery still improves; head and ready timings overlap. - Red: the unchanged baseline fails the explicit local fixture work budget for JIT and deadline buffer reads. Green: the treatment passes. There is no timing threshold in ordinary CI. - Five new PostgreSQL tests and six existing assembled chat tests pass. They cover scoped/null/duplicate pending actions, boolean results, payloads, head barriers, reactions, paused/attention/AgentMail exclusions, retry/processing deadlines, provider ordering, lease loss, lifecycle retry and edit/admission races. - Actual old/new ready selectors have equal error outcomes for unrelated malformed retry timestamps. An initial AND-only oracle comparison was inconclusive and is not defect evidence. - `pnpm -r typecheck` passes. - `pnpm build` passes. - Final head `f3cac298d8a8a5377127b0db7057c52019bbe426` passes the complete [CI workflow](https://github.com/paperclipai/paperclip/actions/runs/40625611887). Greptile is 5/5 on that head with no unresolved review threads. The earlier end-to-end cleanup lock-timeout failure passed on this revision. - The complete local suite was not run. The host's known native PostgreSQL startup failure exhausts SysV semaphores. Focused integration tests use a disposable Docker PostgreSQL constructor through the existing test seam. Complete final-head CI is required. Reproduce the real SQL comparison: ```sh pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-inbound-queue-discovery.ts --baseline-ref=09ad9adce2cf4742944353a3707a2f908e9bb35e --require-work-budget --retained-only ``` Use `--jit-off` for the diagnostic control. The default command uses the repository's bundled PostgreSQL version and records it in output. The reported 17.11 run used the existing test constructor seam with a disposable `postgres:17` container. Raw samples, complete queries, parameters and plans are emitted as JSON. The benchmark uses disposable synthetic data and does not read a live database URL. ## Risks - Query plans depend on data and PostgreSQL settings. These local measurements do not prove deployed speed or request latency. - Head selection still reads historical JSON. Action history remains stored and scanned. This is not a constant-time queue implementation. - The changed false branch has no retry timestamp cast. Ready-only parsing, claims and retries stay on their existing query. - Background SQL totals can overlap. Do not add these savings into an HTTP latency claim. The largest possible saving for an affected selection is bounded by its observed SQL time, about 49–53 ms on average in discovery. ## Model Used OpenAI Codex, GPT-6 family, with tool use and code execution. The runtime does not expose the exact model ID or context-window size. Three parallel agents own separate performance fixes; this PR contains only the inbound queue fix. ## 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 --- .../2026-10-27-inbound-queue-query-proof.md | 168 ++++++++++++++++++ scripts/benchmark-inbound-queue-discovery.ts | 134 ++++++++++++++ .../helpers/inbound-discovery-fixture.ts | 102 +++++++++++ .../__tests__/inbound-queue-discovery.test.ts | 96 ++++++++++ .../chat-channels/inbound/delivery-ledger.ts | 9 + 5 files changed, 509 insertions(+) create mode 100644 doc/plans/2026-10-27-inbound-queue-query-proof.md create mode 100644 scripts/benchmark-inbound-queue-discovery.ts create mode 100644 server/src/__tests__/helpers/inbound-discovery-fixture.ts create mode 100644 server/src/__tests__/inbound-queue-discovery.test.ts diff --git a/doc/plans/2026-10-27-inbound-queue-query-proof.md b/doc/plans/2026-10-27-inbound-queue-query-proof.md new file mode 100644 index 00000000000..da15df6d3a8 --- /dev/null +++ b/doc/plans/2026-10-27-inbound-queue-query-proof.md @@ -0,0 +1,168 @@ +# Inbound queue query proof + +## Contract and cause + +Paperclip's inbound queue discovers its next deadline, selects each conversation +head, and selects ordinary ready deliveries. A retained processed delivery can +still own the head while an issued or processing `inbound_wakeup` action remains. +This work keeps that obligation, company and endpoint scope, lease expiry, +retry timestamps, ordering keys, lifecycle dependencies, and provider exclusions. +It adds no cache, migration, producer, worker, or timer. + +The two open-work selectors used a correlated `EXISTS` inside an `OR`. PostgreSQL +17.11 already used a hashed subplan in the measured fixture. The original +hypothesis of repeated physical action scans was wrong. The estimate still +included a large correlated alternative. It can enable JIT at the normal 100000 +cost threshold. The deadline query also evaluated JSON conditions on retained +processed history before rejecting rows with no pending action. + +The treatment uses a set of `(company_id, endpoint_id, delivery_id)` for pending +ownership. Null delivery IDs cannot match and are excluded, so the predicate +still returns a boolean. Duplicate actions do not multiply delivery rows. +Ready-only probes retain their original scoped `EXISTS`. This keeps the +readiness and retry parsing boundary outside this optimization. + +## Reproduction + +Run from the source checkout with its dependencies built: + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-inbound-queue-discovery.ts --baseline-ref=09ad9adce2cf4742944353a3707a2f908e9bb35e --require-work-budget +``` + +The harness starts disposable PostgreSQL through the repository test helper. +It does not read a live database URL. It loads the pinned pre-fix queue, +entry-point, and ledger sources. Other runtime helpers and dependencies must +match. It captures the complete SQL and parameters through the real service, +including projections, ordering, limits, and the head probe's lease path. +It stops before provider effects. Every selected column is compared as a +positional row on one reserved SQL connection. Fixture deadlines are fixed. The small control has 20 completed deliveries +and actions. The retained-history case has 1500 of each, two open deliveries, +and 32 KiB of synthetic incompressible message text per historical delivery. + +Two warmups precede five alternating samples per arm. Plans use +`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)` outside the timed samples. The explicit +fixture work budget rejects JIT for the two changed selectors and more than +1000 shared buffer accesses (hits plus reads) for the deadline selection. It has no CI timing threshold. +Use `--jit-off` for a diagnostic control. + +Local evidence uses Node 26.4.0 and Linux ARM64 PostgreSQL 17.11 in a disposable +Docker container, matching the hosted major and minor version. The normal +settings are shared_buffers 128 MiB, work_mem 4 MiB, JIT on, jit_above_cost 100000, +and random_page_cost 4. Native macOS PostgreSQL could not start because the host +had exhausted its SysV semaphore allocation. The scratch provider replaces only +the test PostgreSQL constructor. Schema migrations, clients and query execution +remain real. No persistent database is used. + +## Results + +Matched pre-fix baseline: `09ad9adce2cf4742944353a3707a2f908e9bb35e`. +An earlier baseline at `96727b3` had identical ledger and conversation-queue +sources. Master changed an unrelated health notification import in the service +entry-point, so the final comparison was repeated from the newer pre-fix base. +The ordinary-ready selector is an unchanged control. The head query still reads +its existing JSON ordering/filter fields; do not claim it reduces physical +history scans. Local SQL timings do not prove a hosted request speedup. + +Final quiet PostgreSQL 17.11 comparison, in milliseconds. Each arm has two +warmups and five alternating samples. Parentheses show minimum–maximum. + +| Actual selector | Baseline median (range) | Treatment median (range) | Ratio | Equality | +| --- | ---: | ---: | ---: | --- | +| deadline | 18.78 (18.43–20.41) | 1.58 (1.14–1.75) | 11.90x | 2 complete rows equal | +| head | 5.71 (5.38–6.40) | 6.18 (5.56–6.72) | 0.92x | 1 complete rows equal | +| ready | 9.23 (7.97–9.41) | 8.33 (7.72–9.21) | 1.11x | 1 complete rows equal | + +- deadline: shared buffer hits 10662 → 455; JIT present → absent. +- head: shared buffer hits 10600 → 10600; JIT absent → absent. +- ready: shared buffer hits 66 → 66; JIT present → present. + +The matched small control is neutral (overlapping sample ranges). The JIT-off +control remains diagnostic under other local build load: deadline 9.72 → 2.17 ms; +head 9.32 → 9.08 ms; ordinary ready 1.52 → 1.56 ms. The last two ranges overlap. +It separates deadline physical work from an earlier JIT-only head improvement. +The earlier small-then-large fixture put the head estimate at 100369.79 and +enabled JIT. The final pristine retained-only fixture put it at 98866.79, below +the threshold. Its head timing is neutral. Plan-cost threshold crossings are +data-sensitive; do not claim a head speedup for the final fixture. + +## Feedback loop + +| Question | Loop | Measured elapsed | Decision | +| --- | --- | --- | --- | +| Can the real source run locally? | Cached offline install, isolated build | 27.7 s install; 141 s build log span | Reuse the built checkout for each narrow attempt | +| Does the initial scan hypothesis hold? | Captured queries and real PG plans | 4.0 s inside the instrumented baseline harness, plus module loading | Reject repeated physical scans as the cause | +| Does the set improve the actual selectors? | Red work budget, alternating pair, JIT-off control | 28.36 s for all three processes, including module and DB setup | Exploratory variant identified JIT cost; final scope keeps ready-only unchanged | +| Does a selective ready probe remain scoped? | Original and treatment SQL with unrelated malformed retry data | 12 s focused attempt; 28 s after replay/module rebuild | Direct head probes return empty; actual OR selectors both return error 22007; keep readiness unchanged | + +Coordination waits for a quiet benchmark interval are separate from these command +measurements. They are not a reason to run noisy parallel timing samples. The +small real-database regression loop takes about 11 seconds including transforms +and startup. Full repository checks and exact-head CI remain required. + +## Scope of benefit + +The discovery sample attributed 4.731 summed SQL seconds to 90 deadline +selections and 4.681 seconds to 97 head selections over about five minutes. +The final pristine fixture proves a deadline improvement; the head is neutral +at that data point. The ordinary-ready control accounted +for a further 1.389 seconds over 90 calls. These are background selectors, not a +correlated request critical path. The maximum improvement to a single affected +selection cannot exceed its observed SQL duration (about 49–53 ms on average). +No end-to-end latency or rollout claim follows from adding those overlapping +aggregate times. The change is small and avoids a recurring cost under the +hosted planner defaults, so it is worth shipping with the other proven fixes. + +## Correctness and failure accounting + +Five real-PostgreSQL tests cover tuple scope, duplicate/null actions, exact +boolean results, full payloads, future head ownership, independent reactions, +paused/attention/agentmail exclusions, retry/processing deadlines, and provider +sequence ordering. Six existing assembled chat tests also pass. They cover +lease renewal failure, reverse-arrival Slack and Telegram events, lifecycle +retry, edit/admission races, and deadline restoration. + +An exploratory malformed-retry test compared an actual OR selector against an +AND-only oracle. That comparison was inconclusive. It must not be presented as +proof of an introduced defect. The complete actual old and new OR selectors +both return PostgreSQL error 22007 for unrelated malformed pending actions in +the measured fixture, including other-company and null-delivery controls. The +selective direct head probe returns empty on both versions. The final fix leaves +ready-only evaluation unchanged. + +The initial repeated-action-scan hypothesis was also rejected. Both arms scan +action history once in the measured plans. The benefits come from query costing +and filter order, rather than fewer physical action scans. + +## Final verification + +- `pnpm -r typecheck`: pass. +- `pnpm build`: pass. +- New and selected existing assembled PostgreSQL 17 suites: 11 passed. +- Complete local suite: not run because the native host PostgreSQL startup + blocker is already established; complete final-head CI remains the gate. +- Final retained-history benchmark work budget: pass, complete positional + equality for all three selectors. +- The final benchmark process took 34.19 seconds from + its instrumented setup through proof. This includes the coordination wait after + the fixture and queries were ready. The actual timing phase was brief. + +The PR contains the source, fixture and harness. Raw local JSON includes every +sample, query, parameter and plan. No hosted rollout was performed. + +## Additional failures retained + +Greptile's first review found two proof gaps: the budget counted only cache +hits, and the payload test read the row again instead of checking the captured +selection. Both are fixed. The budget includes hits plus reads. The test compares +the captured row's complete normalized payload with its seeded object. + +A prepared timing attempt lost its idle pooled connection while waiting. Its +now()-derived fixture value differed between queries, so equality stopped the +run before samples. The harness now reserves its timing connection and uses an +explicit follower deadline. The final quiet comparison passes. + +The first CI head failed e2e shard 3 in `agent-chat-projects` cleanup: cancelling +a heartbeat run timed out on a row lock and returned HTTP 500. This is outside +the changed queue selector. The final head must still pass that check. The +original log remains at https://github.com/paperclipai/paperclip/actions/runs/40624302918/job/122605509182. diff --git a/scripts/benchmark-inbound-queue-discovery.ts b/scripts/benchmark-inbound-queue-discovery.ts new file mode 100644 index 00000000000..9486a6736a5 --- /dev/null +++ b/scripts/benchmark-inbound-queue-discovery.ts @@ -0,0 +1,134 @@ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { createHash } from "node:crypto"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { createDb, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; +import { captureInboundQuery, runQuery, seedInboundDiscovery, type QueueFactory, type ServiceFactory } from "../server/src/__tests__/helpers/inbound-discovery-fixture.js"; + +const ref = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15); +assert.ok(ref, "Pass an explicit pre-change --baseline-ref="); +const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${ref}^{commit}`], { encoding: "utf8" }).trim(); +const root = fileURLToPath(new URL("../", import.meta.url)); +const sources = ["server/src/services/chat-channels/inbound/delivery-ledger.ts", "server/src/services/chat-channels/inbound/conversation-queue.ts", "server/src/services/chat-channels.ts"]; +const runtimeChanges = execFileSync("git", ["diff", "--name-only", baselineRevision, "--", "server/src/services", "packages/db/src", "packages/shared/src", "pnpm-lock.yaml"], { encoding: "utf8" }).trim().split("\n").filter(Boolean); +assert.deepEqual(runtimeChanges.filter(name => !sources.includes(name)), [], "Baseline must not import changed runtime helpers or dependencies"); +const dir = await fs.mkdtemp(path.join(root, "server/scripts/.inbound-baseline-")); +let database: Awaited> | undefined; +let db: Db | undefined; +const keepAlive = setInterval(() => {}, 1000); +function nodes(plan: Record): Record[] { + return [plan, ...((plan.Plans ?? []) as Record[]).flatMap(nodes)]; +} +const started = performance.now(); +try { + const targets = new Map(sources.map(source => [path.join(root, source).replace(/\.ts$/, ".js"), path.join(dir, path.basename(source))])); + for (const source of sources) { + const original = path.join(root, source); + const text = execFileSync("git", ["show", `${baselineRevision}:${source}`], { encoding: "utf8" }); + const rewritten = text.replace(/from "(\.{1,2}\/[^\"]+)"/g, (_match, specifier: string) => { + const resolved = path.resolve(path.dirname(original), specifier); + return `from ${JSON.stringify(targets.get(resolved) ?? resolved)}`; + }); + await fs.writeFile(path.join(dir, path.basename(source)), rewritten); + } + const oldQueue = await import(pathToFileURL(path.join(dir, "conversation-queue.ts")).href) as { createChatConversationDeliveryQueue: QueueFactory }; + const oldService = await import(pathToFileURL(path.join(dir, "chat-channels.ts")).href) as { chatChannelService: ServiceFactory }; + database = await startEmbeddedPostgresTestDatabase("inbound-discovery-benchmark-"); + db = createDb(database.connectionString, { maxConnections: 1, idleTimeoutSeconds: 0 }); + const [{ version }] = await db.$client.unsafe("select version()"); + const settings = await db.$client.unsafe("select name, setting from pg_settings where name in ('shared_buffers', 'work_mem', 'jit', 'jit_above_cost', 'jit_inline_above_cost', 'random_page_cost', 'max_connections') order by name"); + const results = []; + const malformedReadyControls = []; + let baselineExceedsWorkBudget = false; + const budgetFailures: string[] = []; + for (const fixture of [{ name: "small", history: 20, payloadBytes: 0 }, { name: "retained-history", history: 1500, payloadBytes: 32 * 1024 }].filter(value => !process.argv.includes("--retained-only") || value.name === "retained-history")) { + const f = await seedInboundDiscovery(db, fixture.history, fixture.payloadBytes); + await db.$client.unsafe("begin"); + if (process.argv.includes("--jit-off")) await db.$client.unsafe("set local jit = off"); + try { + const pairs = []; + for (const kind of ["deadline", "head", "ready"] as const) { + const before = await captureInboundQuery(db, f, kind, oldQueue.createChatConversationDeliveryQueue, oldService.chatChannelService); + const after = await captureInboundQuery(db, f, kind); + pairs.push({ kind, before, after }); + } + const connection = await db.$client.reserve(); + const timedDb = new Proxy(db, { get(target, key) { return key === "$client" ? connection : Reflect.get(target, key); } }); + try { + if (process.argv.includes("--wait-for-slot")) { + process.stderr.write(`READY ${fixture.name}: database seeded, all real queries captured; press Enter to time\n`); + await new Promise(resolve => { process.stdin.once("data", () => { process.stdin.pause(); resolve(); }); process.stdin.resume(); }); + } + for (const { kind, before, after } of pairs) { + const samples = { before: [] as number[], after: [] as number[] }; + const expected = await runQuery(timedDb, before); + assert.deepEqual(await runQuery(timedDb, after), expected, `${fixture.name}/${kind}: complete positional output differs`); + for (let n = 0; n < 7; n++) { + for (const arm of n % 2 ? ["after", "before"] as const : ["before", "after"] as const) { + const start = performance.now(); + await runQuery(timedDb, arm === "before" ? before : after); + if (n >= 2) samples[arm].push(performance.now() - start); + } + } + const plans = {} as Record<"before" | "after", unknown>; + const actionLoops = {} as Record<"before" | "after", number>; + for (const arm of ["before", "after"] as const) { + const query = arm === "before" ? before : after; + const [row] = await connection.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]); + plans[arm] = row["QUERY PLAN"]; + actionLoops[arm] = Math.max(0, ...nodes(row["QUERY PLAN"][0].Plan).filter(node => node["Relation Name"] === "chat_actions").map(node => Number(node["Actual Loops"]))); + } + if (fixture.name === "retained-history") { + const oldPlan = (plans.before as { JIT?: unknown; Plan: Record }[])[0]!; + const newPlan = (plans.after as { JIT?: unknown; Plan: Record }[])[0]!; + const oldBuffers = Number(oldPlan.Plan["Shared Hit Blocks"] ?? 0) + Number(oldPlan.Plan["Shared Read Blocks"] ?? 0); + const newBuffers = Number(newPlan.Plan["Shared Hit Blocks"] ?? 0) + Number(newPlan.Plan["Shared Read Blocks"] ?? 0); + baselineExceedsWorkBudget ||= Boolean(oldPlan.JIT) || (kind === "deadline" && oldBuffers > 1000); + if (kind !== "ready" && newPlan.JIT) budgetFailures.push(`${kind}: unnecessary JIT on the retained-history fixture`); + if (kind === "deadline" && newBuffers > 1000) budgetFailures.push(`${kind}: ${newBuffers} shared buffer accesses exceeds fixture budget 1000`); + } + const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!; + results.push({ fixture, kind, equalPositionalRows: expected.length, before, after, samples, beforeMedianMs: median(samples.before), afterMedianMs: median(samples.after), actionLoops, plans }); + process.stderr.write(`${fixture.name}/${kind}: ${median(samples.before).toFixed(2)} -> ${median(samples.after).toFixed(2)} ms; action scans ${actionLoops.before} -> ${actionLoops.after}\n`); + } + } finally { connection.release(); } + if (fixture.name === "small") { + const unrelated = await f.delivery({ state: "filtered" }); + const other = await seedInboundDiscovery(db, 0); + for (const input of [ + { scope: "same-company-unrelated-delivery", companyId: f.companyId, endpointId: f.endpointId, deliveryId: unrelated.id }, + { scope: "other-company", companyId: other.companyId, endpointId: other.endpointId, deliveryId: other.pending.id }, + { scope: "null-delivery", companyId: f.companyId, endpointId: f.endpointId, deliveryId: null }, + ]) { + const action = await f.action(input.deliveryId, { companyId: input.companyId, endpointId: input.endpointId, result: { retryAt: "invalid timestamp" } }); + const before = await captureInboundQuery(db, f, "ready", oldQueue.createChatConversationDeliveryQueue, oldService.chatChannelService, f.pending.id); + const after = await captureInboundQuery(db, f, "ready", undefined, undefined, f.pending.id); + const outcome = async (query: typeof before) => { + await db!.$client.unsafe("savepoint malformed_probe"); + try { return { rows: await runQuery(db!, query) }; } + catch (error) { await db!.$client.unsafe("rollback to savepoint malformed_probe"); return { error: (error as { code?: string }).code }; } + finally { await db!.$client.unsafe("release savepoint malformed_probe"); } + }; + const oldResult = await outcome(before), newResult = await outcome(after); + assert.deepEqual(newResult, oldResult, `${input.scope}: malformed ready selection outcome differs`); + malformedReadyControls.push({ scope: input.scope, before: oldResult, after: newResult }); + await db.$client.unsafe("delete from chat_actions where id = $1::uuid", [action.id]); + } + } + } finally { await db.$client.unsafe("rollback"); } + await db.$client.unsafe("delete from chat_endpoints where id = $1::uuid", [f.endpointId]); + } + const runtimeSources = sources.map(source => ({ path: source, + baselineSha256: createHash("sha256").update(execFileSync("git", ["show", `${baselineRevision}:${source}`])).digest("hex"), + treatmentDiffSha256: createHash("sha256").update(execFileSync("git", ["diff", "--no-ext-diff", baselineRevision, "--", source])).digest("hex"), + })); + console.log(JSON.stringify({ baselineRevision, treatmentRevision: execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8" }).trim(), runtimeSources, + version, settings, jitOffControl: process.argv.includes("--jit-off"), baselineExceedsWorkBudget, budgetFailures, + malformedReadyControls, warmups: 2, timedSamplesPerArm: 5, setupAndProofMs: performance.now() - started, results }, null, 2)); + if (process.argv.includes("--require-work-budget")) assert.deepEqual(budgetFailures, [], "Fixture work budget failed"); +} finally { + try { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); await fs.rm(dir, { recursive: true, force: true }); } + finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/helpers/inbound-discovery-fixture.ts b/server/src/__tests__/helpers/inbound-discovery-fixture.ts new file mode 100644 index 00000000000..c11441771a5 --- /dev/null +++ b/server/src/__tests__/helpers/inbound-discovery-fixture.ts @@ -0,0 +1,102 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { and, eq, getTableName, sql } from "drizzle-orm"; +import { agents, chatActions, chatDeliveries, chatEndpoints, companies, toolApplications, toolConnections, type Db } from "@paperclipai/db"; +import { createChatConversationDeliveryQueue } from "../../services/chat-channels/inbound/conversation-queue.js"; +import { chatChannelService } from "../../services/chat-channels.js"; + +export type CapturedQuery = { sql: string; params: unknown[] }; +export type QueueFactory = typeof createChatConversationDeliveryQueue; +export type ServiceFactory = typeof chatChannelService; +export const fixtureTime = new Date("2026-01-01T12:00:00.000Z"); + +export async function seedInboundDiscovery(db: Db, history = 20, payloadBytes = 0) { + const companyId = randomUUID(), agentId = randomUUID(), endpointId = randomUUID(); + const applicationId = randomUUID(), connectionId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Synthetic inbound discovery", issuePrefix: `Q${companyId.replaceAll("-", "").slice(0, 10).toUpperCase()}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "codex_local" }); + await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Chat fixture", type: "chat" }); + await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Chat", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }); + await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, assignedAgentId: agentId, provider: "slack", publicId: randomUUID(), status: "active" }); + const payload = randomBytes(Math.ceil(payloadBytes * 3 / 4)).toString("base64"); + await db.$client.unsafe(`insert into chat_deliveries (company_id, endpoint_id, provider_event_id, deduplication_key, event_kind, normalized_event, state, received_at) + select $1::uuid, $2::uuid, 'history-' || n, 'history-' || n, 'message', jsonb_build_object('conversation', jsonb_build_object('externalThreadId', 'slack:fixture'), 'message', jsonb_build_object('text', $3::text)), 'processed', $4::timestamptz + n * interval '1 millisecond' from generate_series(1, $5::int) n`, [companyId, endpointId, payload, fixtureTime.toISOString(), history]); + await db.$client.unsafe(`insert into chat_actions (company_id, endpoint_id, delivery_id, kind, provider_action_id, status, payload) + select company_id, endpoint_id, id, 'inbound_wakeup', 'history-' || provider_event_id, 'processed', '{}'::jsonb from chat_deliveries where company_id = $1::uuid`, [companyId]); + const threadId = "slack:fixture"; + async function delivery(overrides: Partial = {}) { + const id = randomUUID(); + const [row] = await db.insert(chatDeliveries).values({ id, companyId, endpointId, providerEventId: id, deduplicationKey: id, + eventKind: "message", state: "received", receivedAt: fixtureTime, nextAttemptAt: fixtureTime, + normalizedEvent: { conversation: { externalThreadId: threadId }, message: { providerSentAt: fixtureTime.toISOString(), providerMessageSequence: "2", text: payload } }, ...overrides }).returning(); + return row!; + } + async function action(deliveryId: string | null, overrides: Partial = {}) { + const [row] = await db.insert(chatActions).values({ companyId, endpointId, deliveryId, kind: "inbound_wakeup", status: "issued", providerActionId: randomUUID(), ...overrides }).returning(); + return row!; + } + async function anotherEndpoint() { + const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.id, connectionId)); + const [endpoint] = await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, endpointId)); + const nextConnection = randomUUID(), nextEndpoint = randomUUID(); + await db.insert(toolConnections).values({ ...connection!, id: nextConnection, uid: randomUUID() }); + await db.insert(chatEndpoints).values({ ...endpoint!, id: nextEndpoint, connectionId: nextConnection, publicId: randomUUID() }); + return nextEndpoint; + } + const pending = await delivery({ state: "processed" }); + await action(pending.id, { result: { retryAt: "2099-01-01T12:00:00.000Z" } }); + await delivery({ receivedAt: new Date(fixtureTime.getTime() + 1000), normalizedEvent: { conversation: { externalThreadId: threadId }, message: { providerSentAt: fixtureTime.toISOString(), providerMessageSequence: "3" } } }); + await db.$client.unsafe("ANALYZE chat_deliveries"); + await db.$client.unsafe("ANALYZE chat_actions"); + return { companyId, endpointId, threadId, delivery, action, anotherEndpoint, pending }; +} + +export function queueOptions() { + return { DELIVERY_DRAIN_LIMIT: 25, DELIVERY_LEASE_TTL_MS: 60_000, durableExternalThreadIdentity: (id: string) => id, + conversationDrainKey: (endpoint: string, thread: string) => `${endpoint}:${thread}`, liveInboundMessages: new Map(), + githubManualMessages: new WeakMap(), githubAutomaticMessages: new WeakMap(), githubIssueMessages: new WeakMap(), + } as Parameters[1]; +} + +export async function captureInboundQuery(db: Db, fixture: Awaited>, kind: "deadline" | "head" | "ready", + queueFactory: QueueFactory = createChatConversationDeliveryQueue, serviceFactory: ServiceFactory = chatChannelService, onlyDeliveryId?: string): Promise { + let captured: CapturedQuery | undefined; + const stop = new Error("inbound discovery query 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 = ((table: Parameters[0]) => { + const selection = from(table), execute = selection.execute.bind(selection); + selection.execute = async () => { + const candidate = selection.toSQL(); + if (getTableName(table as typeof chatDeliveries) === "chat_deliveries" && candidate.sql.includes("inbound_wakeup") && + (kind !== "ready" || candidate.params.includes("reaction_added"))) { + captured = candidate; throw stop; + } + if (kind === "ready") return []; + return execute(); + }; + return selection; + }) as typeof builder.from; + return builder; + }; + }}); + try { + if (kind === "ready") await serviceFactory(captureDb, { heartbeat: {} } as Parameters[1]).processPendingDeliveries(25, onlyDeliveryId, false); + else { + const queue = queueFactory(captureDb, queueOptions()); + if (kind === "deadline") await queue.nextInboundDeliveryAt(); + else await queue.drainConversationDeliveries(fixture.endpointId, fixture.threadId); + } + } catch (error) { if (error !== stop) throw error; } + if (!captured) throw new Error(`Did not capture real ${kind} query`); + return captured; +} + +export async function runQuery(db: Db, query: CapturedQuery) { + return db.$client.unsafe(query.sql, query.params as never[]).values(); +} + +export async function pendingIds(db: Db, predicate: ReturnType, companyId: string) { + return db.select({ id: chatDeliveries.id }).from(chatDeliveries).where(and(eq(chatDeliveries.companyId, companyId), predicate)).orderBy(chatDeliveries.id); +} diff --git a/server/src/__tests__/inbound-queue-discovery.test.ts b/server/src/__tests__/inbound-queue-discovery.test.ts new file mode 100644 index 00000000000..5bd8f84589b --- /dev/null +++ b/server/src/__tests__/inbound-queue-discovery.test.ts @@ -0,0 +1,96 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { and, eq, getTableColumns, sql } from "drizzle-orm"; +import { chatActions, chatDeliveries, chatEndpoints, createDb, startEmbeddedPostgresTestDatabase, type Db } from "@paperclipai/db"; +import { createChatConversationDeliveryQueue } from "../services/chat-channels/inbound/conversation-queue.js"; +import { pendingInboundWakeupCondition } from "../services/chat-channels/inbound/delivery-ledger.js"; +import { captureInboundQuery, pendingIds, queueOptions, runQuery, seedInboundDiscovery } from "./helpers/inbound-discovery-fixture.js"; + +let database: Awaited>; +let db: Db; +beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("inbound-discovery-test-"); db = createDb(database.connectionString); }); +afterAll(async () => { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); }); + +describe("inbound queue discovery", () => { + it("keeps scoped pending ownership without multiplying deliveries for duplicate or null actions", async () => { + const f = await seedInboundDiscovery(db); + const other = await seedInboundDiscovery(db, 0); + const unowned = await f.delivery({ state: "processed" }); + const duplicate = await f.delivery({ state: "processed" }); + await f.action(duplicate.id); await f.action(duplicate.id); + await f.action(null); + await f.action(unowned.id, { endpointId: await f.anotherEndpoint() }); + await f.action(unowned.id, { kind: "other" }); + await f.action(unowned.id, { status: "processed" }); + const oracle = sql`exists (select 1 from chat_actions a where a.delivery_id = chat_deliveries.id and a.endpoint_id = chat_deliveries.endpoint_id and a.company_id = chat_deliveries.company_id and a.kind = 'inbound_wakeup' and a.status in ('issued', 'processing'))`; + const booleans = await db.select({ pending: pendingInboundWakeupCondition(), oracle }).from(chatDeliveries).where(eq(chatDeliveries.companyId, f.companyId)); + expect(booleans.every(row => typeof row.pending === "boolean" && row.pending === row.oracle)).toBe(true); + expect(await pendingIds(db, pendingInboundWakeupCondition(), f.companyId)).toEqual(await pendingIds(db, oracle, f.companyId)); + expect((await pendingIds(db, pendingInboundWakeupCondition(), f.companyId)).map(row => row.id).sort()).toEqual([f.pending.id, duplicate.id].sort()); + }); + + it("preserves scoped ready-head probes with unrelated malformed retry data", async () => { + const f = await seedInboundDiscovery(db); + const other = await seedInboundDiscovery(db, 0); + const unrelated = await f.delivery({ state: "filtered" }); + const oracle = sql`exists (select 1 from chat_actions a where a.delivery_id = chat_deliveries.id and a.endpoint_id = chat_deliveries.endpoint_id and a.company_id = chat_deliveries.company_id and a.kind = 'inbound_wakeup' and a.status in ('issued', 'processing') and ((a.status = 'issued' and (a.result->>'retryAt' is null or (a.result->>'retryAt')::timestamptz <= now())) or (a.status = 'processing' and a.updated_at <= ${new Date(Date.now() - 60_000).toISOString()}::timestamptz)))`; + for (const input of [ + { companyId: f.companyId, endpointId: f.endpointId, deliveryId: unrelated.id }, + { companyId: other.companyId, endpointId: other.endpointId, deliveryId: other.pending.id }, + { companyId: f.companyId, endpointId: f.endpointId, deliveryId: null }, + ]) { + const [action] = await db.insert(chatActions).values({ ...input, kind: "inbound_wakeup", providerActionId: randomUUID(), status: "issued", result: { retryAt: "invalid timestamp" } }).returning(); + try { + expect(await db.select({ id: chatDeliveries.id }).from(chatDeliveries).where(and(eq(chatDeliveries.id, f.pending.id), oracle))).toEqual([]); + expect(await db.select({ id: chatDeliveries.id }).from(chatDeliveries).where(and(eq(chatDeliveries.id, f.pending.id), pendingInboundWakeupCondition(true)))).toEqual([]); + } finally { await db.delete(chatActions).where(eq(chatActions.id, action!.id)); } + } + }); + + it("keeps a future wake at the conversation head and reactions independent", async () => { + const f = await seedInboundDiscovery(db); + const queue = createChatConversationDeliveryQueue(db, queueOptions()); + await db.update(chatEndpoints).set({ status: "paused" }); + await db.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, f.endpointId)); + expect(await queue.nextInboundDeliveryAt()).toBe(Date.parse("2099-01-01T12:00:00.000Z")); + await f.delivery({ eventKind: "reaction_added", state: "retry", nextAttemptAt: new Date("2090-01-01T00:00:00Z") }); + expect(await queue.nextInboundDeliveryAt()).toBe(Date.parse("2090-01-01T00:00:00Z")); + await db.update(chatEndpoints).set({ status: "attention" }).where(eq(chatEndpoints.id, f.endpointId)); + expect(await queue.nextInboundDeliveryAt()).toBeNull(); + await f.delivery({ eventKind: "installation", state: "retry", nextAttemptAt: new Date("2095-01-01T00:00:00Z"), normalizedEvent: { + lifecycle: { provider: "slack", providerEventId: randomUUID(), kind: "endpoint", availability: "available", reason: "Recovered" } } }); + expect(await queue.nextInboundDeliveryAt()).toBe(Date.parse("2095-01-01T00:00:00Z")); + await db.update(chatEndpoints).set({ provider: "agentmail", publicationMode: "explicit", externalExecutionPolicy: "agent" }).where(eq(chatEndpoints.id, f.endpointId)); + expect(await queue.nextInboundDeliveryAt()).toBeNull(); + }); + + it("preserves ready wake retry and processing deadlines and complete real selection rows", async () => { + const f = await seedInboundDiscovery(db, 30, 4096); + const ready = await f.delivery({ state: "processed", normalizedEvent: { conversation: { externalThreadId: "slack:ready" }, message: { text: "complete payload", providerMessageSequence: "not-numeric", providerMessageId: "12", providerUpdateId: "4" } } }); + await f.action(ready.id, { result: { retryAt: "2020-01-01T00:00:00Z" } }); + const stale = await f.delivery({ state: "processed" }); + await f.action(stale.id, { status: "processing", updatedAt: new Date("2020-01-01T00:00:00Z") }); + const live = await f.delivery({ state: "processed" }); + await f.action(live.id, { status: "processing", updatedAt: new Date("2099-01-01T00:00:00Z") }); + const query = await captureInboundQuery(db, f, "ready"); + const rows = await runQuery(db, query); + const ids = rows.map(row => row[0]); + expect(ids).toContain(ready.id); expect(ids).toContain(stale.id); + expect(ids).not.toContain(live.id); expect(ids).not.toContain(f.pending.id); + const returned = rows.find(row => row[0] === ready.id)!; + const normalizedEventIndex = Object.keys(getTableColumns(chatDeliveries)).indexOf("normalizedEvent"); + expect(returned[normalizedEventIndex]).toEqual(ready.normalizedEvent); + }); + + it("keeps provider sequence and final receipt ties for the actual head query", async () => { + const f = await seedInboundDiscovery(db); + await db.update(chatActions).set({ status: "processed" }).where(eq(chatActions.companyId, f.companyId)); + await db.update(chatDeliveries).set({ state: "filtered" }).where(eq(chatDeliveries.companyId, f.companyId)); + const later = await f.delivery({ normalizedEvent: { conversation: { externalThreadId: f.threadId }, message: { providerSentAt: "2026-01-01T12:00:00Z", providerMessageSequence: "10" } } }); + const first = await f.delivery({ normalizedEvent: { conversation: { externalThreadId: f.threadId }, message: { providerSentAt: "2026-01-01T12:00:00Z", providerMessageSequence: "2" } } }); + const rows = await runQuery(db, await captureInboundQuery(db, f, "head")); + expect(rows).toHaveLength(1); expect(rows[0]?.[0]).toBe(first.id); expect(rows[0]?.[0]).not.toBe(later.id); + await db.update(chatDeliveries).set({ nextAttemptAt: new Date("2099-01-01T12:00:00Z"), state: "retry" }).where(eq(chatDeliveries.id, first.id)); + expect((await runQuery(db, await captureInboundQuery(db, f, "head")))[0]?.[0]).toBe(first.id); + }); +}); diff --git a/server/src/services/chat-channels/inbound/delivery-ledger.ts b/server/src/services/chat-channels/inbound/delivery-ledger.ts index 841378675ff..cb6938024f4 100644 --- a/server/src/services/chat-channels/inbound/delivery-ledger.ts +++ b/server/src/services/chat-channels/inbound/delivery-ledger.ts @@ -183,6 +183,15 @@ const DELIVERY_PROCESSING_STALE_MS = 60_000; } function pendingInboundWakeupCondition(readyOnly = false): SQL { + if (!readyOnly) { + return sql`(${chatDeliveries.companyId}, ${chatDeliveries.endpointId}, ${chatDeliveries.id}) in ( + select ${chatActions.companyId}, ${chatActions.endpointId}, ${chatActions.deliveryId} + from ${chatActions} + where ${chatActions.kind} = 'inbound_wakeup' + and ${chatActions.deliveryId} is not null + and ${chatActions.status} in ('issued', 'processing') + )`; + } return sql`exists ( select 1 from ${chatActions} where ${chatActions.deliveryId} = ${chatDeliveries.id}