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}