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")))