diff --git a/doc/plans/2026-10-28-queued-comment-consumption-query.md b/doc/plans/2026-10-28-queued-comment-consumption-query.md new file mode 100644 index 00000000000..f942b8fdd13 --- /dev/null +++ b/doc/plans/2026-10-28-queued-comment-consumption-query.md @@ -0,0 +1,218 @@ +# Queued-comment consumption query proof + +The execution blocker counts saved user comments that no run has consumed. +It must not make consumed input eligible for replay. This change keeps the same +read and cancellation rules. It projects the two wake-comment fields before +PostgreSQL repeats the anti-join for each candidate comment. + +## Cause and scope + +A recent hosted sample had 99 calls of the eight-comment query. They returned +eight rows per call, used about 8,002 shared buffer hits per call, and averaged +131.14 ms of SQL execution. PostgreSQL reported no JIT work. These are SQL +statistics, not correlated request timings. + +The old local plan already scans the matching run relation once. It materializes +raw context values and then reads their large TOAST values again for each +comment. A 1,400-run stress fixture removed 11,200 join pairs and used about +166,645 shared buffer accesses. The hypothesis that there were eight underlying +run scans was rejected before changing production code. + +The replacement subquery selects `wakeCommentId` as text and `wakeCommentIds` as +JSON. `OFFSET 0` prevents PostgreSQL from pulling these expressions back into the +outer anti-join. The local plan materializes these small extracted values. It +still reads lazily and can stop at an early matching consumed run. There is no +new table, index, migration, cache, timestamp cutoff, or extra database round trip. + +The company condition, exact JSON `issueId` text condition, singular `->>` +comparison, array `@>` predicate, user author condition, deletion condition, and +cancelled-before-start exception are unchanged. Native and first-class run issue +IDs do not replace the JSON authority in this path. No JSON arrays are expanded +or cast to UUID. Non-array JSON therefore retains the original meaning. + +## Reproduction + +Use a disposable PostgreSQL database with current repository migrations. An +explicit external database must have a `queued_comments_` name. The script +clears its company graph before setup and after each fixture. Without the +external URL the benchmark starts a disposable embedded PostgreSQL database. + +```sh +QUEUED_COMMENT_BENCHMARK_DATABASE_URL="$DISPOSABLE_QUEUED_COMMENT_DATABASE_URL" \ + pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-queued-comment-consumption.ts \ + --baseline-ref=8279c990080b15abdcd9981f5793290fab37f9ad --require-budget +``` + +Repeat with `--jit-off` for the planner control. `--representative-only` selects +the forty-run case. `--wait-for-slot` waits once after initial setup so concurrent +local workers can stop before timing starts. + +The script loads the pinned original `execution-blocker.ts` from Git. It rewrites +relative imports to the same unchanged dependencies as the treatment. It rejects +other runtime dependency changes. The report records the source hashes, database +version and settings, every sample, SQL and parameters, full positional rows, +full execution blocker result, and execution plans. SQL results have no promised +order, so equality compares complete row multisets. Service objects compare +exactly. Capture, assertions, and `EXPLAIN ANALYZE` are outside the timed calls. +Each arm has two warmups and five alternating measured samples. + +## Fixture fidelity + +The hosted PostgreSQL version is 17.11. Anonymous aggregates over 1,198 task +groups gave 2, 11, and 40 runs at the median, 90th, and 99th percentiles, with a +maximum of 69. These groups use native or first-class bindings; the changed query +uses JSON issue bindings, so the counts are a sizing guide rather than exact +query selectivity. A bounded block sample of 262 stored contexts gave 27,075, +71,979, and 201,832 bytes at those percentiles. Stored sizes are not decompressed +logical JSON sizes. Synthetic random Base64 payloads avoid repeated-string +compression and are close to the selected stored-byte envelopes. + +The benchmark includes zero comments, empty history, two runs with 27 KiB +contexts, eleven runs with 72 KiB contexts, forty runs with 72 KiB contexts, +and a joint tail of sixty-nine runs with 200 KiB contexts. It separately checks +one queued comment, eight immediately consumed comments, half consumed comments, +and sixty-four queued comments. It also retains a 1,400-run stress case, which +exceeds observed task history and is not called representative. Each nonempty +history has 1,400 unrelated runs with approximately 27 KiB contexts. This checks +that the existing company and JSON issue index can limit the target task. + +The first-consumed fixture inserts its matching run before physical history +rows, with a newer creation time. Thus a sequential scan and the existing +creation-time index can both find it first. This prevents a misleading fast-path +control in which the matching run is actually last in physical storage. + +## Acceptance and selection timing + +The original eight-comment, 300-run regression fixture used 35,711 shared buffer +accesses. The pre-change guard failed against a budget of 10,600, which permits +32 accesses per run plus 1,000 fixed accesses. Five semantic controls passed at +that same revision. The preliminary single-scan assertion passed the baseline +and is not counted as a red result. + +After the initial projection trial, the final timing criteria were selected +before the quiet measurement matrices. The explicit local benchmark requires +at least a 1.5-fold SQL improvement for +the forty-run, eight-unconsumed-comment case. Zero/empty/small/one-comment and +immediately consumed service controls must add no more than 25 percent or 2 ms, +whichever permits more overhead. These wall-clock limits belong to the explicit +benchmark, not the uncontrolled CI test environment. The CI regression uses +query work and exact behavior instead. + +## Final local measurements + +The final quiet measurements used PostgreSQL 17.11 on ARM64 Linux, with 128 MiB +shared buffers and 4 MiB work memory. JIT was enabled at its default 100,000-cost +threshold for the primary matrix, but no measured query used JIT. The control +explicitly disabled JIT. Two warmups and five alternating samples were used for +each arm. The initial matrix had eleven cases. Review added a twelfth deep +first-consumed case and repeated the two existing consumed cases with a wide +consuming-run context. All twenty-four final fixture/JIT cells returned equal +complete positional SQL rows and equal full service objects. Every selected +budget passed. The initial thin consuming-row results remain in private raw +evidence and are superseded by the wide-row controls below. + +| Fixture | SQL before → after, ms | Full blocker before → after, ms | Shared buffers before → after | +| --- | ---: | ---: | ---: | +| No saved comments | no query | 17.24 → 17.46 | no query | +| Empty history | 2.39 → 4.23 | 26.07 → 29.97 | 1 → 1 | +| 2 runs, 27 KiB | 2.33 → 3.02 | 20.99 → 26.02 | 196 → 28 | +| 11 runs, 72 KiB | 8.35 → 5.05 | 37.60 → 35.13 | 2,116 → 268 | +| 40 runs, 72 KiB | 11.47 → 3.82 | 29.48 → 18.85 | 7,686 → 966 | +| 69 runs, 200 KiB | 43.35 → 8.41 | 68.43 → 31.60 | 31,623 → 3,959 | +| One saved comment | 4.31 → 4.63 | 29.09 → 29.00 | 965 → 965 | +| Eight immediately consumed, wide run | 1.60 → 1.28 | 13.90 → 16.59 | 196 → 28 | +| Half consumed, wide run | 3.13 → 1.75 | 14.44 → 11.49 | 4,038 → 990 | +| 64 saved comments | 80.88 → 6.98 | 111.08 → 39.90 | 61,446 → 966 | +| 1,400-run stress | 172.82 → 44.27 | 212.60 → 73.94 | 175,089 → 37,889 | +| 1,400 runs, first wide run consumes all | 1.54 → 1.99 | 19.00 → 18.23 | 122 → 24 | + +The forty-run SQL samples ranged from 10.36–12.33 ms before and 2.83–4.30 ms +after. Full blocker samples ranged from 27.34–33.22 ms before and 18.58–21.81 ms +after. This is a 3.00-fold SQL gain and a 36 percent service reduction. Its buffer +work is close to the hosted eight-comment shape, but hosted execution time is +much higher. These local times do not predict a 131 ms hosted saving. + +Small and empty-history service medians rose 5.03 ms and 3.90 ms with JIT enabled. +The accepted control budget is 25 percent or 2 ms, whichever permits more +overhead. Small and empty-history medians are inside that budget. This accepts +a possible small-path cost; it does not claim a universal improvement. The small +service ranges overlap: 18.49–36.48 ms before and 17.29–41.89 ms after. +These observations do not isolate the cause of the service difference. The +higher medians remain visible as accepted local costs. Hosted before/after +correlation still requires deployment acceptance. +The unchanged no-comment path runs no changed SQL; its 0.21 ms difference cannot +be attributed to the projection. One-comment results are neutral. The widened immediately consumed service +median rose 2.69 ms (19 percent), inside the selected control budget; its ranges +overlap at 13.34–27.72 ms before and 13.18–18.10 ms after. These controls remain part of the report, not omitted slower cases. + +With JIT disabled, the forty-run SQL median was 10.96 → 3.63 ms and the service +median was 32.18 → 24.75 ms. Small service medians were 15.58 → 14.68 ms; empty +history was 18.74 → 18.23 ms; the widened immediately consumed control was 13.59 → 14.64 ms. Plans had +no JIT work in either matrix, so this change addresses repeated JSON reads. + +The reviewer-requested 1,400-run first-consumed case reads exactly one underlying +run row in both arms. Each materialization returns one row and is revisited eight +times; neither arm eagerly reads the remaining history. Shared buffer accesses +fall from 122 to 24. Its JIT-on SQL ranges overlap at 0.86–2.79 ms before and +0.77–4.38 ms after, so this is an early-exit work proof, not a latency win. +With JIT disabled it is 1.08 → 1.07 ms of SQL and 11.36 → 11.53 ms of service +time, with the same buffer counts. The consuming row has the same 32 KiB random +payload as the deep history. The forty-run consuming controls use 72 KiB random +payloads on the consuming row too. This fixes the original thin-row fixture. +`--fixture=deep-consumed-first` selects this case; repeat `--fixture=` to select +more than one case. The six affected review-control cells passed in a coordinated +40-second quiet window. Production source did not change during this follow-up. + +The final runtime source SHA-256 is +`f750b3d12be73ed49b0a3be9ca9c769514eb561b98a17c1ff0bb909d9f880809`. +The original source SHA-256 is +`c0fa9a2c27af3d97d8775cfe0489e651efddb1c10402c0f177eef33d23961767`. + +## Validation and remaining limits + +The real PostgreSQL regression suite covers exact scalar, array, nested-array, +object, null and missing JSON behavior; noncanonical JSON issue IDs; conflicting +native IDs; cross-company rows; deleted and agent-authored comments; duplicate +array entries; cancelled-before-start and started-cancelled runs; and queued +runs that already carry a consumed comment. It checks complete execution blocker +objects and verifies that reads do not create runs or replay input. + +Local improvements establish this SQL mechanism and its service effect on +synthetic data. They do not establish an equal change in hosted request latency. +A successful deployment still needs a fresh equivalent SQL window and correlated +request timings. PostgreSQL can choose a different plan as data or versions +change; the plan-work regression protects the repeated-context mechanism on the +provided fixture. + +## Feedback-loop record + +The first attempted scan-count regression passed the original code. Profiling +showed that repeated TOAST reads, not repeated relation scans, were the relevant +work. The revised buffer-work guard failed before the source change: 35,711 +accesses against a 10,600 budget. The projection then passed the same guard and +all semantic cases. + +Initial direct test invocations spent 44–70 seconds, mostly rebuilding unchanged +module transforms. A private Vitest configuration enabled the existing persistent +transform cache. The assembled route and regression suites then completed 91 +cases in 128.21 seconds, with transforms accounting for 18 percent. That larger +suite checks integration rather than comparing an identical timing loop. No +repository test configuration was changed to get this result. + +Scratch setup first failed on a TypeScript module extension and then on a URL +string replacement that changed the username. Both setup errors were corrected +before the real red result. These are not performance regressions. Native +PostgreSQL startup was unsupported on this host, so its default focused run +skipped six cases. Docker PostgreSQL 17.11 then ran all 91 targeted and assembled +cases through the repository database allocation, migration and client path. +Only the native server process constructor was replaced in private test setup. +These local runtime limitations remain distinct from the Linux CI result. + +The required full native test command was attempted. Its unchanged execution +suite completed 940 cases with 16 PostgreSQL startup failures and 924 passes. +The command was stopped at that infrastructure boundary, so the full local run +is incomplete and is not counted as a pass. An independent empty native initdb +failed with `semget: No space left on device`, which identifies the macOS +semaphore limit. Exactly those 16 cases then passed under Docker PostgreSQL +17.11 in 181.22 seconds. The full workspace typecheck and build passed. Complete +Linux CI remains a separate gate on the final PR head. diff --git a/scripts/benchmark-queued-comment-consumption.ts b/scripts/benchmark-queued-comment-consumption.ts new file mode 100644 index 00000000000..fdc3412569e --- /dev/null +++ b/scripts/benchmark-queued-comment-consumption.ts @@ -0,0 +1,124 @@ +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 { getExecutionBlocker } from "../server/src/services/execution-blocker.js"; +import { captureQueuedCommentQuery, planNodes, seedQueuedCommentConsumption, type BlockerFactory, type CapturedCommentQuery } from "../server/src/__tests__/helpers/queued-comment-consumption-fixture.js"; + +const baselineRef = process.argv.find(value => value.startsWith("--baseline-ref="))?.slice(15); +assert.ok(baselineRef, "Pass an explicit pre-change --baseline-ref="); +const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8" }).trim(); +const root = fileURLToPath(new URL("../", import.meta.url)); +const sourcePath = "server/src/services/execution-blocker.ts"; +const changedRuntime = 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(changedRuntime.filter(source => source !== sourcePath), [], "The baseline must not import changed runtime helpers"); +const baselineSource = execFileSync("git", ["show", `${baselineRevision}:${sourcePath}`], { encoding: "utf8" }); +const baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.queued-comment-baseline-")); +const rewritten = baselineSource.replace(/from "(\.{1,2}\/[^\"]+)"/g, (_match, specifier: string) => `from ${JSON.stringify(path.resolve(root, "server/src/services", specifier))}`); +await fs.writeFile(path.join(baselineDir, "execution-blocker.ts"), rewritten); +const old = await import(pathToFileURL(path.join(baselineDir, "execution-blocker.ts")).href) as { getExecutionBlocker: BlockerFactory }; +const targetUrl = process.env.QUEUED_COMMENT_BENCHMARK_DATABASE_URL; +if (targetUrl) assert.ok(new URL(targetUrl).pathname.startsWith("/queued_comments_"), "Use a disposable queued_comments_ database"); +let database: Awaited> | undefined; +let db: Db | undefined; +const keepAlive = setInterval(() => {}, 1000); +const started = performance.now(); +const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!; +const multiset = (rows: unknown[]) => rows.map(row => JSON.stringify(row)).sort(); +const fixtureDefinitions = [ + { name: "no-comments", comments: 0, history: 0, payloadBytes: 0, matched: 0 }, + { name: "empty-history", comments: 8, history: 0, payloadBytes: 0, matched: 0 }, + { name: "small", comments: 8, history: 2, payloadBytes: 27 * 1024, matched: 0 }, + { name: "moderate-eleven", comments: 8, history: 11, payloadBytes: 72 * 1024, matched: 0 }, + { name: "representative-eight", comments: 8, history: 40, payloadBytes: 72 * 1024, matched: 0 }, + { name: "wide-tail", comments: 8, history: 69, payloadBytes: 200 * 1024, matched: 0 }, + { name: "one-unconsumed", comments: 1, history: 40, payloadBytes: 72 * 1024, matched: 0 }, + { name: "eight-consumed-first", comments: 8, history: 40, payloadBytes: 72 * 1024, matched: 8 }, + { name: "eight-mixed", comments: 8, history: 40, payloadBytes: 72 * 1024, matched: 4 }, + { name: "large-queue", comments: 64, history: 40, payloadBytes: 72 * 1024, matched: 0 }, + { name: "deep-stress", comments: 8, history: 1400, payloadBytes: 32 * 1024, matched: 0 }, + { name: "deep-consumed-first", comments: 8, history: 1400, payloadBytes: 32 * 1024, matched: 8 }, +]; +try { + if (!targetUrl) database = await startEmbeddedPostgresTestDatabase("queued-comment-benchmark-"); + db = createDb(targetUrl ?? database!.connectionString, { maxConnections: 1, idleTimeoutSeconds: 0 }); + await db.$client.unsafe("set client_min_messages = warning"); + if (process.argv.includes("--jit-off")) await db.$client.unsafe("set jit = off"); + 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', 'jit_optimize_above_cost', 'random_page_cost') order by name"); + await db.$client.unsafe("TRUNCATE companies CASCADE"); + const results = []; + const budgetFailures: string[] = []; + let waitedForSlot = false; + const fixtureSelectors = process.argv.filter(value => value.startsWith("--fixture=")).map(value => value.slice(10)); + const selectedFixtures = fixtureDefinitions.filter(value => (!fixtureSelectors.length || fixtureSelectors.includes(value.name)) && + (!process.argv.includes("--representative-only") || value.name === "representative-eight")); + assert.ok(selectedFixtures.length, "No benchmark fixtures match the requested selectors"); + for (const fixture of selectedFixtures) { + const f = await seedQueuedCommentConsumption(db, { ...fixture, unrelatedHistory: fixture.history ? 1400 : 0 }); + const before = await captureQueuedCommentQuery(db, f, old.getExecutionBlocker), after = await captureQueuedCommentQuery(db, f); + assert.deepEqual(after.result, before.result, `${fixture.name}: complete execution blocker differs`); + assert.equal(before.result?.savedMessageCount, fixture.comments - fixture.matched); + const queryRows = async (query: CapturedCommentQuery | undefined) => query ? db!.$client.unsafe(query.sql, query.params as never[]).values() : []; + const expected = await queryRows(before.query), actual = await queryRows(after.query); + assert.deepEqual(multiset(actual), multiset(expected), `${fixture.name}: complete positional SQL output differs`); + if (process.argv.includes("--wait-for-slot") && !waitedForSlot) { + waitedForSlot = true; + process.stderr.write(`READY ${fixture.name}: seeded and captured real service queries; press Enter to time\n`); + await new Promise(resolve => { process.stdin.once("data", () => { process.stdin.pause(); resolve(); }); process.stdin.resume(); }); + } + const samples = { sqlBefore: [] as number[], sqlAfter: [] as number[], serviceBefore: [] as number[], serviceAfter: [] as number[] }; + for (let iteration = 0; iteration < 7; iteration++) { + for (const arm of iteration % 2 ? ["after", "before"] as const : ["before", "after"] as const) { + const sqlStart = performance.now(); + const rows = await queryRows(arm === "before" ? before.query : after.query); + const sqlMs = performance.now() - sqlStart; + assert.deepEqual(multiset(rows), multiset(expected)); + const serviceStart = performance.now(); + const result = await (arm === "before" ? old.getExecutionBlocker : getExecutionBlocker)(db, f.companyId, f.issueId); + const serviceMs = performance.now() - serviceStart; + assert.deepEqual(result, before.result); + if (iteration >= 2) { + samples[arm === "before" ? "sqlBefore" : "sqlAfter"].push(sqlMs); + samples[arm === "before" ? "serviceBefore" : "serviceAfter"].push(serviceMs); + } + } + } + const plans: Record = {}; + const runScanLoops: Record = {}; + for (const [arm, captured] of [["before", before], ["after", after]] as const) { + if (!captured.query) continue; + const [row] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${captured.query.sql}`, captured.query.params as never[]); + plans[arm] = row["QUERY PLAN"]; + runScanLoops[arm] = planNodes(row["QUERY PLAN"][0].Plan).filter(node => node["Relation Name"] === "heartbeat_runs") + .reduce((total, node) => total + Number(node["Actual Loops"]), 0); + } + const medians = Object.fromEntries(Object.entries(samples).map(([arm, values]) => [arm, median(values)])); + if (fixture.name === "representative-eight" && medians.sqlBefore! < medians.sqlAfter! * 1.5) budgetFailures.push(`${fixture.name}: SQL gain below 1.5x`); + if (["no-comments", "empty-history", "small", "one-unconsumed", "eight-consumed-first", "deep-consumed-first"].includes(fixture.name) && + medians.serviceAfter! > Math.max(medians.serviceBefore! * 1.25, medians.serviceBefore! + 2)) budgetFailures.push(`${fixture.name}: service overhead exceeds 25% or 2 ms`); + if (fixture.name === "deep-consumed-first") { + const buffers = (arm: string) => { + const plan = (plans[arm] as { Plan: Record }[])[0]!.Plan; + return Number(plan["Shared Hit Blocks"]) + Number(plan["Shared Read Blocks"]); + }; + if (buffers("after") > buffers("before") * 1.25 + 32) budgetFailures.push(`${fixture.name}: lost early-exit buffer work`); + } + const equality = { completePositionalSqlRows: expected.length, completeServiceResult: before.result }; + results.push({ fixture, before: before.query, after: after.query, equality, samples, medians, runScanLoops, plans }); + process.stderr.write(`${fixture.name}: SQL ${medians.sqlBefore!.toFixed(2)} -> ${medians.sqlAfter!.toFixed(2)} ms; service ${medians.serviceBefore!.toFixed(2)} -> ${medians.serviceAfter!.toFixed(2)} ms; history scan loops ${JSON.stringify(runScanLoops)}\n`); + await db.$client.unsafe("TRUNCATE companies CASCADE"); + } + const treatmentSource = await fs.readFile(path.join(root, sourcePath)); + console.log(JSON.stringify({ baselineRevision, treatmentRevision: execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8" }).trim(), + sourcePath, baselineSha256: createHash("sha256").update(baselineSource).digest("hex"), treatmentSha256: createHash("sha256").update(treatmentSource).digest("hex"), + baselineDependenciesHeldConstant: true, version, settings, jitOff: process.argv.includes("--jit-off"), warmups: 2, timedSamplesPerArm: 5, + budgetFailures, setupAndProofMs: performance.now() - started, results }, null, 2)); + if (process.argv.includes("--require-budget")) assert.deepEqual(budgetFailures, [], "Local performance acceptance failed"); +} finally { + try { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); await fs.rm(baselineDir, { recursive: true, force: true }); } + finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/helpers/queued-comment-consumption-fixture.ts b/server/src/__tests__/helpers/queued-comment-consumption-fixture.ts new file mode 100644 index 00000000000..61928d5f809 --- /dev/null +++ b/server/src/__tests__/helpers/queued-comment-consumption-fixture.ts @@ -0,0 +1,70 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { getTableName } from "drizzle-orm"; +import { agents, agentWakeupRequests, companies, heartbeatRuns, issueComments, issueRecoveryActions, issues, type Db } from "@paperclipai/db"; +import { getExecutionBlocker } from "../../services/execution-blocker.js"; + +export type BlockerFactory = typeof getExecutionBlocker; +export type CapturedCommentQuery = { sql: string; params: unknown[] }; +export const fixtureTime = new Date("2026-10-01T12:00:00.000Z"); + +export async function seedQueuedCommentConsumption(db: Db, options: { comments?: number; history?: number; payloadBytes?: number; consumed?: boolean; matched?: number; unrelatedHistory?: number } = {}) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), otherIssueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Synthetic queued comments", issuePrefix: `Q${companyId.replaceAll("-", "").slice(0, 10).toUpperCase()}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "codex_local" }); + await db.insert(issues).values([{ id: issueId, companyId, title: "Held task", status: "blocked", assigneeAgentId: agentId }, + { id: otherIssueId, companyId, title: "Unrelated task" }]); + const commentIds = Array.from({ length: options.comments ?? 8 }, () => randomUUID()); + if (commentIds.length) await db.insert(issueComments).values(commentIds.map(id => ({ id, companyId, issueId, authorType: "user" as const, authorUserId: "fixture-user", body: "Queued input", createdAt: fixtureTime, updatedAt: fixtureTime }))); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "issue_comment", status: "deferred_issue_execution", + payload: { issueId, _paperclipWakeContext: { wakeCommentIds: commentIds } } }).returning(); + const [action] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog", status: "active", + ownerType: "board", cause: "legacy_execution_requires_reconciliation", fingerprint: randomUUID(), evidence: {}, nextAction: "Inspect saved work.", createdAt: fixtureTime, updatedAt: fixtureTime }).returning(); + const payload = randomBytes(Math.ceil((options.payloadBytes ?? 0) * 3 / 4)).toString("base64"); + if (options.matched) await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", startedAt: fixtureTime, + contextSnapshot: { issueId, wakeCommentIds: commentIds.slice(0, options.matched), prompt: payload } }); + if (options.history) await db.$client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, started_at, context_snapshot, created_at) + select $1::uuid, $2::uuid, 'succeeded', $3::timestamptz, + jsonb_build_object('issueId', $4::text, 'wakeCommentId', case when $7::boolean then $8::text else gen_random_uuid()::text end, + 'wakeCommentIds', jsonb_build_array(gen_random_uuid()::text), 'prompt', $5::text), $3::timestamptz + n * interval '1 second' + from generate_series(1, $6::int) n`, [companyId, agentId, fixtureTime.toISOString(), issueId, payload, options.history, options.consumed ?? false, commentIds[0] ?? randomUUID()]); + const unrelatedPayload = randomBytes(20 * 1024).toString("base64"); + if (options.unrelatedHistory) await db.$client.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, started_at, context_snapshot, created_at) + select $1::uuid, $2::uuid, 'succeeded', $3::timestamptz, jsonb_build_object('issueId', $4::text, 'prompt', $5::text), + $3::timestamptz + n * interval '1 second' from generate_series(1, $6::int) n`, + [companyId, agentId, fixtureTime.toISOString(), otherIssueId, unrelatedPayload, options.unrelatedHistory]); + await db.$client.unsafe("ANALYZE heartbeat_runs"); + await db.$client.unsafe("ANALYZE issue_comments"); + async function run(contextSnapshot: unknown, overrides: Partial = {}) { + const [row] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", startedAt: fixtureTime, + contextSnapshot: contextSnapshot as Record, ...overrides }).returning({ id: heartbeatRuns.id }); + return row!; + } + return { companyId, agentId, issueId, otherIssueId, commentIds, wake: wake!, action: action!, run }; +} + +export async function captureQueuedCommentQuery(db: Db, fixture: Awaited>, factory: BlockerFactory = getExecutionBlocker) { + let query: CapturedCommentQuery | undefined; + const observed = 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 issueComments) === "issue_comments" && candidate.sql.includes("consumed")) query = candidate; + return execute(); + }; + return selection; + }) as typeof builder.from; + return builder; + }; + }}); + const result = await factory(observed, fixture.companyId, fixture.issueId); + if (fixture.commentIds.length && !query) throw new Error("The real execution blocker did not reach its saved-comment query"); + return { query, result }; +} + +export function planNodes(plan: Record): Record[] { + return [plan, ...((plan.Plans ?? []) as Record[]).flatMap(planNodes)]; +} diff --git a/server/src/__tests__/queued-comment-consumption.test.ts b/server/src/__tests__/queued-comment-consumption.test.ts new file mode 100644 index 00000000000..53009b2b0bc --- /dev/null +++ b/server/src/__tests__/queued-comment-consumption.test.ts @@ -0,0 +1,85 @@ +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { and, eq, sql } from "drizzle-orm"; +import { agentWakeupRequests, createDb, heartbeatRuns, issueComments, issueRecoveryActions, type Db } from "@paperclipai/db"; +import { getExecutionBlocker } from "../services/execution-blocker.js"; +import { captureQueuedCommentQuery, fixtureTime, planNodes, seedQueuedCommentConsumption } from "./helpers/queued-comment-consumption-fixture.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const external = process.env.QUEUED_COMMENT_TEST_DATABASE_URL; +const describePostgres = external || support.supported ? describe : describe.skip; +if (!external && !support.supported) console.warn(`Skipping queued-comment consumption tests: ${support.reason ?? "unsupported PostgreSQL runtime"}`); + +describePostgres("queued-comment consumption in the execution blocker", () => { + let db: Db; + let database: Awaited> | undefined; + beforeAll(async () => { + if (external && !new URL(external).pathname.startsWith("/queued_comments_")) throw new Error("Use a disposable queued_comments_ database"); + if (!external) database = await startEmbeddedPostgresTestDatabase("queued-comment-consumption-"); + db = createDb(external ?? database!.connectionString, { maxConnections: 1 }); + await db.$client.unsafe("set client_min_messages = warning"); + }, 30_000); + afterEach(async () => { await db?.execute(sql`TRUNCATE companies CASCADE`); }); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + it("bounds repeated run-context reads for eight unconsumed saved comments", async () => { + const f = await seedQueuedCommentConsumption(db, { comments: 8, history: 300, payloadBytes: 32 * 1024 }); + const captured = await captureQueuedCommentQuery(db, f); + expect(captured.result?.savedMessageCount).toBe(8); + const [explain] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${captured.query!.sql}`, captured.query!.params as never[]); + const scans = planNodes(explain["QUERY PLAN"][0].Plan).filter(node => node["Relation Name"] === "heartbeat_runs"); + expect(scans.length).toBeGreaterThan(0); + expect(scans.reduce((total, node) => total + Number(node["Actual Loops"]), 0)).toBeLessThanOrEqual(1); + const plan = explain["QUERY PLAN"][0].Plan; + expect(Number(plan["Shared Hit Blocks"]) + Number(plan["Shared Read Blocks"])).toBeLessThanOrEqual(300 * 32 + 1000); + }); + + it("preserves JSON containment, text IDs, scope, author and cancellation rules", async () => { + const f = await seedQueuedCommentConsumption(db, { comments: 16 }); + const [singular, array, nested, scalar, object, upper, cancelled, startedCancelled, wrongTask, nativeOnly, otherCompany, deleted, agent, pending] = f.commentIds; + await f.run({ issueId: f.issueId, wakeCommentId: singular }); + await f.run({ issueId: f.issueId, wakeCommentIds: [array, array, 1, {}, null] }); + await f.run({ issueId: f.issueId, wakeCommentIds: [[nested]] }); + await f.run({ issueId: f.issueId, wakeCommentIds: scalar }); + await f.run({ issueId: f.issueId, wakeCommentIds: { [object!]: true } }); + await f.run({ issueId: f.issueId.toUpperCase(), wakeCommentId: upper }); + await f.run({ issueId: f.issueId, wakeCommentId: cancelled }, { status: "cancelled", startedAt: null }); + await f.run({ issueId: f.issueId, wakeCommentId: startedCancelled }, { status: "cancelled" }); + await f.run({ issueId: f.otherIssueId, wakeCommentId: wrongTask }, { nativeIssueId: f.issueId }); + await f.run({ wakeCommentId: nativeOnly }, { issueId: f.issueId, nativeIssueId: f.issueId }); + const other = await seedQueuedCommentConsumption(db, { comments: 0 }); + await other.run({ issueId: f.issueId, wakeCommentId: otherCompany }); + await f.run({ issueId: f.issueId, wakeCommentIds: [pending] }, { status: "queued", startedAt: null }); + await f.run(null); + await f.run([]); + await f.run("text"); + await db.update(issueComments).set({ deletedAt: fixtureTime }).where(eq(issueComments.id, deleted!)); + await db.update(issueComments).set({ authorType: "agent", authorAgentId: f.agentId }).where(eq(issueComments.id, agent!)); + const captured = await captureQueuedCommentQuery(db, f); + const expected = await db.$client.unsafe(`select id from issue_comments where company_id = $1::uuid and issue_id = $2::uuid + and id = any($3::uuid[]) and author_type = 'user' and deleted_at is null + and not exists (select 1 from heartbeat_runs consumed where consumed.company_id = $1::uuid + and consumed.context_snapshot->>'issueId' = $2::text and not (consumed.status = 'cancelled' and consumed.started_at is null) + and (consumed.context_snapshot->>'wakeCommentId' = issue_comments.id::text + or consumed.context_snapshot->'wakeCommentIds' @> jsonb_build_array(issue_comments.id::text)))`, [f.companyId, f.issueId, f.commentIds]); + const actual = await db.$client.unsafe(captured.query!.sql, captured.query!.params as never[]).values(); + expect(actual.map(row => row[0]).sort()).toEqual(expected.map(row => row.id).sort()); + expect(captured.result?.savedMessageCount).toBe(expected.length); + expect(expected.map(row => row.id)).toEqual(expect.arrayContaining([nested, scalar, object, upper, cancelled, wrongTask, nativeOnly, otherCompany])); + expect(expected.map(row => row.id)).not.toEqual(expect.arrayContaining([singular])); + expect(expected).toHaveLength(10); + }); + + it.each([0, 1, 8, 64])("counts %s queued comments without a run or a replay", async count => { + const f = await seedQueuedCommentConsumption(db, { comments: count }); + const before = await db.select().from(heartbeatRuns); + const result = await getExecutionBlocker(db, f.companyId, f.issueId); + expect(result).toEqual({ recoveryActionId: f.action.id, runId: null, agentId: null, cause: f.action.cause, + nextAction: f.action.nextAction, runStatus: null, runError: null, canContinue: false, canRetry: false, savedMessageCount: count }); + expect(await db.select().from(heartbeatRuns)).toEqual(before); + await db.update(agentWakeupRequests).set({ status: "consumed" }).where(eq(agentWakeupRequests.id, f.wake.id)); + expect((await getExecutionBlocker(db, f.companyId, f.issueId))?.savedMessageCount).toBe(0); + await db.update(issueRecoveryActions).set({ status: "resolved" }).where(and(eq(issueRecoveryActions.companyId, f.companyId), eq(issueRecoveryActions.id, f.action.id))); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + }); +}); diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 61d6ded503b..c22437f1a47 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -187,11 +187,15 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st const saved = savedIds.length ? await db.select({ id: issueComments.id }).from(issueComments).where(and( eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), inArray(issueComments.id, savedIds), eq(issueComments.authorType, "user"), sql`${issueComments.deletedAt} is null`, - sql`not exists (select 1 from ${heartbeatRuns} consumed where consumed.company_id = ${companyId} - and consumed.context_snapshot->>'issueId' = ${issueId} - and not (consumed.status = 'cancelled' and consumed.started_at is null) - and (consumed.context_snapshot->>'wakeCommentId' = ${issueComments.id}::text - or consumed.context_snapshot->'wakeCommentIds' @> jsonb_build_array(${issueComments.id}::text)))`, + sql`not exists (select 1 from ( + select consumed.context_snapshot->>'wakeCommentId' as comment_id, + consumed.context_snapshot->'wakeCommentIds' as comment_ids + from ${heartbeatRuns} consumed where consumed.company_id = ${companyId} + and consumed.context_snapshot->>'issueId' = ${issueId} + and not (consumed.status = 'cancelled' and consumed.started_at is null) + offset 0 + ) consumed where consumed.comment_id = ${issueComments.id}::text + or consumed.comment_ids @> jsonb_build_array(${issueComments.id}::text))`, )) : []; const cancellation = run ? readRunCancellation(run.resultJson) : null; const eligibleContinuation = Boolean(!restoreHold && run && conversation?.assigneeAgentId === run.agentId &&