From f69dfc0e23a5896aa9e8338476a421e811e8fad3 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 07:18:17 -0500 Subject: [PATCH] perf: use indexed latest-run probes for task interruption lists (#17969) 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. > - Task lists show the latest run's interruption state for each visible task. > - The old lookup applies a coalesced text expression to company run history. > - That lookup cannot use the separate binding indexes for a small task page. > - This pull request selects the newest run through those existing indexes. > - The benefit is less database work with the same interruption result. ## Linked Issues or Issue Description **What existing behavior does this improve?** The latest-run lookup used by task-list interruption enrichment. **Current behavior** The query scans company history for a coalesced native, first-class, or legacy task ID. A local two-task fixture inspects 1,506 runs to select six candidates. A recent hosted sample recorded about 110 ms per latest-run lookup. This is one part of the list handler, not its whole latency. **Proposed behavior** Probe each existing binding index for the newest candidate. Pick the newest candidate across the three binding branches. Read the same evidence fields in the same SQL statement. Preserve native precedence, first-class precedence, legacy text matching, company scope, and ordering ties. **Reason and benefit** The representative local 100-task page improves from 17.17 ms to 3.00 ms (5.72×). A modern 100-task page with 40 runs per task improves from 26.78 ms to 3.82 ms (7.02×). All positional SQL fields, row order, and full service results match. **Breaking changes** None. Stored history, API fields, visibility filtering, execution authority, and external-conversation state are unchanged. Related behavior: #16965, #17509, #17552. The duplicate search found no open PR for this query rewrite. ROADMAP.md has no duplicate planned feature. ## What Changed - Use three disjoint indexed candidate branches with a lateral newest-run selection. - Guard UUID parameters to preserve the old text matching for noncanonical input. - Keep final evidence projection in the same database snapshot. - Add real PostgreSQL work and contract regressions, a pinned-source benchmark, and a verification note. - Require the plan regression to find real run-table nodes. Record the deep-native-history follow-up without adding indexes or DDL. ## Verification - Baseline red: the real two-task lookup inspects 1,506 run rows and fails the work bound. The unchanged semantic oracle passes. - Treatment green: both PostgreSQL query regressions pass. They cover conflicting bindings, malformed JSON, empty values, noncanonical IDs, duplicates, foreign companies, missing runs, and timestamp ties across branches. - Seven paired PostgreSQL 17.11 cases pass complete positional SQL and full service equality. Each arm has two warmups and five alternating measured samples. Fixture payload sizes are random input bytes before Base64 encoding: 24 KiB becomes 32 KiB of stored text per noise field. - Representative 100 × 12 history: 17.17 ms (16.79–17.48) → 3.00 ms (2.64–4.28). Shared buffer hits: 30,033 → 3,400. - Modern 100 × 40 history: 26.78 ms (25.79–27.40) → 3.82 ms (3.47–4.78). - The benchmark command, all cases, plans, fixture limits, and feedback-loop timing are described in [the verification note](doc/plans/2026-10-27-task-interruption-query-performance.md). - Reproduce with `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-task-interruption.ts --baseline-ref=09ad9adce2cf4742944353a3707a2f908e9bb35e`. - `pnpm -r typecheck` and `pnpm build` passed at `20d46e3a16d10cf3c5399fc1211b17db6a4afdc6`. The current head has the identical production runtime; it adds one test assertion and documentation. - The focused query, real project-access repair/retry, and presentation suites pass: 19 tests across three files. The earlier plugin-sdk loading failure is resolved by normal build setup. - `pnpm test:run` was attempted, then stopped after repeated native embedded PostgreSQL bootstrap failures. The full local suite is incomplete. All 16 affected real-DB executor cases pass when rerun with disposable PostgreSQL 17 in Docker. - Linux CI passes the full server suites on this head. One unchanged Runner close fixture failed on the first CI attempt; its source tree is identical to the passing upstream #17439 baseline. One targeted job retry passed on the original head. Fresh checks run again for the test/documentation review fix. A separate macOS diagnostic fails the unchanged close variants too; that close issue is not diagnosed or counted as a local pass here. - The original head passed 55 checks with one neutral result and two expected Storybook skips. Greptile gave 5/5. The current head adds one nonempty-plan assertion and documents the deep-history follow-up. Its focused PostgreSQL suite passes 2/2. Final head `c1db106d7ab56b530af82d2bb0556feb0eda290b` is mergeable with all CI checks complete and passing, Greptile 5/5, and no unresolved review findings; the documented per-task history limitation remains an explicit follow-up. ## Risks - More index probes add about 0.43 ms to the one-run tiny control. The maximum 1,000-task page with one modern run per task is 9.99 → 14.11 ms, a 4.12 ms regression. Dense history pages save substantially more in the local fixture. - Each branch can still walk one task's own history. The first-class and legacy probes can discard native-bound rows after index lookup. The plan note specifies a 10,000-run follow-up, migration-free alternatives, and a measured index-build budget if indexes prove necessary. This PR does not make work independent of history depth. - Local SQL results do not prove deployed HTTP gains. The affected query averaged about 110 ms in the hosted sample; its critical-path contribution can be smaller because other enrichment reads run concurrently. - The main issue-list SQL and external-conversation gates remain separate work. This PR has no migration or startup job. ## Model Used OpenAI Codex, GPT-6 family, with code execution and repository tool use. The runtime does not expose the exact model ID or context window size. ## 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 --- ...-27-task-interruption-query-performance.md | 124 ++++++++++++++++++ scripts/benchmark-task-interruption.ts | 80 +++++++++++ .../helpers/task-interruption-fixture.ts | 72 ++++++++++ .../__tests__/task-interruption-query.test.ts | 66 ++++++++++ .../task-interruption-presentation.ts | 25 +++- 5 files changed, 362 insertions(+), 5 deletions(-) create mode 100644 doc/plans/2026-10-27-task-interruption-query-performance.md create mode 100644 scripts/benchmark-task-interruption.ts create mode 100644 server/src/__tests__/helpers/task-interruption-fixture.ts create mode 100644 server/src/__tests__/task-interruption-query.test.ts diff --git a/doc/plans/2026-10-27-task-interruption-query-performance.md b/doc/plans/2026-10-27-task-interruption-query-performance.md new file mode 100644 index 00000000000..276e7890f8f --- /dev/null +++ b/doc/plans/2026-10-27-task-interruption-query-performance.md @@ -0,0 +1,124 @@ +# Task interruption latest-run query performance + +The issue list displays a current interruption state for each visible task. +`taskInterruptionPresentations` owns that read-only projection. Its old latest-run +query grouped `coalesce(native_issue_id::text, issue_id::text, +context_snapshot->>'issueId')` across company history. That expression cannot use +the existing individual binding indexes to find a small page of tasks. + +The query now probes the native, first-class, and legacy text indexes separately. +Each branch supplies its newest candidate. A lateral query selects the newest +candidate across those three branches by `created_at DESC, id DESC`. The final +join reads the same interruption fields for that winner in one SQL snapshot. +No history is deleted or hidden. This changes only the latest-run enrichment; +the main issue-list query and external-conversation state gates are unchanged. + +## Contract and regression proof + +Native binding takes precedence over first-class binding, then legacy JSON text. +A non-null mismatching native or first-class binding prevents legacy fallback. +A guarded UUID parameter keeps uppercase and noncanonical caller strings from +matching a UUID where the former text comparison would not. Legacy text keeps +PostgreSQL `->>` behavior for empty, missing, scalar, array, object, and null data. +Company boundaries, duplicate inputs, absent runs, and timestamp ties are covered. +The existing project-access interruption tests exercise real repair and retry +consumers. Presentation remains read-only; execution rechecks its own authority. + +The baseline work regression selects six runs from a two-task page but inspects +1,506 run rows. The treatment meets a deterministic bound of fewer than 30 rows +in any run-table scan on that same fixture. The regression has no timing gate. +The baseline semantics test passes before the optimization. + +## Local measurement + +Run the actual old and new query builders against disposable PostgreSQL: + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-task-interruption.ts --baseline-ref=09ad9adce2cf4742944353a3707a2f908e9bb35e +``` + +The pinned pre-fix service is loaded from Git. The harness rejects changed runtime +dependencies. Its service source is identical to the original discovery baseline +`96727b3cfefb61f85ffe1fbc29a9b6dd6837efe8` (Git blob +`eb489f1ef6d35c9c243307e60e4c31d58afddfa3`). Both arms use the same current dependencies. +The treatment service SHA-256 is +`4607c8382aea8ec9cc76f628caa9b4a031b19dd43ff4beff12e596079742b763`. + +The fixtures use a second company, unrelated histories, mixed binding types, +fixed dates, and incompressible random context/result bodies. Mixed fixtures +retain historical null bindings by disabling only the fixture database's modern +scope-fill trigger during insert. The modern control leaves that trigger enabled. +The harness captures complete generated SQL and parameters, compares every +positional SQL field and row order, and compares full service maps. It warms each +arm twice, alternates five measured samples per arm, and saves local +`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)` plans. + +PostgreSQL 17.11 Linux aarch64 matches the observed hosted major/minor version. +Local settings: 128 MiB shared buffers, 4 MiB work memory, JIT enabled. These are +warm local SQL timings, including driver round trips, not deployed HTTP timings. + +| Page × history per task | Unrelated runs | JSON noise bytes per field | Before median (range) ms | After median (range) ms | Speedup | +| --- | ---: | ---: | ---: | ---: | ---: | +| 1 × 1 | 0 | 128 | 0.61 (0.55–0.88) | 1.04 (0.90–1.92) | 0.58× | +| 100 × 1, mixed | 1,200 | 24,576 | 6.99 (5.89–8.41) | 2.59 (2.49–2.79) | 2.70× | +| 100 × 12, mixed | 1,200 | 24,576 | 17.17 (16.79–17.48) | 3.00 (2.64–4.28) | 5.72× | +| 1 × 120, mixed | 1,200 | 24,576 | 5.05 (4.39–5.89) | 2.45 (2.31–2.88) | 2.06× | +| 100 × 40, mixed | 1,200 | 12,288 | 31.51 (30.68–31.61) | 3.05 (2.50–3.23) | 10.32× | +| 100 × 40, modern | 1,200 | 12,288 | 26.78 (25.79–27.40) | 3.82 (3.47–4.78) | 7.02× | +| 1,000 × 1, modern | 1,200 | 12,288 | 9.99 (9.51–10.79) | 14.11 (13.72–16.33) | 0.71× | + +All seven pairs have exact positional SQL and full service equality. The 100 × 12 +plan drops shared buffer hits from 30,033 to 3,400. The modern dense control drops +45,563 to 7,537. The tiny control adds about 0.43 ms. The maximum-size shallow +modern page adds 4.12 ms (9.99 to 14.11 ms), with buffer hits increasing from +15,467 to 22,008. The extra probes cost more for such wide, shallow history. +These absolute costs are retained as tradeoffs; no adaptive threshold is added. +Each branch can still inspect a task's own history. The native index lacks +creation-time ordering and must sort native candidates. First-class and legacy +indexes also contain native-bound rows; the precedence guards can discard those +rows only after fetching them. A deeply native-bound task can therefore cause +three per-task history walks, even though each branch returns at most one row. +The modern 40-run control includes this effect; it does not establish constant +work at 10,000 runs per task. + +Immediate follow-up for that remaining cost: profile a synthetic 10,000-run, +mostly native-bound task with complete result equality and per-branch +`EXPLAIN (ANALYZE, BUFFERS)` counters. First test a migration-free selection that +avoids repeated walks while retaining precedence and timestamp/ID ties. If an +index is still necessary, evaluate a company/native-issue/creation-time index, +a first-class binding index restricted to `native_issue_id IS NULL`, and a legacy +binding index restricted to both native and first-class bindings being null. +Measure the complete index builds and blocking on a production-size disposable +fixture. If a build exceeds the one-to-two-minute budget, first optimize it and +flag the remaining cost for human review. No partial index or startup DDL is +included in this PR. + +## Feedback loop and practical limit + +| Question | Environment and loop | Measured elapsed | Result | +| --- | --- | --- | --- | +| Does coalesced lookup inspect unrelated history? | Actual service SQL, disposable PostgreSQL | Initial red test 13.55 s | 1,506 inspected rows; semantic oracle passes | +| Does indexed selection preserve the contract? | Two real PostgreSQL regressions | 5.09 s repeated local test loop | Both pass | +| Is the first batched union enough? | Pinned source pair, PostgreSQL 18.6 | 3.73 s inside benchmark | 1.62× at 100 × 12; scans remain | +| Is selective lateral lookup better? | Seven paired PostgreSQL 17.11 cases | 8.39 s full command; 2.68 s database setup | 5.72× representative, 7.02× modern dense | + +Dependency setup took 22.1 s. The baseline's patch configuration did not match its +committed lockfile, so the isolated install needed a local non-frozen resolution. +The lockfile is unchanged in this PR; its effective dependency record is retained +with private raw evidence. A larger existing consumer suite initially failed to +load because plugin-sdk had not been built. After normal build setup, all 19 +focused query and authority-consumer cases passed. Full recursive typecheck and +build passed too. The full local suite was stopped after repeated native +PostgreSQL bootstrap failures; all 16 affected executor cases passed with the +Docker PostgreSQL 17 substitute. The full local suite is not counted as a pass. +The first Linux CI attempt also had one failure in an unchanged Runner close +fixture. One targeted job retry passed. A separate macOS diagnostic of that close +fixture still fails; its cause is not diagnosed by this query optimization. + +The affected hosted query averaged about 110 ms in the discovery window. Even +eliminating that query can save at most its serial critical-path contribution; +other enrichment reads run concurrently. The local measurements do not establish +that contribution or explain the whole issue-list request. This bounded change +is useful alongside the other measured query fixes. Deployed latency remains to +be measured after rollout. No migration, startup job, deployment, or hosted +mutation is part of this PR. diff --git a/scripts/benchmark-task-interruption.ts b/scripts/benchmark-task-interruption.ts new file mode 100644 index 00000000000..43753195907 --- /dev/null +++ b/scripts/benchmark-task-interruption.ts @@ -0,0 +1,80 @@ +import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; +import { execFileSync } from "node:child_process"; +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 { taskInterruptionPresentations } from "../server/src/services/task-interruption-presentation.js"; +import { captureInterruptionQuery, seedInterruptionFixture, type CapturedQuery } from "../server/src/__tests__/helpers/task-interruption-fixture.js"; + +const startedAt = performance.now(); +const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15); +assert.ok(baselineRef, "Pass an explicit pre-fix --baseline-ref"); +const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8" }).trim(); +const modulePath = "server/src/services/task-interruption-presentation.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(file => file !== modulePath), [], "Changed baseline dependency requires explicit isolation"); +const keepAlive = setInterval(() => {}, 1000); +let database: Awaited> | undefined; +let db: Db | undefined, baselineDir: string | undefined; +try { + const oldSource = execFileSync("git", ["show", `${baselineRevision}:${modulePath}`], { encoding: "utf8" }); + const moduleUrl = new URL(`../${modulePath}`, import.meta.url); + const source = oldSource.replace(/from "(\.{1,2}\/[^\"]+)"/g, (_match, specifier: string) => + `from ${JSON.stringify(fileURLToPath(new URL(specifier, moduleUrl)))}`); + baselineDir = await fs.mkdtemp(path.join(fileURLToPath(new URL("../server/scripts/", import.meta.url)), ".interruption-baseline-")); + const baselineFile = path.join(baselineDir, "task-interruption.ts"); + await fs.writeFile(baselineFile, source); + const baseline = await import(pathToFileURL(baselineFile).href) as typeof import("../server/src/services/task-interruption-presentation.js"); + database = await startEmbeddedPostgresTestDatabase("interruption-benchmark-"); + db = createDb(database.connectionString); + const settings = await db.$client.unsafe("select version(), current_setting('shared_buffers') shared_buffers, current_setting('work_mem') work_mem, current_setting('jit') jit"); + const setupMs = performance.now() - startedAt; + const results = []; + for (const fixture of [ + { selected: 1, history: 1, unrelated: 0, payloadBytes: 128 }, + { selected: 100, history: 1, unrelated: 1200, payloadBytes: 24 * 1024 }, + { selected: 100, history: 12, unrelated: 1200, payloadBytes: 24 * 1024 }, + { selected: 1, history: 120, unrelated: 1200, payloadBytes: 24 * 1024 }, + { selected: 100, history: 40, unrelated: 1200, payloadBytes: 12 * 1024 }, + { selected: 100, history: 40, unrelated: 1200, payloadBytes: 12 * 1024, preserveLegacyBindings: false }, + { selected: 1000, history: 1, unrelated: 1200, payloadBytes: 12 * 1024, preserveLegacyBindings: false }, + ]) { + await db.$client.begin(async client => { + await client.unsafe("set local client_min_messages = warning"); + await client.unsafe("truncate heartbeat_runs cascade"); + }); + const seedStart = performance.now(); + const f = await seedInterruptionFixture(db, fixture); + const seedMs = performance.now() - seedStart; + const before = await captureInterruptionQuery(db, baseline.taskInterruptionPresentations, f.companyId, f.input); + const after = await captureInterruptionQuery(db, taskInterruptionPresentations, f.companyId, f.input); + const execute = async (query: CapturedQuery) => db!.$client.unsafe(query.sql, query.params as never[]).values(); + const beforeRows = await execute(before), afterRows = await execute(after); + assert.deepEqual(afterRows, beforeRows, "Every positional SQL field and row order must match"); + assert.deepEqual(await taskInterruptionPresentations(db, f.companyId, f.input), + await baseline.taskInterruptionPresentations(db, f.companyId, f.input), "Full service output must match"); + const samples = { baseline: [] as number[], treatment: [] as number[] }; + for (let round = 0; round < 7; round++) { + for (const arm of round % 2 ? ["treatment", "baseline"] as const : ["baseline", "treatment"] as const) { + const start = performance.now(); + await execute(arm === "baseline" ? before : after); + if (round >= 2) samples[arm].push(performance.now() - start); + } + } + const explain = async (query: CapturedQuery) => (await db!.$client.unsafe(`explain (analyze, buffers, format json) ${query.sql}`, query.params as never[]))[0]!["QUERY PLAN"]; + const summarize = (values: number[]) => { const sorted = [...values].sort((a, b) => a - b); return { medianMs: sorted[2], minMs: sorted[0], maxMs: sorted[4], samplesMs: values }; }; + const baselineTiming = summarize(samples.baseline), treatmentTiming = summarize(samples.treatment); + results.push({ fixture, seedMs, returnedRows: afterRows.length, positionalEquality: true, serviceEquality: true, + baseline: { ...baselineTiming, query: before, plan: await explain(before) }, + treatment: { ...treatmentTiming, query: after, plan: await explain(after) }, + speedup: baselineTiming.medianMs! / treatmentTiming.medianMs! }); + } + console.log(JSON.stringify({ baselineRevision, treatmentRevision: execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8" }).trim(), + treatmentSourceSha256: createHash("sha256").update(await fs.readFile(fileURLToPath(new URL(`../${modulePath}`, import.meta.url)))).digest("hex"), + changedRuntime, settings, node: process.version, setupMs, totalMs: performance.now() - startedAt, warmupsPerArm: 2, measuredSamplesPerArm: 5, results }, null, 2)); +} finally { + try { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true }); } + finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/helpers/task-interruption-fixture.ts b/server/src/__tests__/helpers/task-interruption-fixture.ts new file mode 100644 index 00000000000..9d7bd3ef95e --- /dev/null +++ b/server/src/__tests__/helpers/task-interruption-fixture.ts @@ -0,0 +1,72 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { agents, companies, issues, type Db } from "@paperclipai/db"; +import type { taskInterruptionPresentations } from "../../services/task-interruption-presentation.js"; + +export type InterruptionReader = typeof taskInterruptionPresentations; +export type CapturedQuery = { sql: string; params: unknown[] }; + +export async function captureInterruptionQuery(db: Db, reader: InterruptionReader, + companyId: string, input: Parameters[2]) { + let captured: CapturedQuery | undefined; + const stop = new Error("captured interruption query"); + const captureDb = new Proxy(db, { get(target, key) { + if (key !== "select" && key !== "selectDistinctOn") return Reflect.get(target, key); + return (...args: unknown[]) => { + const builder = Reflect.apply(Reflect.get(target, key), target, args); + const from = builder.from.bind(builder); + builder.from = (...tables: unknown[]) => { + const selection = from(...tables); + selection.execute = async () => { + const query = selection.toSQL() as CapturedQuery; + if (query.sql.includes('"heartbeat_runs"')) captured = query; + throw stop; + }; + return selection; + }; + return builder; + }; + }}); + try { await reader(captureDb, companyId, input); } + catch (error) { if (error !== stop) throw error; } + if (!captured) throw new Error("The real interruption caller did not issue a run query"); + return captured; +} + +export async function seedInterruptionFixture(db: Db, options: { + selected: number; history: number; unrelated: number; payloadBytes: number; preserveLegacyBindings?: boolean; +}) { + const companyId = randomUUID(), otherCompanyId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values([companyId, otherCompanyId].map((id, index) => ({ + id, name: `Synthetic interruption ${index}`, issuePrefix: `I${id.slice(0, 7)}`, + }))); + await db.insert(agents).values({ id: agentId, companyId, name: "Synthetic agent", adapterType: "process" }); + const input = await db.insert(issues).values(Array.from({ length: options.selected }, (_, n) => ({ + companyId, title: `Synthetic task ${n}`, status: "in_progress", assigneeAgentId: agentId, + }))).returning(); + const contextNoise = randomBytes(options.payloadBytes).toString("base64"); + const resultNoise = randomBytes(options.payloadBytes).toString("base64"); + const ids = input.map(issue => issue.id); + if (options.preserveLegacyBindings !== false) await db.$client.unsafe("alter table heartbeat_runs disable trigger heartbeat_runs_set_scope_kind"); + await db.$client.unsafe(`insert into heartbeat_runs + (company_id, agent_id, scope_kind, native_issue_id, issue_id, status, created_at, context_snapshot, result_json) + select $1::uuid, $2::uuid, case when n % 3 = 1 then 'issue' else 'company' end, + case when n % 3 = 0 then task.id end, + case when n % 3 = 1 then task.id end, + case when n = $4::integer then 'failed' else 'succeeded' end, + timestamp '2026-10-27 00:00:00+00' + n * interval '1 second', + jsonb_build_object('issueId', task.id::text, 'noise', $5::text), + jsonb_build_object('configurationIncomplete', false, 'noise', $6::text) + from unnest($3::uuid[]) task(id) cross join generate_series(1, $4::integer) n`, + [companyId, agentId, ids, options.history, contextNoise, resultNoise]); + await db.$client.unsafe(`insert into heartbeat_runs + (company_id, agent_id, status, created_at, context_snapshot, result_json) + select case when n % 10 = 0 then $2::uuid else $1::uuid end, $3::uuid, 'succeeded', + timestamp '2026-10-28 00:00:00+00', + jsonb_build_object('issueId', gen_random_uuid()::text, 'noise', $5::text), + jsonb_build_object('noise', $6::text) + from generate_series(1, $4::integer) n`, + [companyId, otherCompanyId, agentId, options.unrelated, contextNoise, resultNoise]); + if (options.preserveLegacyBindings !== false) await db.$client.unsafe("alter table heartbeat_runs enable trigger heartbeat_runs_set_scope_kind"); + await db.$client.unsafe("analyze heartbeat_runs"); + return { companyId, otherCompanyId, agentId, input }; +} diff --git a/server/src/__tests__/task-interruption-query.test.ts b/server/src/__tests__/task-interruption-query.test.ts new file mode 100644 index 00000000000..69e06c4dde0 --- /dev/null +++ b/server/src/__tests__/task-interruption-query.test.ts @@ -0,0 +1,66 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, createDb, heartbeatRuns, issues, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { taskInterruptionPresentations } from "../services/task-interruption-presentation.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { captureInterruptionQuery, seedInterruptionFixture } from "./helpers/task-interruption-fixture.js"; + +describe("task interruption latest run query", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("interruption-query-"); db = createDb(database.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + it("does not inspect unrelated company history for a small issue page", async () => { + const f = await seedInterruptionFixture(db, { selected: 2, history: 3, unrelated: 1500, payloadBytes: 4096 }); + const query = await captureInterruptionQuery(db, taskInterruptionPresentations, f.companyId, f.input); + const [explain] = await db.$client.unsafe(`explain (analyze, buffers, format json) ${query.sql}`, query.params as never[]); + const inspected: number[] = []; + const visit = (node: Record) => { + if (node["Relation Name"] === "heartbeat_runs") inspected.push( + (Number(node["Actual Rows"] ?? 0) + Number(node["Rows Removed by Filter"] ?? 0)) * Number(node["Actual Loops"] ?? 1)); + for (const child of node.Plans as Record[] ?? []) visit(child); + }; + visit(explain["QUERY PLAN"][0].Plan); + expect(inspected.length).toBeGreaterThan(0); + expect(Math.max(...inspected)).toBeLessThan(30); + const presentations = await taskInterruptionPresentations(db, f.companyId, f.input); + for (const issue of f.input) expect(presentations.get(issue.id)).toMatchObject({ state: "failed", canRetry: true }); + }); + + it("preserves native, first-class and text legacy precedence, ties, and tenant scope", async () => { + const companyId = randomUUID(), otherId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values([companyId, otherId].map(id => ({ id, name: "Selectors", issuePrefix: `S${id.slice(0, 7)}` }))); + await db.insert(agents).values({ id: agentId, companyId, name: "Selector agent", adapterType: "process" }); + const input = await db.insert(issues).values(Array.from({ length: 10 }, (_, n) => ({ companyId, title: `Selector ${n}`, status: "in_progress", assigneeAgentId: agentId }))).returning(); + const base = { companyId, agentId, status: "failed", createdAt: new Date("2026-10-27T00:00:00Z") }; + const winnerIds = Array.from({ length: 7 }, () => randomUUID()); + await db.$client.unsafe("alter table heartbeat_runs disable trigger heartbeat_runs_set_scope_kind"); + await db.insert(heartbeatRuns).values([ + { ...base, id: winnerIds[0], nativeIssueId: input[0]!.id, issueId: input[1]!.id, scopeKind: "issue", contextSnapshot: { issueId: input[2]!.id } }, + { ...base, id: winnerIds[1], issueId: input[1]!.id, scopeKind: "issue", contextSnapshot: { issueId: input[2]!.id } }, + { ...base, id: winnerIds[2], contextSnapshot: { issueId: input[2]!.id } }, + { ...base, nativeIssueId: randomUUID(), contextSnapshot: { issueId: input[3]!.id } }, + { ...base, nativeIssueId: null, issueId: input[9]!.id, scopeKind: "issue", contextSnapshot: { issueId: input[3]!.id } }, + { ...base, contextSnapshot: { issueId: input[3]!.id.toUpperCase() } }, + ...[null, "", [], 5, { issueId: "" }, { issueId: {} }, { issueId: [] }].map(contextSnapshot => ({ ...base, contextSnapshot: contextSnapshot as never })), + { ...base, id: winnerIds[3], nativeIssueId: input[4]!.id, contextSnapshot: null }, + { ...base, id: winnerIds[4], issueId: input[5]!.id, scopeKind: "issue", contextSnapshot: [input[3]!.id] as never }, + { ...base, id: "00000000-0000-0000-0000-000000000001", contextSnapshot: { issueId: input[6]!.id } }, + { ...base, id: "ffffffff-ffff-ffff-ffff-ffffffffffff", nativeIssueId: input[6]!.id, contextSnapshot: { issueId: "" } }, + { ...base, companyId: otherId, nativeIssueId: input[3]!.id, createdAt: new Date("2026-10-28T00:00:00Z") }, + ]); + await db.$client.unsafe("alter table heartbeat_runs enable trigger heartbeat_runs_set_scope_kind"); + const result = await taskInterruptionPresentations(db, companyId, [...input, input[0]!, { ...input[0]!, companyId: otherId }]); + for (let n = 0; n < 3; n++) expect(result.get(input[n]!.id)?.observedRunId).toBe(winnerIds[n]); + expect(result.get(input[3]!.id)).toBeNull(); + expect(result.get(input[4]!.id)?.observedRunId).toBe(winnerIds[3]); + expect(result.get(input[5]!.id)?.observedRunId).toBe(winnerIds[4]); + expect(result.get(input[6]!.id)?.observedRunId).toBe("ffffffff-ffff-ffff-ffff-ffffffffffff"); + expect(result.get(input[7]!.id)).toBeNull(); + expect(result.get(input[8]!.id)).toBeNull(); + const upper = { ...input[4]!, id: input[4]!.id.toUpperCase() }; + expect((await taskInterruptionPresentations(db, companyId, [upper])).get(upper.id)).toBeNull(); + expect(await taskInterruptionPresentations(db, companyId, [])).toEqual(new Map()); + }); +}); diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index eb489f1ef6d..a614a066baf 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -1,6 +1,7 @@ -import { and, desc, eq, inArray, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm"; import { agents, agentWakeupRequests, approvals, companies, heartbeatRuns, issueApprovals, issueRecoveryActions, issues, issueTreeHoldMembers, issueTreeHolds, projects, type Db } from "@paperclipai/db"; +import { unionAll } from "drizzle-orm/pg-core"; import type { TaskInterruptionPresentation } from "@paperclipai/shared"; import { isAiAuthenticationBlocked } from "./ai-auth-failure.js"; import { readSystemRecovery } from "./system-recovery-state.js"; @@ -83,20 +84,34 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i const result = new Map(); if (!selected.length) return result; const ids = selected.map(issue => issue.id); - const runIssue = sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`; + const requested = sql`unnest(${sql.param([...new Set(ids)])}::text[]) as requested_interruption_issues(requested_issue_id)`; + const requestedIssueId = sql`"requested_interruption_issues"."requested_issue_id"`; + const requestedUuid = sql`case when ${requestedIssueId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' + then ${requestedIssueId}::uuid else null end`; + const legacyIssue = sql`${heartbeatRuns.contextSnapshot}->>'issueId'`; + const candidate = (condition: ReturnType) => db.select({ id: heartbeatRuns.id, createdAt: heartbeatRuns.createdAt }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), condition)) + .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); + const candidates = unionAll( + candidate(eq(heartbeatRuns.nativeIssueId, requestedUuid)), + candidate(and(isNull(heartbeatRuns.nativeIssueId), eq(heartbeatRuns.issueId, requestedUuid))), + candidate(and(isNull(heartbeatRuns.nativeIssueId), isNull(heartbeatRuns.issueId), eq(legacyIssue, requestedIssueId))), + ).as("interruption_run_candidates"); + const latestRuns = db.select({ id: candidates.id }).from(candidates) + .orderBy(desc(candidates.createdAt), desc(candidates.id)).limit(1).as("latest_interruption_runs"); const wakeIssue = sql`${agentWakeupRequests.payload}->>'issueId'`; const agentIds = selected.flatMap(issue => issue.assigneeAgentId ? [issue.assigneeAgentId] : []); const projectIds = selected.flatMap(issue => issue.projectId ? [issue.projectId] : []); const [runs, waits, agentRows, projectRows, pauseRows, approvalRows, companyRows] = await Promise.all([ - db.selectDistinctOn([runIssue], { issueId: runIssue, id: heartbeatRuns.id, status: heartbeatRuns.status, + db.select({ issueId: requestedIssueId, id: heartbeatRuns.id, status: heartbeatRuns.status, agentId: heartbeatRuns.agentId, error: sql`case when ${heartbeatRuns.errorCode} = 'setup_failed' and ${heartbeatRuns.error} = 'Project is outside this low-trust boundary.' then ${heartbeatRuns.error} else null end`, errorCode: heartbeatRuns.errorCode, livenessState: heartbeatRuns.livenessState, scheduledRetryAt: heartbeatRuns.scheduledRetryAt, resultJson: sql>`jsonb_build_object('configurationIncomplete', ${heartbeatRuns.resultJson}->'configurationIncomplete', 'projectAccessDenied', ${heartbeatRuns.resultJson}->'projectAccessDenied')`, - }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), inArray(runIssue, ids))) - .orderBy(runIssue, desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)), + }).from(requested).innerJoinLateral(latestRuns, sql`true`).innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, latestRuns.id))) + .orderBy(requestedIssueId), db.selectDistinctOn([wakeIssue], { issueId: wakeIssue, id: agentWakeupRequests.id, requestedAt: agentWakeupRequests.requestedAt }) .from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), inArray(wakeIssue, ids), eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.requestedByActorType, "user")))