From 81c52e3666128186c5818ea7ab8a65bfe662fee3 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:25:59 -0600 Subject: [PATCH] perf(chat): find terminal GitHub responses by issue run indexes (#20671) 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 GitHub chat channel posts a terminal response when an agent run for a GitHub request stops. A background sweep finds acknowledged response comments (`processed`, not final) whose issue is blocked or whose latest run is terminal > - For each candidate, the sweep reads the status of the latest run of the assigned agent for the issue. It matches runs with `coalesce(native_issue_id::text, context_snapshot->>'issueId')` > - No index has that expression. So PostgreSQL reads every run of the agent for each candidate, and it decompresses the large legacy `context_snapshot` of each run > - On a hosted test database, this statement ran 13 times at a mean of 11 s in a 38-minute window (149 s in total), with single runs of 16 s and 41 s > - This pull request splits the run match into its two disjoint branches. Each branch uses an existing issue index > - The benefit is that each candidate reads a few index entries instead of the agent's full run history. The returned rows do not change, and no migration is necessary ## Linked Issues or Issue Description **What existing behavior does this improve?** The discovery page query in `registerGitHubTerminalResponseWork` in `server/src/services/chat-channels/github/terminal-response-work.ts`. The query runs when the sweep reconstructs its candidates, and once or twice for each candidate check. **Subsystem affected** GitHub chat channel terminal-response recovery. **Current behavior** The `WHERE` clause has `issues.status = 'blocked' or (select r.status from heartbeat_runs r where r.company_id = ... and r.agent_id = ... and coalesce(r.native_issue_id::text, r.context_snapshot->>'issueId') = order by r.created_at desc, r.id desc limit 1) in ()`. The planner can only use `heartbeat_runs_company_agent_started_idx`. It reads all runs of the agent, then evaluates the `coalesce` on each row. That decompresses the context of each legacy run. The cost is the number of candidates times the number of agent runs. **Proposed behavior** The subquery takes the latest run from two branches and then the latest of the two: - runs with `native_issue_id = `, through `heartbeat_runs_company_native_issue_contract_id_uq` - runs with `native_issue_id is null and context_snapshot->>'issueId' = `, through `heartbeat_runs_company_ctx_issue_created_idx` The two branches are disjoint, and together they are exactly the rows where the `coalesce` is equal to the issue ID. Both branches and the outer query use the same order (`created_at desc, id desc`). So the selected run is the same, including ties. **Reason and benefit** This statement was the largest single-statement cost in the hosted window after the earlier database fixes. A local reproduction is about 490x faster on the default fixture, and the gain grows with the agent's run history. **Breaking changes** None. The API, the candidate rules and the results do not change. ## What Changed - Rewrite the latest-run status subquery in the GitHub terminal-response discovery page as two index-backed branches (native binding, and legacy context without a native binding). - Add two integration tests for the sweep. In the first, newer runs of the same agent that do not bind the issue (one with a native binding to another issue, but a legacy context that names this issue) must not hide the stopped run. In the second, a newer active run with a native binding to the issue keeps the request waiting. - Add `scripts/benchmark-github-terminal-response-sweep.ts`. It captures the SQL that the real worker generates at a baseline commit and at `HEAD`, compares all rows, and measures both. ## Verification Cause and thesis. The cost comes from the `coalesce` run match, which has no index and makes each candidate scan the agent's run history. Thesis: if each precedence branch uses its own issue index, the cost of the page no longer depends on the agent's run count, and the rows stay identical. The benchmark confirms the thesis. Local SQL benchmark: ```sh pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-github-terminal-response-sweep.ts --baseline-ref=7b33384446 ``` Fixture (synthetic): two companies, each with one GitHub endpoint and one assigned agent. Each issue has one GitHub response comment. The first N comments are acknowledged and not final; the others are final. Every tenth issue is blocked. Runs alternate between native binding and legacy context, and each run has a 4 KiB incompressible context. The fixture also has these edge cases: newer active runs with a native binding to another issue but a legacy context that names this issue, active runs that tie on `created_at` with the latest run (between native and legacy runs), and newer active runs of another agent. The benchmark compares every column of every candidate row (the page limit is lifted for this check). Embedded PostgreSQL 18.1 on an Apple laptop. The hosted database is Neon with different compute, so absolute times are local SQL measurements only. Each size has 2 warmups and then alternates the baseline and the treatment. | Size | Runs (both companies) | Candidate rows equal | Baseline median (range) | Treatment median (range) | Speedup | | --- | --- | --- | --- | --- | --- | | Small: 200 issues, 3 runs each, 20 candidates, 7 samples | 1,372 | 26 of 26 | 18.7 ms (17.7–20.8) | 3.8 ms (3.3–4.3) | 4.9x | | Default: 1,500 issues, 8 runs each, 150 candidates, 7 samples | 25,300 (156 MB) | 200 of 200 | 3,697 ms (2,875–4,032) | 7.6 ms (6.6–9.6) | 490x | | Large: 3,000 issues, 10 runs each, 400 candidates, 5 samples | 62,600 (382 MB) | 538 of 538 | 57,055 ms (48,343–66,079) | 31 ms (16–79) | 1,800x | Sizes are set with `BENCH_ISSUES`, `BENCH_RUNS_PER_ISSUE`, `BENCH_CANDIDATES` and `BENCH_ITERATIONS`. The timed query is the real first page (limit 100). Plan (default size): the baseline runs `SubPlan` 270 times. Each run is a `Bitmap Heap Scan` on `heartbeat_runs_company_agent_started_idx` that removes about 12,000 rows by filter, for about 8.2 million buffer reads in total. The treatment runs an `Append` of two `Limit` index scans (`heartbeat_runs_company_native_issue_contract_id_uq` and `heartbeat_runs_company_ctx_issue_created_idx`), for about 3,800 buffer reads in total. Tests: - `cd server && npx vitest run src/__tests__/chat-channels.integration.test.ts -t "terminal GitHub"`: 13 passed, including the 2 new tests. - Mutation check: when the legacy branch does not exclude runs with a native binding, the first new test fails. It passes with the fix. - `cd server && npx tsc --noEmit`: passed. ## Risks - Low risk. The change is one SQL subquery. The match is the same set of runs: `native_issue_id::text = ` is equal to `native_issue_id = `, and a null `native_issue_id` makes the `coalesce` use the legacy context. - The candidate set itself is not bounded by this change. Acknowledged responses that never become final stay in the candidate set. Each candidate is now cheap, but the count still grows with that history. - `readGitHubResponseRun` in `response-run-state.ts` uses the same `coalesce` match. The sweep calls it once for each candidate row, and other GitHub paths also call it. Each call is a separate, smaller statement. This pull request does not change it. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic. Used through Claude Code with tool use (shell, file edits, and local PostgreSQL benchmarks). ## 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 --- ...enchmark-github-terminal-response-sweep.ts | 192 ++++++++++++++++++ .../chat-channels.integration.test.ts | 48 +++++ .../github/terminal-response-work.ts | 21 +- 3 files changed, 257 insertions(+), 4 deletions(-) create mode 100644 scripts/benchmark-github-terminal-response-sweep.ts diff --git a/scripts/benchmark-github-terminal-response-sweep.ts b/scripts/benchmark-github-terminal-response-sweep.ts new file mode 100644 index 00000000000..a7b31f3a41d --- /dev/null +++ b/scripts/benchmark-github-terminal-response-sweep.ts @@ -0,0 +1,192 @@ +import assert from "node:assert/strict"; +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 { registerGitHubTerminalResponseWork } from "../server/src/services/chat-channels/github/terminal-response-work.js"; +import { captureReviewWaitQueries, type CapturedReviewWaitQuery } from "../server/src/__tests__/helpers/github-review-wait.js"; + +// Measures the GitHub terminal-response discovery page against a synthetic +// history: many agent runs with incompressible legacy contexts and a set of +// acknowledged (processed, non-final) response comments. +// pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-github-terminal-response-sweep.ts --baseline-ref= +const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15); +assert.ok(baselineRef, "Pass an explicit pre-fix --baseline-ref="); +const number = (name: string, fallback: number) => Number(process.env[name] ?? fallback); +const fixture = { + issues: number("BENCH_ISSUES", 1_500), + runsPerIssue: number("BENCH_RUNS_PER_ISSUE", 8), + contextPadBytes: number("BENCH_CONTEXT_PAD_BYTES", 4_096), + candidates: number("BENCH_CANDIDATES", 150), + blockedEvery: 10, + decoyCompanies: 1, +}; +const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8" }).trim(); +const treatmentRevision = execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8" }).trim(); +const modulePath = "server/src/services/chat-channels/github/terminal-response-work.ts"; +const root = fileURLToPath(new URL("../", import.meta.url)); +const keepAlive = setInterval(() => {}, 1_000); +let temporary: Awaited> | undefined; +let db: Db | undefined; +let baselineDir: string | undefined; + +async function seed(db: Db, companyIndex: number) { + const sql = db.$client; + const [{ company }] = await sql.unsafe(`insert into companies (name, issue_prefix) values ($1, $2) returning id as company`, + [`Bench ${companyIndex}`, `BN${companyIndex}`]); + const [{ agent }] = await sql.unsafe(`insert into agents (company_id, name) values ($1, 'GitHub agent') returning id as agent`, [company]); + const [{ application }] = await sql.unsafe(`insert into tool_applications (company_id, application_key, name, type) + values ($1, 'github', 'GitHub', 'chat') returning id as application`, [company]); + const [{ connection }] = await sql.unsafe(`insert into tool_connections (company_id, application_id, name, uid, transport, connection_purpose) + values ($1, $2, 'GitHub', $3, 'chat_sdk', 'channel') returning id as connection`, [company, application, `github-${companyIndex}`]); + const [{ endpoint }] = await sql.unsafe(`insert into chat_endpoints (company_id, connection_id, provider, public_id, assigned_agent_id, status) + values ($1, $2, 'github', $3, $4, 'active') returning id as endpoint`, [company, connection, `bench-${companyIndex}`, agent]); + const [{ resource }] = await sql.unsafe(`insert into chat_endpoint_resources (company_id, endpoint_id, type, provider_resource_id, label, metadata) + values ($1, $2, 'repository', 'R_1', 'paperclipai/paperclip', '{"providerRepositoryId":"R_1"}') returning id as resource`, [company, endpoint]); + // Issue numbers are 1..N; every tenth issue is blocked. + await sql.unsafe(`insert into issues (company_id, title, status) + select $1, 'Issue ' || n, case when n % ${fixture.blockedEvery} = 0 then 'blocked' else 'in_progress' end + from generate_series(1, ${fixture.issues}) n`, [company]); + await sql.unsafe(`create unlogged table if not exists bench_issue (n int, id uuid, company uuid)`); + await sql.unsafe(`delete from bench_issue`); + await sql.unsafe(`insert into bench_issue select (regexp_replace(title, '\\D', '', 'g'))::int, id, company_id from issues where company_id = $1`, [company]); + await sql.unsafe(`insert into chat_conversations (company_id, endpoint_id, issue_id, external_conversation_id, external_label, + external_thread_id, resource_id) + select $1, $2, i.id, 'github:paperclipai/paperclip:' || i.n, 'PR #' || i.n, 'github:paperclipai/paperclip:' || i.n, $3 + from bench_issue i`, [company, endpoint, resource]); + await sql.unsafe(`insert into chat_deliveries (company_id, endpoint_id, provider_event_id, deduplication_key, event_kind, + normalized_event, conversation_id, state) + select $1, $2, 'event-' || i.n, 'dedupe-' || i.n, 'message', '{}', c.id, 'processed' + from bench_issue i join chat_conversations c on c.issue_id = i.id`, [company, endpoint]); + // One response comment per issue. The first N issues are acknowledged and + // not final; the rest finished long ago with a final response. + await sql.unsafe(`insert into chat_actions (company_id, endpoint_id, delivery_id, conversation_id, kind, provider_action_id, status, result) + select $1, $2, d.id, c.id, 'github_response_comment', 'response-' || i.n, 'processed', + case when i.n <= ${fixture.candidates} then jsonb_build_object('id', 'IC_' || i.n, 'final', false) + else jsonb_build_object('id', 'IC_' || i.n, 'final', true) end + from bench_issue i join chat_conversations c on c.issue_id = i.id join chat_deliveries d on d.conversation_id = c.id`, [company, endpoint]); + await sql.unsafe(`insert into chat_message_links (company_id, endpoint_id, conversation_id, delivery_id, provider_message_id, direction) + select $1, $2, c.id, d.id, 'IC_in_' || i.n, 'inbound' + from bench_issue i join chat_conversations c on c.issue_id = i.id join chat_deliveries d on d.conversation_id = c.id`, [company, endpoint]); + // Runs: odd issues are legacy (context_snapshot only), even issues are native. + // Legacy contexts carry incompressible padding like real prompt context. + await sql.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, native_issue_id, context_snapshot, created_at) + select $1, $2, case when r = ${fixture.runsPerIssue} and i.n % 3 = 0 then 'running' else 'succeeded' end, + case when i.n % 2 = 0 then i.id end, + jsonb_build_object('issueId', i.id::text, 'pad', + (select string_agg(md5(random()::text || g::text || i.n::text || r::text), '') from generate_series(1, ${Math.ceil(fixture.contextPadBytes / 32)}) g)), + timestamptz '2026-01-01' + (i.n * ${fixture.runsPerIssue} + r) * interval '1 second' + from bench_issue i cross join generate_series(1, ${fixture.runsPerIssue}) r`, [company, agent]); + // Precedence decoys: a newer active run natively bound to another issue whose + // legacy context still names this issue. It must not count for this issue. + await sql.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, native_issue_id, context_snapshot, created_at) + select $1, $2, 'running', gen_random_uuid(), jsonb_build_object('issueId', i.id::text), + timestamptz '2026-01-01' + (i.n * ${fixture.runsPerIssue} + ${fixture.runsPerIssue} + 1) * interval '1 second' + from bench_issue i where i.n % 7 = 0`, [company, agent]); + // Ordering ties: an active run with the latest creation time of the issue's + // other runs (native and legacy). The id decides which run is latest. + await sql.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, native_issue_id, context_snapshot, created_at) + select $1, $2, 'running', case when i.n % 2 = 1 then i.id end, jsonb_build_object('issueId', i.id::text), + timestamptz '2026-01-01' + (i.n * ${fixture.runsPerIssue} + ${fixture.runsPerIssue}) * interval '1 second' + from bench_issue i where i.n % 11 = 0`, [company, agent]); + // Another agent's newer active runs on the same issues. + const [{ other }] = await sql.unsafe(`insert into agents (company_id, name) values ($1, 'Other agent') returning id as other`, [company]); + await sql.unsafe(`insert into heartbeat_runs (company_id, agent_id, status, native_issue_id, context_snapshot, created_at) + select $1, $2, 'running', case when i.n % 2 = 0 then i.id end, jsonb_build_object('issueId', i.id::text), timestamptz '2027-01-01' + from bench_issue i where i.n % 5 = 0`, [company, other]); + return { company, agent, endpoint }; +} + +try { + const oldSource = execFileSync("git", ["show", `${baselineRevision}:${modulePath}`], { encoding: "utf8" }); + const moduleUrl = new URL(`../${modulePath}`, import.meta.url); + baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.terminal-response-baseline-")); + const baselineFile = path.join(baselineDir, "terminal-response-work.ts"); + await fs.writeFile(baselineFile, oldSource.replace(/from "(\.{1,2}\/[^\"]+)"/g, + (_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, moduleUrl)))}`)); + const baseline = await import(pathToFileURL(baselineFile).href) as typeof import("../server/src/services/chat-channels/github/terminal-response-work.js"); + temporary = await startEmbeddedPostgresTestDatabase("github-terminal-response-benchmark-"); + db = createDb(temporary.connectionString); + const seedStart = performance.now(); + const main = await seed(db, 0); + for (let n = 1; n <= fixture.decoyCompanies; n++) await seed(db, n); + await db.$client.unsafe("DROP TABLE bench_issue"); + await db.$client.unsafe("ANALYZE"); + const seedMs = performance.now() - seedStart; + + const stop = new Error("captured"); + async function capture(register: typeof registerGitHubTerminalResponseWork) { + const queries: CapturedReviewWaitQuery[] = []; + const proxy = captureReviewWaitQueries(db!, queries); + let task: { run(signal: AbortSignal): Promise } | undefined; + const coordinator = { register: (_queue: unknown, value: typeof task) => { task = value; return { wake() {}, stop: async () => {} }; } }; + register(new Proxy(proxy, { get(target, key) { + const value = Reflect.get(target, key); + if (key !== "select") return value; + return (...args: unknown[]) => { + if (queries.some(query => query.sql.includes("github-publication:"))) throw stop; + return (value as (...args: unknown[]) => unknown)(...args); + }; + } }), coordinator as never, async () => {}, () => true); + await task!.run(new AbortController().signal).catch(() => {}); + const query = queries.find(query => query.sql.includes("github-publication:")); + assert.ok(query, "Capture the real discovery page query"); + return query; + } + const queries = { + baseline: await capture(baseline.registerGitHubTerminalResponseWork), + treatment: await capture(registerGitHubTerminalResponseWork), + }; + + // Compare every candidate, not only the first page: lift the page limit. + async function rows(query: CapturedReviewWaitQuery) { + assert.match(query.sql, /limit \$\d+$/); + const params = [...query.params.slice(0, -1), 1_000_000]; + return (await db!.$client.unsafe(query.sql, params as never[]).values()).map(row => row.map(value => + value === null ? null : typeof value === "object" && !(value instanceof Date) ? JSON.stringify(value) : String(value))); + } + const before = await rows(queries.baseline), after = await rows(queries.treatment); + assert.deepEqual(after, before, "Every returned column and row order must match"); + assert.ok(before.length > 0, "The fixture must return terminal candidates"); + + async function timed(query: CapturedReviewWaitQuery) { + const start = performance.now(); + await db!.$client.unsafe(query.sql, query.params as never[]).values(); + return performance.now() - start; + } + const iterations = number("BENCH_ITERATIONS", 7); + const samples = { baseline: [] as number[], treatment: [] as number[] }; + for (let n = 0; n < iterations + 2; n++) { + const versions = n % 2 ? ["treatment", "baseline"] as const : ["baseline", "treatment"] as const; + for (const version of versions) { + const elapsed = await timed(queries[version]); + if (n >= 2) samples[version].push(elapsed); + } + } + const stats = (values: number[]) => { + const sorted = [...values].sort((a, b) => a - b); + return { samplesMs: values.map(value => Math.round(value * 10) / 10), medianMs: sorted[Math.floor(sorted.length / 2)], + minMs: sorted[0], maxMs: sorted.at(-1) }; + }; + async function plan(query: CapturedReviewWaitQuery) { + return (await db!.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS) ${query.sql}`, query.params as never[]).values()).map(row => row[0]); + } + const counts = (await db.$client.unsafe(`select (select count(*) from heartbeat_runs)::int runs, + (select count(*) from chat_actions)::int actions, pg_size_pretty(pg_total_relation_size('heartbeat_runs')) runs_size`))[0]; + console.log(JSON.stringify({ baselineRevision, treatmentRevision, + treatmentDiff: execFileSync("git", ["diff", "HEAD", "--", modulePath], { cwd: root, encoding: "utf8" }), + postgresVersion: (await db.$client.unsafe("SELECT version()"))[0]!.version, + fixture: { ...fixture, ...counts, seedMs: Math.round(seedMs), mainCompany: main.company, synthetic: true }, + correctness: { completeOutputEqual: true, rows: before.length }, + timings: { warmups: 2, samples: iterations, order: "alternating", baseline: stats(samples.baseline), treatment: stats(samples.treatment) }, + plans: { baseline: await plan(queries.baseline), treatment: await plan(queries.treatment) }, + generatedQueries: { baseline: queries.baseline, treatment: queries.treatment }, + }, null, 2)); +} finally { + try { + await db?.$client.end({ timeout: 0 }); + await temporary?.cleanup(); + if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true }); + } finally { clearInterval(keepAlive); } +} diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 9fa666c38a6..909f9bd60a8 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -4194,6 +4194,54 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { await coordinator.stop(); vi.useRealTimers(); } }); + it("finds terminal GitHub responses by native issue binding before legacy run context", async () => { + const f = await toolReplyFixture(); + const issueId = f.conversation.issueId!; + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + // Newer runs of the same agent that do not bind this issue: one is natively + // bound to another issue although its legacy context names this issue, and + // one names another issue in its context. Neither may hide the stopped run. + await db.insert(heartbeatRuns).values([ + { companyId: f.companyId, agentId: f.assignedAgentId, status: "running", runtimeMode: "native", + nativeIssueId: randomUUID(), contextSnapshot: { issueId }, createdAt: new Date(Date.now() + 60_000) }, + { companyId: f.companyId, agentId: f.assignedAgentId, status: "running", + contextSnapshot: { issueId: randomUUID() }, createdAt: new Date(Date.now() + 120_000) }, + ]); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }), + { timeout: 5000 }).toEqual({ next: null, active: 0 }); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("keeps terminal GitHub responses waiting while a natively bound newer run is active", async () => { + const f = await toolReplyFixture(); + const issueId = f.conversation.issueId!; + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + // The newer run binds this issue natively; its legacy context is ignored. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId, + status: "running", runtimeMode: "native", nativeIssueId: issueId, contextSnapshot: { issueId: randomUUID() }, + createdAt: new Date(Date.now() + 60_000) }); + const baseline = f.responseWrites.length; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }), + { timeout: 5000 }).toEqual({ next: null, active: 0 }); + expect(f.responseWrites).toHaveLength(baseline); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), + eq(chatActions.kind, "github_response_comment"))); + expect(action).toMatchObject({ status: "processed" }); + expect(action.result?.final).not.toBe(true); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + it("settles a deleted Board-origin request without starving unrelated terminal GitHub replies", async () => { const invalid = await toolReplyFixture(); const valid = await toolReplyFixture(); diff --git a/server/src/services/chat-channels/github/terminal-response-work.ts b/server/src/services/chat-channels/github/terminal-response-work.ts index c43910d550e..cb1f240bb9e 100644 --- a/server/src/services/chat-channels/github/terminal-response-work.ts +++ b/server/src/services/chat-channels/github/terminal-response-work.ts @@ -21,6 +21,22 @@ export function registerGitHubTerminalResponseWork( const repositoryId = sql`${chatEndpointResources.metadata}->>'providerRepositoryId'`; const number = sql`substring(${chatConversations.externalThreadId} from '^github:[^:]+:(?:issue:)?([1-9][0-9]*)(?::rc:[1-9][0-9]*)?$')`; + // Status of the assigned agent's latest run bound to the issue, where a run's + // issue is coalesce(native_issue_id, context_snapshot->>'issueId'). The two + // precedence branches are disjoint and each uses an issue index, so a + // candidate does not walk (and detoast) the agent's whole run history. + const latestIssueRunStatus = sql`select latest.status from ( + (select r.status, r.created_at, r.id from ${heartbeatRuns} r + where r.company_id = ${chatActions.companyId} and r.native_issue_id = ${chatConversations.issueId} + and r.agent_id = ${chatEndpoints.assignedAgentId} + order by r.created_at desc, r.id desc limit 1) + union all + (select r.status, r.created_at, r.id from ${heartbeatRuns} r + where r.company_id = ${chatActions.companyId} and r.native_issue_id is null + and r.context_snapshot->>'issueId' = ${chatConversations.issueId}::text + and r.agent_id = ${chatEndpoints.assignedAgentId} + order by r.created_at desc, r.id desc limit 1) + ) latest order by latest.created_at desc, latest.id desc limit 1`; async function page(scope?: Scope, cursor?: string, actionId?: string) { return db.select({ id: chatActions.id, companyId: chatActions.companyId, endpointId: chatActions.endpointId, @@ -42,10 +58,7 @@ export function registerGitHubTerminalResponseWork( .where(and(eq(chatActions.kind, "github_response_comment"), eq(chatActions.status, "processed"), isNotNull(chatActions.deliveryId), isNotNull(chatEndpoints.assignedAgentId), sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`, - sql`(${issues.status} = 'blocked' or (select r.status from ${heartbeatRuns} r - where r.company_id = ${chatActions.companyId} and r.agent_id = ${chatEndpoints.assignedAgentId} - and coalesce(r.native_issue_id::text, r.context_snapshot->>'issueId') = ${chatConversations.issueId}::text - order by r.created_at desc, r.id desc limit 1) in ('succeeded', 'failed', 'interrupted', 'timed_out', 'cancelled'))`, + sql`(${issues.status} = 'blocked' or (${latestIssueRunStatus}) in ('succeeded', 'failed', 'interrupted', 'timed_out', 'cancelled'))`, scope ? eq(chatActions.companyId, scope.companyId) : undefined, scope?.issueId ? eq(issues.id, scope.issueId) : undefined, scope?.conversationId ? eq(chatActions.conversationId, scope.conversationId) : undefined,