perf: avoid repeated run-context reads for queued comments (#18021)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Execution recovery shows how many saved user messages remain
unconsumed.
> - This read must preserve the rule that consumed input cannot be
replayed.
> - The current query reads a large run context again for each saved
comment.
> - PostgreSQL already scans the run relation once, but repeats the JSON
reads.
> - This pull request projects the two wake-comment fields before the
repeated comparison.
> - The query uses less database work and returns the same blocker
state.

## Linked Issues or Issue Description

**What happened?**

The eight-comment consumption query averaged 131.14 ms across 99 calls
in a recent hosted sample. Each call returned all eight comments. The
query used about 8,002 shared buffer hits per call. It used no JIT
compilation. The local plan showed repeated TOAST reads in the
anti-join, despite a single run scan.

**Expected behavior**

Count saved, undeleted user comments without reading each large context
again for each comment. Keep exact JSON issue bindings, JSON
containment, company scope, and cancellation rules. Consumed input must
remain excluded.

**Steps to reproduce**

Run the committed benchmark against disposable PostgreSQL with
repository migrations. Pass the pinned baseline below. Seed forty task
runs with 72 KiB random contexts, eight unconsumed comments, and 1,400
unrelated runs. Compare complete SQL rows and the full execution blocker
result.

**Paperclip version or commit**

Pre-fix baseline: `8279c990080b15abdcd9981f5793290fab37f9ad`. This
branch is based on current master. The intervening master changes do not
change the measured service or its SQL helpers.

**Deployment mode**

The report came from hosted PostgreSQL 17.11. The local proof uses
disposable PostgreSQL 17.11 on ARM64 Linux.

Related correctness work: #16528, #16664, and #18012. This query change
does not replace those fixes or resolve the replay report in #17782. The
duplicate search found no open performance fix for this query.

## What Changed

- Select only `wakeCommentId` as text and `wakeCommentIds` as JSON
before the anti-join.
- Use `OFFSET 0` as a PostgreSQL projection boundary. Keep lazy
first-match behavior and the existing company/JSON-issue index.
- Preserve the original singular text comparison and JSON containment
expression. Keep cancelled-before-start, deleted-comment, user-author,
and company filters unchanged.
- Add real PostgreSQL regression tests and a reproducible paired
benchmark. Compare every positional SQL column and the entire service
result.
- Record the rejected scan-count hypothesis, slower controls, full
samples, and acceptance limits in the [proof
document](doc/plans/2026-10-28-queued-comment-consumption-query.md).

## Verification

- Red before the source change: the eight-comment/300-run fixture used
35,711 shared buffer accesses and failed the selected 10,600-work
budget. Five semantic cases passed. The earlier single-scan guard passed
the original code and is not counted as red.
- Green: six new real PostgreSQL cases and 85 existing queued-comment
route cases pass together: **91/91**. Coverage includes
cancelled-before-start, started cancellation, no replay, conflicting
native IDs, noncanonical JSON IDs, cross-company data, deleted/agent
comments, and scalar/object/nested-array JSON.
- `pnpm -r typecheck` and `pnpm build`: passed. `pnpm test:run` was
attempted and stopped after 16 native PostgreSQL startup failures in an
unchanged executor suite; 924 other cases in that file passed. The full
local attempt is incomplete and is not counted as a pass. Those same 16
cases then passed with Docker PostgreSQL 17.11. Full Linux CI passed on
the final head.
- A standalone empty native `initdb` fails with `semget: No space left
on device` from the macOS semaphore limit. Native PostgreSQL startup is
unsupported on this host. Its focused attempt skipped six tests. The
91-case run used a private Docker PostgreSQL constructor substitute.
Repository database creation, migrations, clients, and cleanup were
retained. No test infrastructure change is committed.
- Final head `d2921d9220e22624f85f7057a577a6e2e98dc097`: all 58 check
contexts are terminal and clean (55 success, two expected skips, one
neutral). The full [PR
workflow](https://github.com/paperclipai/paperclip/actions/runs/40699607944)
passed. Greptile reviewed this exact head at 5/5. Paperclip Review
originally requested the deep first-consumed control; it now passes and
that thread is resolved. Its latest automated review attempt stopped
before completion and reported neutral, so no fresh Paperclip 5/5 is
claimed.

Two warmups and five alternating samples per arm, with capture, equality
checks, and plans outside the timers:

| Local fixture | SQL median before → after | Full blocker median before
→ after | Shared buffer accesses |
| --- | ---: | ---: | ---: |
| Representative: 40 runs, 72 KiB, 8 unconsumed comments | **11.47 →
3.82 ms (3.00×)** | **29.48 → 18.85 ms** | **7,686 → 966** |
| Same fixture, JIT disabled | 10.96 → 3.63 ms | 32.18 → 24.75 ms |
7,686 → 966 |
| 11 runs, 72 KiB | 8.35 → 5.05 ms | 37.60 → 35.13 ms | 2,116 → 268 |
| 8 comments consumed by the first wide run | 1.60 → 1.28 ms | 13.90 →
16.59 ms | 196 → 28 |
| 1,400-run stress, above observed task depth | 172.82 → 44.27 ms |
212.60 → 73.94 ms | 175,089 → 37,889 |
| 1,400 runs, first wide run consumes all 8 | 1.54 → 1.99 ms | 19.00 →
18.23 ms | 122 → 24 |

Representative SQL ranges: 10.36–12.33 ms before and 2.83–4.30 ms after.
Service ranges: 27.34–33.22 ms before and 18.58–21.81 ms after. All 24
final fixture/JIT cells return identical complete rows and service
objects. The follow-up widens the consuming run to the same payload size
as history and supersedes the initial thin-row consuming controls. All
final-matrix benchmark budgets pass. The deterministic buffer-work
budget was selected before the production change. The timing criteria
were selected after the initial exploratory projection trial and before
the final quiet matrices.

Small and empty-history service medians rose by 5.03 ms and 3.90 ms in
the JIT-on matrix. Small-case ranges overlap: 18.49–36.48 ms before and
17.29–41.89 ms after. These differences do not establish a causal
service regression. The JIT-off small and empty-history controls
improved slightly. The unchanged no-comment path runs no changed SQL.
The one-comment control is neutral. The widened immediately consumed
service median rose 2.69 ms (19 percent), inside the selected 25 percent
or 2 ms service budget. Its ranges overlap: 13.34–27.72 ms before and
13.18–18.10 ms after. This accepts a possible bounded small-path cost;
it does not claim every path gets faster. All controls and full
distributions remain in the proof.

The reviewer-requested deep first-consumed plan reads exactly **one**
underlying run row in each arm. Both materializations return one row,
revisited eight times. The remaining history is not read eagerly. JIT-on
SQL ranges overlap at 0.86–2.79 ms before and 0.77–4.38 ms after; this
proves early exit rather than a latency gain. With JIT disabled, SQL is
1.08 → 1.07 ms and service time is 11.36 → 11.53 ms, with the same 122 →
24 buffer accesses. Production source is unchanged by this proof update.

Reproduce:

```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
```

Use an isolated `queued_comments_` database. Repeat with `--jit-off`.
Without the URL, the script creates disposable embedded PostgreSQL. The
benchmark rejects changed runtime helpers and records source hashes,
settings, SQL/parameters, complete results, samples, and plans.

## Risks

- PostgreSQL planner changes can alter the materialization strategy. The
regression guard checks repeated context work on the supplied fixture.
- Small local service medians are noisy. Their ranges and slower medians
remain visible. A faster local query does not establish the same hosted
request saving.
- JSON authority and no-replay rules are sensitive. The query keeps the
original predicates and has real database tests for non-array and
conflicting-ID cases.
- There is no migration, index, cache, history cutoff, extra round trip,
deployment, or hosted write.

## Model Used

OpenAI Codex, GPT-6 family, with reasoning and tool use. The exact model
ID and context window are not exposed by this session. They are not
inferred from a model alias.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

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=<commit>");
+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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<void>(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<string, unknown> = {};
+    const runScanLoops: Record<string, number> = {};
+    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<string, number> }[])[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); }
+}