From ab80fea62f74f76a9d36a6a4eddd13d7842eafb1 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:12:02 -0600 Subject: [PATCH] perf(recovery): probe run bindings per comment in the lost-wake check (#20412) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The ownership reconciler finds user comments whose wake a restart lost, and it saves a wake for them (plan invariant 4) > - Its lost-wake check matched a task's runs with one OR across the legacy context ID and the native issue column > - With that OR, PostgreSQL can join runs only by company. It then tests each recent comment against every run of its company and reads each run's context > - On a hosted staging database, this statement used about 23 s per call on average (384 calls, about 8,900 s in total). One call ran for 35 s. Other requests waited behind it > - This pull request reads the check in three stages and probes the two run bindings separately with existing indexes > - The benefit is the same result with bounded, index-driven work for each recent comment ## Linked Issues or Issue Description **What happened?** The reconciler's user-comment lost-wake check was one of the most expensive statements on a hosted staging database. Query statistics showed 384 calls with a mean of about 23 s. One active call was seen at 35 s. The database compute was busy, and short authentication reads waited behind long statements. The statement matched runs with `(context_snapshot->>'issueId' = issue OR native_issue_id = issue)` in three subqueries. The OR gives PostgreSQL no join key except `company_id`. A local plan on synthetic data showed the cost: a hash join of 273 comments against all 38,002 runs, which took 1.0 s of a 1.3 s statement. The work grows with recent comments multiplied by the company's run count, and each test reads a run context. **Expected behavior** Each recent comment uses a small number of indexed probes. The check returns the same comments in the same order, with the same limit and grace logic. **Steps to reproduce** 1. Seed a company with a deep run history and a few hundred recent user comments. 2. Let some recent comments have no wake (for example, after a restart lost the wake of a steered comment). 3. Run the check. PostgreSQL can pick a hash join on `company_id` with the OR as a join filter. With nested loops off, the old statement takes seconds. The local benchmark below shows this. **Paperclip version or commit** Baseline: `02bfe71a6b`. **Deployment mode** Server with PostgreSQL. Local proof uses disposable PostgreSQL 18.1. Production and staging use hosted PostgreSQL. Related: the earlier change that split native and legacy bindings for other ownership reads (see the "Execution recovery query performance" section in `doc/DATABASE.md`). A search found no open pull request for this check. ## What Changed - `checkUserCommentsWithoutWake` reads in three materialized stages: 1. Recent user comments on assigned, non-conversation tasks that no wake request carries. This stage keeps the same issue, status and agent-closed rules. 2. Of those, comments with no live or later run and with at least one earlier run. 3. Of those, comments that no live run acknowledged in an item event (the steer check). Only this stage reads event payloads. - Each run lookup is two probes: one on the legacy context ID and one on the native issue column. Both use existing indexes. `exists (A or B)` becomes `exists (A) or exists (B)`, which is logically the same. - Each probe ends with `offset 0`. This keeps it a per-row index lookup. Without it, PostgreSQL can build a hashed subplan over all runs, or a hash join over all item events of the company. The repository already uses this fence in `execution-blocker.ts`. - The final `order by c.created_at limit N`, the projection, the grace window and the reported reason do not change. - Add `scripts/benchmark-comment-wake-check.ts`. It loads the check from an explicit baseline commit and from the working tree, and it compares the complete service result and the complete positional SQL rows before it records time. - Add a regression test for each run binding on its own: native only, legacy only, live run, later run, steered run, an acknowledgement before the comment, a log-only mention, a run that ended before the comment, and a run in another company. - Document the check and the benchmark command in `doc/DATABASE.md`. - No schema change, index, migration, or new startup work. ## Verification - `pnpm exec vitest run server/src/__tests__/ownership-reconciliation-report.test.ts --maxWorkers=1`: 34 passed. This includes the existing lost-wake, restart, steered and agent-closed tests and the new binding test. - `pnpm --filter @paperclipai/server typecheck`: passed. - `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-comment-wake-check.ts --baseline-ref=02bfe71a6b --samples=7` Benchmark setup: PostgreSQL 18.1 (embedded, local), `shared_buffers` 128 MB, `work_mem` 4 MB. All data is synthetic, with md5-hex filler so contexts and payloads do not compress well. There are two companies per fixture. Two warmup rounds, then 7 timed samples for each arm, in alternating order. Each sample runs the real check function. Before any timing, each fixture asserts that the complete check result and every positional SQL row are equal for baseline and treatment, and that the fixture reports at least one owed comment. Every timed sample also asserts the same result. | Fixture | Runs / recent comments / owed | Baseline median | Treatment median | Speedup | Hash-join plan, baseline → treatment | | --- | --- | --- | --- | --- | --- | | small | 2,541 / 50 / 6 | 8.5 ms | 2.6 ms | 3.2x | 19 ms → 3 ms | | representative | 75,502 / 750 / 75 | 26.4 ms | 23.3 ms | 1.1x | 2,262 ms → 17 ms | | lost steer wakes | 38,002 / 750 / 75 | 454.7 ms | 66.9 ms | 6.8x | 2,717 ms → 68 ms | | deep history | 150,750 / 1,125 / 114 | 26.9 ms | 18.8 ms | 1.4x | 4,006 ms → 34 ms | | busy hour | 128,750 / 5,625 / 200 (limit) | 79.2 ms | 47.4 ms | 1.7x | 20,047 ms → 75 ms | | long steered runs (stress) | 19,100 / 600 / 60 | 626.4 ms | 457.4 ms | 1.4x | 5,871 ms → 502 ms | - The table above used baseline `02bfe71a6b`, the branch base at that time. After the rebase, a repeat run with baseline `8004ad069e` (the new merge base, with the helper guard fixed to run from the repository root) gave the same picture on a busier machine: small 15.4 → 9.1 ms, lost steer wakes 1,733 → 139 ms, busy hour 103 → 65 ms, all with equal results. - The "lost steer wakes" fixture shows the slow plan with default planner settings. PostgreSQL hashed all runs by company and tested 273 comments against 38,002 runs. - The last column runs each statement once with `enable_nestloop = off`. This makes the company-only hash join that the old statement allows visible, and it shows that the new statement has no such plan. The result is checked as equal in this mode too. Staging's 23 s mean matches this growth: recent comments multiplied by the company's run count. - Shared buffers fell in every fixture except "small" with default settings (for example, busy hour 176,416 → 89,803). - Neutral path: in the "long steered runs" stress fixture, every steered comment lost its wake. These comments correctly reach the steer check. That check must read the payload of each item event after the comment in the run that was live, so it is the remaining cost. In normal operation a comment sent to a live run has a deferred wake that carries it, so stage one removes it. ## Risks - Low risk. The predicate is the same conjunction of conditions. Only the evaluation order and the probe shape change, and the benchmark asserts complete-row equality on six fixtures and in both planner modes. - Rows with the same `created_at` have no defined order, as before. The benchmark uses distinct times so positional comparison is meaningful. - `offset 0` stops PostgreSQL from turning these probes into joins. If a task someday has a very large number of runs, each probe still uses the company and issue indexes. - Upgrade safety: this change reads saved run, wake and event state only. It changes no persisted shape, run identity, or saved session, so state saved by the previous release carries forward unchanged. A cutover test is not applicable. - Not verified on the hosted database. The local proof uses synthetic data, and I did not capture a hosted plan. Confirm with a fresh query statistics window after deployment. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use (shell, file editing, local PostgreSQL) in Claude Code. ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/DATABASE.md | 16 ++ scripts/benchmark-comment-wake-check.ts | 266 ++++++++++++++++++ .../ownership-reconciliation-report.test.ts | 58 ++++ .../recovery/ownership-invariant-report.ts | 71 +++-- 4 files changed, 384 insertions(+), 27 deletions(-) create mode 100644 scripts/benchmark-comment-wake-check.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index c3cf5eddf7c..6ee2badb769 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -1929,6 +1929,22 @@ delivery queries, including complete positional results and database plans. All data is synthetic. Query timings exclude process liveness probes, HTTP and provider work. +The reconciler's lost-wake check for user comments reads in three stages. +The first stage keeps recent user comments that no wake request carries. The +second stage removes comments with a live or later run, and it requires an +earlier run. Each stage probes native and legacy run bindings separately with +existing indexes. Only the remaining comments reach the steer check, which +reads run event payloads. An OR across both bindings let PostgreSQL hash the +runs by company only and test each comment against every run of its company. +No new index or migration is needed. + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-comment-wake-check.ts --baseline-ref=02bfe71a6b +``` + +This disposable fixture compares the complete check result, the complete +positional SQL rows, plans, and timings with and without nested-loop joins. + ### Agent runtime synchronization journal Migration 0347 adds the opt-in `agent_runtime` resource stream. Database triggers diff --git a/scripts/benchmark-comment-wake-check.ts b/scripts/benchmark-comment-wake-check.ts new file mode 100644 index 00000000000..b8a475902a3 --- /dev/null +++ b/scripts/benchmark-comment-wake-check.ts @@ -0,0 +1,266 @@ +/** + * Local red/green proof for the reconciler's user-comment wake check + * (`checkUserCommentsWithoutWake` in the ownership invariant report). + * + * Loads the check from an explicit pre-change revision and from the working + * tree, runs both against the same synthetic PostgreSQL fixture, and asserts + * that the complete service result and the complete positional SQL rows are + * equal before timing anything. + * + * pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-comment-wake-check.ts --baseline-ref= + * + * Options: --fixture= (repeatable), --samples=. + */ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { createHash, randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import type { SQL } from "drizzle-orm"; +import { createDb, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; + +type Check = (db: Db, now: Date, limit: number, processStartedAt: Date) => Promise; +type Query = { sql: string; params: unknown[] }; + +const baselineRef = process.argv.find(value => value.startsWith("--baseline-ref="))?.slice(15); +assert.ok(baselineRef, "Pass an explicit pre-change --baseline-ref="); +// Run every git command from the repository root: the documented command runs from server/. +const root = fileURLToPath(new URL("../", import.meta.url)); +const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8", cwd: root }).trim(); +const samplesPerArm = Number(process.argv.find(value => value.startsWith("--samples="))?.slice(10) ?? 5); +const sourcePath = "server/src/services/recovery/ownership-invariant-report.ts"; +// Every helper the check imports must be unchanged, or the baseline is contaminated. +const changedRuntime = execFileSync("git", ["diff", "--name-only", baselineRevision, "--", + "server/src/services", "packages/db/src", "packages/shared/src", "pnpm-lock.yaml"], { encoding: "utf8", cwd: root }) + .trim().split("\n").filter(Boolean); +assert.deepEqual(changedRuntime.filter(file => file !== sourcePath), [], "The baseline must not import changed runtime helpers"); + +const baselineSource = execFileSync("git", ["show", `${baselineRevision}:${sourcePath}`], { encoding: "utf8", cwd: root }); +const baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.comment-wake-baseline-")); +const originalUrl = new URL(`../${sourcePath}`, import.meta.url); +const rewritten = baselineSource.replace(/from "(\.{1,2}\/[^"]+)"/g, + (_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, originalUrl)))}`) + + "\nexport { checkUserCommentsWithoutWake as __benchmarkCheck };\n"; +await fs.writeFile(path.join(baselineDir, "ownership-invariant-report.ts"), rewritten); +const baseline = (await import(pathToFileURL(path.join(baselineDir, "ownership-invariant-report.ts")).href) as { __benchmarkCheck: Check }).__benchmarkCheck; +const treatment = (await import("../server/src/services/recovery/ownership-invariant-report.js") as unknown as { checkUserCommentsWithoutWake: Check }).checkUserCommentsWithoutWake; + +const fixtures = [ + // `steerWakeLostEvery`: a comment sent to a live run normally has a deferred + // wake that carries it. Steered comments on issues with n % steerWakeLostEvery = 2 + // lost that wake (a restart), so only the steer check can rule them out. + // Few comments, shallow history: the check must stay cheap. + { name: "small", issues: 300, runs: 2_000, eventsPerRun: 6, hotIssues: 20, commentsPerHotIssue: 2, steerEvents: 40, payloadBytes: 1_024, contextBytes: 1_024, steerWakeLostEvery: 6 }, + // A busy instance: deep run history and wide contexts, a few hundred recent comments. + { name: "representative", issues: 4_000, runs: 60_000, eventsPerRun: 6, hotIssues: 200, commentsPerHotIssue: 3, steerEvents: 400, payloadBytes: 2_048, contextBytes: 8_192, steerWakeLostEvery: 60 }, + // The same instance after a restart lost the wakes of steered comments: more + // comments reach the run checks. + { name: "lost-steer-wakes", issues: 4_000, runs: 30_000, eventsPerRun: 10, hotIssues: 200, commentsPerHotIssue: 3, steerEvents: 400, payloadBytes: 2_048, contextBytes: 4_096, steerWakeLostEvery: 6 }, + // Deeper history: twice the runs per company with the same recent traffic. + { name: "deep-history", issues: 6_000, runs: 120_000, eventsPerRun: 2, hotIssues: 300, commentsPerHotIssue: 3, steerEvents: 200, payloadBytes: 2_048, contextBytes: 8_192, steerWakeLostEvery: 60 }, + // A busy hour: thousands of recent user comments over a deep run history. + { name: "busy-hour", issues: 6_000, runs: 100_000, eventsPerRun: 2, hotIssues: 1_500, commentsPerHotIssue: 3, steerEvents: 100, payloadBytes: 2_048, contextBytes: 2_048, steerWakeLostEvery: 60 }, + // Stress: every steered comment lost its wake, and each steered run has thousands + // of large item events after the comment. + { name: "long-steered-runs", issues: 2_000, runs: 15_000, eventsPerRun: 8, hotIssues: 120, commentsPerHotIssue: 4, steerEvents: 2_000, payloadBytes: 4_096, contextBytes: 4_096, steerWakeLostEvery: 6 }, +]; + +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(); +let database: Awaited> | undefined; +let db: Db | undefined; + +/** Synthetic, incompressible text of about `bytes` characters (md5 hex chain). */ +const filler = (bytes: number, seed: string) => + `(select string_agg(md5(${seed}::text || ':' || g::text), '') from generate_series(1, ${Math.max(1, Math.ceil(bytes / 32))}) g)`; + +async function seed(db: Db, fixture: typeof fixtures[number], now: Date) { + const run = (text: string, params: unknown[] = []) => db.$client.unsafe(text, params as never[]); + for (const [company, scale] of [[randomUUID(), 1], [randomUUID(), 0.25]] as const) { + const prefix = `B${company.slice(0, 6).toUpperCase()}`; + const issueCount = Math.ceil(fixture.issues * scale), runCount = Math.ceil(fixture.runs * scale); + const hot = Math.ceil(fixture.hotIssues * scale); + await run(`insert into companies (id, name, issue_prefix) values ($1::uuid, 'Synthetic', $2)`, [company, prefix]); + await run(`insert into agents (company_id, name, adapter_type) select $1::uuid, 'agent-' || n, 'codex_local' from generate_series(1, 20) n`, [company]); + await run(`create temp table if not exists bench_agents (company_id uuid, n int, id uuid)`); + await run(`insert into bench_agents select company_id, row_number() over (order by id), id from agents where company_id = $1::uuid`, [company]); + // Issue n: status by n % 10; the first `hot` issues receive recent user comments. + await run(`insert into issues (company_id, title, status, assignee_agent_id, identifier, issue_number) + select $1::uuid, 'I-' || s.num, (array['done','done','todo','in_progress','done','cancelled','todo','in_progress','done','todo'])[1 + s.num % 10], + (select a.id from bench_agents a where a.company_id = $1::uuid and a.n = 1 + s.num % 20), $2 || '-' || s.num, s.num + from generate_series(1, $3::int) s(num)`, [company, prefix, issueCount]); + await run(`create temp table if not exists bench_issues (company_id uuid, n int, id uuid, agent_id uuid)`); + await run(`insert into bench_issues select company_id, issue_number, id, assignee_agent_id from issues where company_id = $1::uuid`, [company]); + // Run history over 14 days. Legacy runs bind by context, native runs by column, some by both. + await run(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, native_issue_id, context_snapshot, + created_at, started_at, finished_at) + select $1::uuid, i.agent_id, 'succeeded', 'assignment', case when r % 3 <> 0 then i.id end, + case when r % 3 = 1 then '{}'::jsonb else jsonb_build_object('issueId', i.id::text, 'prompt', ${filler(fixture.contextBytes, "r")}) end, + $2::timestamptz - interval '2 hours' - (r * interval '14 days' / $3::int), $2::timestamptz - interval '2 hours' - (r * interval '14 days' / $3::int), + $2::timestamptz - interval '2 hours' - (r * interval '14 days' / $3::int) + interval '20 minutes' + from generate_series(1, $3::int) r join bench_issues i on i.company_id = $1::uuid and i.n = 1 + (r * 7919) % $4::int`, + [company, now.toISOString(), runCount, issueCount]); + await run(`insert into heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, created_at, payload) + select r.company_id, r.id, r.agent_id, e, (array['item.started','item.completed','item.started','log','item.completed'])[1 + e % 5], + r.started_at + e * interval '1 second', jsonb_build_object('prpEvent', jsonb_build_object('itemId', 'item-' || e, 'text', ${filler(fixture.payloadBytes, "r.id::text || e")})) + from heartbeat_runs r cross join generate_series(1, $2::int) e where r.company_id = $1::uuid`, [company, fixture.eventsPerRun]); + // Hot issues: one recent run each, then user comments in the last hour. Shape by issue n % 6: + // 0,1 normal (a wake carries each comment, a later run reads it); 2 steered (the run that + // was live at comment time acknowledged it in an item event); 3 lost (no wake, no later run: + // the reported case); 4 live run (queued); 5 carried by a goal wake context list. + await run(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, native_issue_id, context_snapshot, created_at, started_at, finished_at) + select i.company_id, i.agent_id, case when i.n % 6 = 4 then 'queued' else 'succeeded' end, 'assignment', + case when i.n % 2 = 0 then i.id end, jsonb_build_object('issueId', i.id::text, 'prompt', ${filler(fixture.contextBytes, "i.n")}), + $2::timestamptz - interval '58 minutes', $2::timestamptz - interval '58 minutes', + case when i.n % 6 = 2 then $2::timestamptz - interval '1 minute' when i.n % 6 = 4 then null else $2::timestamptz - interval '57 minutes' end + from bench_issues i where i.company_id = $1::uuid and i.n <= $3::int`, [company, now.toISOString(), hot]); + await run(`insert into issue_comments (company_id, issue_id, author_user_id, body, created_at) + select i.company_id, i.id, 'user-1', 'comment ' || k, + -- Distinct times: the check orders by creation time only. + $2::timestamptz - interval '55 minutes' + ((k - 1) * $3::int + i.n) * (interval '50 minutes' / ($3::int * $4::int)) + + $5::int * interval '1 millisecond' + from bench_issues i cross join generate_series(1, $4::int) k where i.company_id = $1::uuid and i.n <= $3::int`, + [company, now.toISOString(), hot, fixture.commentsPerHotIssue, scale === 1 ? 0 : 500]); + // Older comment history outside the lookback window. + await run(`insert into issue_comments (company_id, issue_id, author_user_id, body, created_at) + select i.company_id, i.id, 'user-1', 'old', $2::timestamptz - interval '2 hours' - (k * interval '3 hours') + from bench_issues i cross join generate_series(1, 8) k where i.company_id = $1::uuid`, [company, now.toISOString()]); + await run(`insert into agent_wakeup_requests (company_id, agent_id, source, status, payload, created_at, updated_at) + select c.company_id, i.agent_id, 'automation', case when i.n % 6 = 2 then 'coalesced' else 'completed' end, + case when i.n % 6 = 5 then jsonb_build_object('issueId', c.issue_id::text, '_paperclipWakeContext', jsonb_build_object('wakeCommentIds', jsonb_build_array(c.id::text))) + else jsonb_build_object('issueId', c.issue_id::text, 'commentId', c.id::text) end, c.created_at, c.created_at + from issue_comments c join bench_issues i on i.id = c.issue_id + where c.company_id = $1::uuid and (i.n % 6 in (0, 1, 5) or (i.n % 6 = 2 and i.n % $3::int <> 2) + or c.created_at < $2::timestamptz - interval '2 hours')`, [company, now.toISOString(), fixture.steerWakeLostEvery]); + // Normal comments get a run after them. + await run(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, context_snapshot, created_at, started_at, finished_at) + select c.company_id, i.agent_id, 'succeeded', 'automation', jsonb_build_object('issueId', c.issue_id::text, 'commentId', c.id::text), + c.created_at + interval '2 seconds', c.created_at + interval '2 seconds', c.created_at + interval '30 seconds' + from issue_comments c join bench_issues i on i.id = c.issue_id + where c.company_id = $1::uuid and i.n % 6 in (0, 1) and c.created_at >= $2::timestamptz - interval '1 hour'`, [company, now.toISOString()]); + // Steered runs: many large item events after each comment; one acknowledges the comment. + await run(`insert into heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, created_at, payload) + select r.company_id, r.id, r.agent_id, 1000 + e, (array['item.started','item.completed'])[1 + e % 2], + r.started_at + interval '2 minutes' + e * (interval '55 minutes' / $2::int), + jsonb_build_object('prpEvent', jsonb_build_object('itemId', 'item-' || e, 'text', ${filler(fixture.payloadBytes, "r.id::text || e")})) + from heartbeat_runs r join bench_issues i on i.company_id = r.company_id + and (r.native_issue_id = i.id or r.context_snapshot->>'issueId' = i.id::text) + cross join generate_series(1, $2::int) e + where r.company_id = $1::uuid and i.n <= $3::int and i.n % 6 = 2 and r.started_at >= $4::timestamptz - interval '1 hour'`, + [company, fixture.steerEvents, hot, now.toISOString()]); + await run(`insert into heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, created_at, payload) + select c.company_id, r.id, r.agent_id, 100000 + row_number() over (partition by r.id order by c.created_at), 'item.completed', c.created_at + interval '3 seconds', + jsonb_build_object('prpEvent', jsonb_build_object('itemId', 'provider:steer:' || c.id::text, 'payload', jsonb_build_object('kind', 'steering_acknowledgement'))) + from issue_comments c join bench_issues i on i.id = c.issue_id + join heartbeat_runs r on r.company_id = c.company_id and r.native_issue_id = c.issue_id and r.started_at < c.created_at and r.finished_at >= c.created_at + where c.company_id = $1::uuid and i.n % 6 = 2 and c.created_at >= $2::timestamptz - interval '1 hour'`, [company, now.toISOString()]); + // Closures of done tasks: agent closes even n, a person closes odd n. + await run(`insert into activity_log (company_id, actor_type, actor_id, action, entity_type, entity_id, details, created_at) + select i.company_id, case when i.n % 2 = 0 then 'agent' else 'user' end, 'actor', 'issue.updated', 'issue', i.id::text, + jsonb_build_object('toStatus', 'done'), $2::timestamptz - interval '3 hours' + from bench_issues i join issues s on s.id = i.id where i.company_id = $1::uuid and s.status = 'done'`, [company, now.toISOString()]); + } + await run("analyze"); +} + +function captureQuery(db: Db, check: Check, now: Date, limit: number, processStartedAt: Date): Promise<{ query: Query; result: unknown }> { + let query: Query | undefined; + const dialect = (db as unknown as { dialect: { sqlToQuery(sql: SQL): Query } }).dialect; + const captureDb = new Proxy(db, { get(target, key) { + if (key !== "execute") return Reflect.get(target, key); + return (statement: SQL) => { query = dialect.sqlToQuery(statement); return target.execute(statement); }; + } }); + return check(captureDb, now, limit, processStartedAt).then(result => { assert.ok(query, "query captured"); return { query: query!, result }; }); +} + +try { + database = await startEmbeddedPostgresTestDatabase("comment-wake-benchmark-"); + db = createDb(database.connectionString, { maxConnections: 1, idleTimeoutSeconds: 0 }); + await db.$client.unsafe("set client_min_messages = warning"); + 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', 'random_page_cost') order by name"); + const selectors = process.argv.filter(value => value.startsWith("--fixture=")).map(value => value.slice(10)); + const results = []; + for (const fixture of fixtures.filter(value => !selectors.length || selectors.includes(value.name))) { + await db.$client.unsafe("TRUNCATE companies CASCADE"); + await db.$client.unsafe("drop table if exists bench_agents, bench_issues"); + const now = new Date(), processStartedAt = new Date(now.getTime() - 2 * 60 * 60_000), limit = 200; + const seedStart = performance.now(); + await seed(db, fixture, now); + const seedMs = performance.now() - seedStart; + const counts = await db.$client.unsafe(`select (select count(*) from heartbeat_runs) runs, (select count(*) from heartbeat_run_events) events, + (select count(*) from issue_comments where created_at >= now() - interval '1 hour') recent_comments, + (select pg_size_pretty(pg_total_relation_size('heartbeat_run_events'))) events_size, + (select pg_size_pretty(pg_total_relation_size('heartbeat_runs'))) runs_size`); + const before = await captureQuery(db, baseline, now, limit, processStartedAt); + const after = await captureQuery(db, treatment, now, limit, processStartedAt); + assert.deepEqual(after.result, before.result, `${fixture.name}: complete service result differs`); + const rows = async (query: Query) => db!.$client.unsafe(query.sql, query.params as never[]).values(); + const expected = await rows(before.query), actual = await rows(after.query); + assert.ok(expected.length > 0, `${fixture.name}: the fixture must report owed comments`); + // Order is part of the contract: compare positionally, then as multisets. + assert.deepEqual(actual, expected, `${fixture.name}: complete positional SQL output differs`); + assert.deepEqual(multiset(actual), multiset(expected)); + const samples = { before: [] as number[], after: [] as number[] }; + for (let iteration = 0; iteration < samplesPerArm + 2; iteration++) { + for (const arm of iteration % 2 ? ["after", "before"] as const : ["before", "after"] as const) { + const start = performance.now(); + const result = await (arm === "before" ? baseline : treatment)(db, now, limit, processStartedAt); + const ms = performance.now() - start; + assert.deepEqual(result, before.result); + if (iteration >= 2) samples[arm].push(ms); + } + } + const plans: Record = {}; + const buffers: Record = {}; + for (const [arm, captured] of [["before", before], ["after", after]] as const) { + const [row] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${captured.query.sql}`, captured.query.params as never[]); + plans[arm] = row["QUERY PLAN"]; + const plan = (row["QUERY PLAN"] as { Plan: Record }[])[0]!.Plan; + buffers[arm] = Number(plan["Shared Hit Blocks"]) + Number(plan["Shared Read Blocks"]); + } + const medians = { before: median(samples.before), after: median(samples.after) }; + // Hash-join plan: without nested loops, PostgreSQL can join the baseline's + // OR-bound run lookups only by company, testing each comment against every + // run of its company. The treatment's per-comment probes have no such plan. + // One sample each, outside the timed interval, with the same result check. + const hashPlan: Record = {}; + if (!process.argv.includes("--skip-hash-plan")) { + await db.$client.unsafe("set enable_nestloop = off"); + await db.$client.unsafe("set statement_timeout = '300s'"); + try { + for (const [arm, captured] of [["before", before], ["after", after]] as const) { + const start = performance.now(); + try { + assert.deepEqual(await rows(captured.query), expected, `${fixture.name}: hash-plan output differs`); + hashPlan[arm] = performance.now() - start; + } catch (error) { + if ((error as { code?: string }).code !== "57014") throw error; + hashPlan[arm] = "timeout after 300 s"; + } + } + } finally { + await db.$client.unsafe("reset enable_nestloop"); + await db.$client.unsafe("reset statement_timeout"); + } + } + results.push({ fixture, counts: counts[0], seedMs, reportedRows: expected.length, queries: { before: before.query.sql, after: after.query.sql }, + samples, medians, buffers, hashPlan, plans }); + process.stderr.write(`${fixture.name}: ${JSON.stringify(counts[0])} owed=${expected.length}; ` + + `median ${medians.before.toFixed(1)} -> ${medians.after.toFixed(1)} ms (${(medians.before / medians.after).toFixed(1)}x); ` + + `buffers ${buffers.before} -> ${buffers.after}; hash plan ${JSON.stringify(hashPlan)}; samples before=${samples.before.map(v => v.toFixed(0)).join(",")} after=${samples.after.map(v => v.toFixed(0)).join(",")}\n`); + } + const treatmentSource = await fs.readFile(path.join(root, sourcePath)); + const output = JSON.stringify({ baselineRevision, treatmentRevision: execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8", cwd: root }).trim(), + sourcePath, baselineSha256: createHash("sha256").update(baselineSource).digest("hex"), + treatmentSha256: createHash("sha256").update(treatmentSource).digest("hex"), version, settings, + warmups: 2, timedSamplesPerArm: samplesPerArm, setupAndProofMs: performance.now() - started, results }, null, 2); + const out = process.argv.find(value => value.startsWith("--out="))?.slice(6); + if (out) await fs.writeFile(out, output); else console.log(output); +} 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__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index e9e92261278..40e50b67da0 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -723,6 +723,64 @@ describe("report-only ownership reconciliation", () => { .toEqual([lost.commentId]); }); + // The check reads a task's runs through two separate bindings (legacy + // context or native column) and reads event payloads only for comments that + // the indexed checks keep. Each binding must count alone, and only in the + // comment's company. + it("binds each task run by its legacy context or its native column in the user-comment wake check", async () => { + const otherCompanyId = randomUUID(), otherAgentId = randomUUID(); + await db.insert(companies).values({ id: otherCompanyId, name: "Other company", issuePrefix: "OTH" }); + await db.insert(agents).values({ id: otherAgentId, companyId: otherCompanyId, name: "Other", role: "engineer", adapterType: "codex_local" }); + const task = async (title: string, runs: Array<{ binding: "context" | "native" | "both"; status?: string; + startedAt: Date; finishedAt?: Date | null; companyId?: string }>, commentAt: Date) => { + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title, status: "todo", assigneeAgentId: agentId }); + const runIds: string[] = []; + for (const run of runs) { + const runCompanyId = run.companyId ?? companyId; + const [row] = await db.insert(heartbeatRuns).values({ companyId: runCompanyId, agentId: runCompanyId === companyId ? agentId : otherAgentId, + status: run.status ?? "succeeded", invocationSource: "manual", + contextSnapshot: run.binding === "native" ? {} : { issueId }, + nativeIssueId: run.binding === "context" ? null : issueId, + startedAt: run.startedAt, createdAt: run.startedAt, finishedAt: run.finishedAt === undefined ? minutesAgo(3) : run.finishedAt }).returning(); + runIds.push(row!.id); + } + const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorUserId: "user-1", body: title, + createdAt: commentAt }).returning(); + return { issueId, runIds, commentId: comment!.id }; + }; + const steerEvent = (runId: string, commentId: string, eventType: string, createdAt: Date) => ({ companyId, runId, agentId, + seq: 1, eventType, stream: "system", level: "info", sourceEventId: randomUUID(), createdAt, + payload: { prpEvent: { itemId: `provider:steer:${commentId}`, payload: { kind: "steering_acknowledgement" } } } }); + const lostNative = await task("lost, native prior run", [{ binding: "native", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }], minutesAgo(8)); + const lostContext = await task("lost, legacy prior run", [{ binding: "context", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }], minutesAgo(7)); + await task("no run at all", [], minutesAgo(8)); + await task("only another company's run", [{ binding: "both", startedAt: minutesAgo(10), finishedAt: minutesAgo(9), companyId: otherCompanyId }], minutesAgo(8)); + await task("native later run", [{ binding: "context", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }, + { binding: "native", startedAt: minutesAgo(5) }], minutesAgo(8)); + await task("legacy later run", [{ binding: "native", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }, + { binding: "context", startedAt: minutesAgo(5) }], minutesAgo(8)); + await task("native live run", [{ binding: "native", status: "running", startedAt: minutesAgo(10), finishedAt: null }], minutesAgo(8)); + const nativeSteered = await task("native steered run", [{ binding: "native", startedAt: minutesAgo(10) }], minutesAgo(8)); + const contextSteered = await task("legacy steered run", [{ binding: "context", startedAt: minutesAgo(10) }], minutesAgo(8)); + const earlyAck = await task("acknowledged before the comment", [{ binding: "native", startedAt: minutesAgo(10) }], minutesAgo(6)); + const logOnly = await task("named only in a log event", [{ binding: "context", startedAt: minutesAgo(10) }], minutesAgo(5)); + const otherRunAck = await task("acknowledged by a run that started after it", [{ binding: "native", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }], minutesAgo(4)); + await db.insert(heartbeatRunEvents).values([ + steerEvent(nativeSteered.runIds[0]!, nativeSteered.commentId, "item.completed", minutesAgo(7)), + steerEvent(contextSteered.runIds[0]!, contextSteered.commentId, "item.started", minutesAgo(7)), + steerEvent(earlyAck.runIds[0]!, earlyAck.commentId, "item.completed", minutesAgo(7)), + steerEvent(logOnly.runIds[0]!, logOnly.commentId, "log", minutesAgo(4)), + // The run that names this comment ended before it: not a steer of this comment. + steerEvent(otherRunAck.runIds[0]!, otherRunAck.commentId, "item.completed", minutesAgo(3)), + ]); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false, + processStartedAt: minutesAgo(30) }); + // Oldest first, as the check orders them. + expect(report.violations.filter(row => row.reason === "user_comment_without_wake").map(row => row.recordId)) + .toEqual([lostNative.commentId, lostContext.commentId, earlyAck.commentId, logOnly.commentId, otherRunAck.commentId]); + }); + it("keeps a backstop deadline for a lifted pause still inside its grace window", async () => { const updatedAt = minutesAgo(1); await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "deferred_issue_execution", diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index c00cb3b4517..8b2bae2353c 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -1,4 +1,4 @@ -import { and, asc, eq, inArray, isNotNull, isNull, lt, notExists, or, sql } from "drizzle-orm"; +import { and, asc, eq, inArray, isNotNull, isNull, lt, notExists, or, sql, type SQL } from "drizzle-orm"; import { agentWakeupRequests, heartbeatRuns, @@ -467,36 +467,53 @@ async function checkStaleDroppedComments(db: Db, now: Date, limit: number): Prom * finds a user comment on an assigned, non-conversation task with no live run * or lock, newer than every run of the task, that no wake request carries. * The reconciliation worker saves it as a deferred receipt and promotes it. */ -async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number, processStartedAt: Date): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> { +export async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number, processStartedAt: Date): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> { const since = new Date(now.getTime() - INTENT_LOOKBACK_MS).toISOString(); + // A run belongs to the task by its legacy context or its native column. Each + // binding is a separate indexed probe per comment: one OR across both lets + // PostgreSQL hash every run of the company and read each run's context. + // `offset 0` keeps each probe, and the steer check's event lookup, a + // per-row index lookup; without it PostgreSQL can also build a hashed + // subplan or a hash join over all runs or all events of the company. + const taskRuns = (alias: string, condition: SQL) => { + const run = sql.raw(alias); + return sql`(exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id + and ${run}.context_snapshot->>'issueId' = c.issue_id::text and ${condition} offset 0) + or exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id + and ${run}.native_issue_id = c.issue_id and ${condition} offset 0))`; + }; + // Three stages, cheapest first. Most user comments have a wake (stage one) + // or a later run (stage two). Only the few comments left reach the steer + // check, which reads run event payloads. const rows = await db.execute<{ id: string; company_id: string; issue_id: string; created_at: string }>(sql` - select c.id, c.company_id, c.issue_id, c.created_at::text as created_at from issue_comments c - join issues i on i.id = c.issue_id and i.company_id = c.company_id - where c.author_user_id is not null and c.deleted_at is null - and c.author_agent_id is null and c.created_by_run_id is null - and c.created_at >= ${since}::timestamptz - and i.assignee_agent_id is not null and i.conversation_agent_id is null - and (i.status in ('todo', 'in_progress') - or (i.status = 'done' and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)})) - and i.execution_run_id is null - and not exists (select 1 from heartbeat_runs steered join heartbeat_run_events steer_event - on steer_event.company_id = steered.company_id and steer_event.run_id = steered.id - where steered.company_id = c.company_id - and (steered.context_snapshot->>'issueId' = c.issue_id::text or steered.native_issue_id = c.issue_id) - and coalesce(steered.started_at, steered.created_at) <= c.created_at - and (steered.finished_at is null or steered.finished_at >= c.created_at) + with unwoken as materialized ( + select c.id, c.company_id, c.issue_id, c.created_at from issue_comments c + join issues i on i.id = c.issue_id and i.company_id = c.company_id + where c.author_user_id is not null and c.deleted_at is null + and c.author_agent_id is null and c.created_by_run_id is null + and c.created_at >= ${since}::timestamptz + and i.assignee_agent_id is not null and i.conversation_agent_id is null + and (i.status in ('todo', 'in_progress') + or (i.status = 'done' and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)})) + and i.execution_run_id is null + and not exists (select 1 from agent_wakeup_requests w where w.company_id = c.company_id + and w.payload->>'issueId' = c.issue_id::text + and (w.payload->>'commentId' = c.id::text + or coalesce(w.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text)) + ), candidate as materialized ( + select c.id, c.company_id, c.issue_id, c.created_at from unwoken c + where not ${taskRuns("r", sql`(r.status in ('queued', 'running', 'scheduled_retry') + or coalesce(r.started_at, r.created_at) >= c.created_at)`)} + and ${taskRuns("prior", sql`true`)} + ) + select c.id, c.company_id, c.issue_id, c.created_at::text as created_at from candidate c + where not ${taskRuns("steered", sql`coalesce(steered.started_at, steered.created_at) <= c.created_at + and (steered.finished_at is null or steered.finished_at >= c.created_at) + and exists (select 1 from heartbeat_run_events steer_event + where steer_event.company_id = steered.company_id and steer_event.run_id = steered.id and steer_event.event_type in ('item.started', 'item.completed') and steer_event.created_at >= c.created_at - and strpos(steer_event.payload::text, c.id::text) > 0) - and exists (select 1 from heartbeat_runs prior where prior.company_id = c.company_id - and (prior.context_snapshot->>'issueId' = c.issue_id::text or prior.native_issue_id = c.issue_id)) - and not exists (select 1 from heartbeat_runs r where r.company_id = c.company_id - and (r.context_snapshot->>'issueId' = c.issue_id::text or r.native_issue_id = c.issue_id) - and (r.status in ('queued', 'running', 'scheduled_retry') or coalesce(r.started_at, r.created_at) >= c.created_at)) - and not exists (select 1 from agent_wakeup_requests w where w.company_id = c.company_id - and w.payload->>'issueId' = c.issue_id::text - and (w.payload->>'commentId' = c.id::text - or coalesce(w.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text)) + and strpos(steer_event.payload::text, c.id::text) > 0 offset 0)`)} order by c.created_at limit ${Math.min(limit, MAX_RECORDS_PER_KIND)}`); // A comment inside the grace may still get its normal wake: check it again // when the grace ends (a saved deadline), and report only older ones. This