From 1ad8f25c7793e4e1d94c1afd29942d05d7778d87 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 24 Oct 2026 23:53:39 -0700 Subject: [PATCH] perf(heartbeat): narrow queued comment polling run reads (#17560) Read current control facts and typed issue/adapter bindings for queued-comment polling. Keep full mutation and execution reads, native preparation and steering, authorization, and stored evidence. Validation: full build and typecheck, 160 focused tests plus two assembled route regressions; exact-head Greptile 5/5 (medium), full green CI, zero unresolved threads. Old-route counterfactual transfers 2.5 MB versus less than 2 KB. Co-Authored-By: Paperclip --- doc/DEVELOPING.md | 2 + .../issue-queued-comments-routes.test.ts | 32 +++++++++- server/src/routes/issues.ts | 14 +++-- server/src/services/heartbeat.ts | 6 ++ .../src/services/heartbeat/run-state.test.ts | 59 +++++++++++++++++++ server/src/services/heartbeat/run-state.ts | 25 ++++++++ 6 files changed, 130 insertions(+), 8 deletions(-) diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index e399790fbbb..432eb547d45 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1812,6 +1812,8 @@ It owns bounded run projections, database encoding checks, task session reads an writes, explicit resumes, session compaction, and usage/billing helpers. `createHeartbeatRunState(db)` binds these operations without doing database work during construction. The encoding-check cache belongs to each factory instance. +Queued-comment polling reads current issue control fields, the typed issue binding, +and the selected adapter type. Mutation and execution paths retain their full run reads. `heartbeat.ts` keeps run execution and re-exports the existing public helpers. Keep session policy changes separate from run orchestration changes. diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index bb42982787b..60985954bea 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -4,6 +4,7 @@ import { randomUUID } from "node:crypto"; import express from "express"; import request from "supertest"; import { eq, sql } from "drizzle-orm"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { agentRuntimeState, @@ -79,7 +80,7 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { await tempDb?.cleanup(); }); - function app(companyId: string, userId = "queue-owner", agentActor?: { agentId: string; runId: string }) { + function app(companyId: string, userId = "queue-owner", agentActor?: { agentId: string; runId: string }, database = db) { const testApp = express(); testApp.use(express.json()); testApp.use((req, _res, next) => { @@ -94,7 +95,7 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { }; next(); }); - testApp.use("/api", issueRoutes(db, {} as any, {})); + testApp.use("/api", issueRoutes(database, {} as any, {})); testApp.use(errorHandler); return testApp; } @@ -192,6 +193,33 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { return { companyId, agentId, issueId, runId, wakeId, commentIds }; } + it.each([false, true])("polls queued comments without transferring run snapshots (agent fallback: %s)", async fallback => { + const f = await seedQueue(), payload = "saved-execution-input🙂".repeat(25_000); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, prompt: payload }, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner", adapterConfig: { instructions: payload } } }, + stdoutExcerpt: payload, stderrExcerpt: payload, + }).where(eq(heartbeatRuns.id, f.runId)); + if (fallback) await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, f.issueId)); + const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const observed = createDb(tempDb!.connectionString), events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(observed, { heartbeatRuns }, capture); + try { + const response = await request(app(f.companyId, "queue-owner", undefined, observed)) + .get(`/api/issues/${f.issueId}/queued-comments`).expect(200); + expect(response.body.queueId).toBe(f.wakeId); + expect(response.body.targetRunId).toBe(f.runId); + expect(response.body.entries.map((entry: { comment: { id: string } }) => entry.comment.id)).toEqual(f.commentIds); + capture.flush(); + const reads = events.filter(event => event.table === "heartbeat_runs" && event.operation === "select"); + expect(reads.length).toBeGreaterThan(0); + expect(reads.reduce((bytes, event) => bytes + Number(event.responseValueBytes), 0)).toBeLessThan(2048); + await request(app(randomUUID(), "queue-owner", undefined, observed)) + .get(`/api/issues/${f.issueId}/queued-comments`).expect(404); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(before); + } finally { capture.flush(); await observed.$client.end(); } + }); + it.each(["cancelled", "running"])("rejects a late Done from an interrupted %s task run at the write boundary", async status => { const seeded = await seedQueue(); await db.update(heartbeatRuns).set({ status, resultJson: { executionCancellation: { state: "requested" } } }) diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 9a29f1c9d91..6aa65dfb774 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -7247,18 +7247,20 @@ export function issueRoutes( id: string; assigneeAgentId: string | null; executionRunId?: string | null; - }) { + }, options?: { queueStateOnly: true }) { let runToInterrupt = issue.executionRunId - ? await heartbeat.getRun(issue.executionRunId) + ? options?.queueStateOnly + ? await heartbeat.getRunIssueControl(issue.executionRunId) + : await heartbeat.getRun(issue.executionRunId) : null; if ( (!runToInterrupt || runToInterrupt.status !== "running") && issue.assigneeAgentId ) { - const activeRun = await heartbeat.getActiveRunForAgent( - issue.assigneeAgentId, - ); + const activeRun = options?.queueStateOnly + ? await heartbeat.getActiveRunIssueControlForAgent(issue.assigneeAgentId) + : await heartbeat.getActiveRunForAgent(issue.assigneeAgentId); const activeIssueId = activeRun && activeRun.contextSnapshot && @@ -16343,7 +16345,7 @@ export function issueRoutes( const queue = await buildQueuedCommentQueue({ executor: db, issue, - activeRun: await resolveActiveIssueRun(issue), + activeRun: await resolveActiveIssueRun(issue, { queueStateOnly: true }), actor: getActorInfo(req), }); res.json( diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 9a18721aaa7..f177921a36d 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -832,6 +832,8 @@ export function heartbeatService( listEvents, getRetryExhaustedReason, getRunIssueSummary, + getRunIssueControl, + getActiveRunIssueControlForAgent, getActiveRunForAgent, getActiveRunIssueSummaryForAgent, } = createHeartbeatRunState(db); @@ -4153,6 +4155,10 @@ export function heartbeatService( getRunIssueSummary, + getRunIssueControl, + + getActiveRunIssueControlForAgent, + getActiveRunForAgent, getActiveRunIssueSummaryForAgent, diff --git a/server/src/services/heartbeat/run-state.test.ts b/server/src/services/heartbeat/run-state.test.ts index ccc91735df4..e42d93d7286 100644 --- a/server/src/services/heartbeat/run-state.test.ts +++ b/server/src/services/heartbeat/run-state.test.ts @@ -20,6 +20,7 @@ import { import * as legacy from "../heartbeat.js"; import * as extracted from "./run-state.js"; import { readRunCancellation } from "../run-cancellation.js"; +import { decideQueuedCommentQueueSteering } from "../issue-queued-comment-queue.js"; // SQL_ASCII needs a separate database cluster. Inspect its selected SQL here; // the PostgreSQL tests below exercise the session queries and transactions. @@ -158,6 +159,64 @@ describePostgres("heartbeat run state database wiring", () => { return { companyId, agent, issue, state: extracted.createHeartbeatRunState(db) }; } + it("reads fresh issue control facts without transferring execution snapshots", async () => { + const { companyId, agent, issue } = await fixture(); + const payload = randomBytes(128 * 1024).toString("base64"); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, status: "running", + runtimeMode: "legacy", runtimeModeResolvedAt: null, + startedAt: new Date("2020-01-01T00:00:00Z"), + contextSnapshot: { issueId: issue.id, prompt: payload }, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner", adapterConfig: { prompt: payload } }, savedInput: payload }, + resultJson: { stdout: payload }, stdoutExcerpt: payload, stderrExcerpt: payload, + }).returning(); + const observed = createDb(database.connectionString); + const events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(observed, { heartbeatRuns }, capture); + try { + const state = extracted.createHeartbeatRunState(observed); + const direct = await state.getRunIssueControl(run.id), active = await state.getActiveRunIssueControlForAgent(agent.id); + expect(active).toEqual(direct); + expect(direct).toEqual({ id: run.id, companyId, agentId: agent.id, status: "running", runtimeMode: "legacy", + runtimeModeResolvedAt: null, startedAt: run.startedAt, createdAt: run.createdAt, + contextSnapshot: { issueId: issue.id }, runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } } }); + capture.flush(); + const summary = events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!; + expect(summary.completedQueries).toBe(2); + expect(summary.responseValueBytes).toBeLessThan(2048); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toEqual(run); + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, run.id)); + expect((await state.getRunIssueControl(run.id))?.status).toBe("failed"); + expect(await state.getActiveRunIssueControlForAgent(agent.id)).toBeNull(); + expect(await state.getActiveRunIssueControlForAgent(randomUUID())).toBeNull(); + expect(await state.getRunIssueControl(randomUUID())).toBeNull(); + } finally { capture.flush(); await observed.$client.end(); } + }); + + it("preserves native preparation, legacy steering, and typed issue context in issue control reads", async () => { + const { companyId, agent, issue, state } = await fixture(); + for (const mode of ["native", "legacy"] as const) { + for (const value of [null, false, 42, [], {}, { nested: true }, "paperclip_runner", "process"]) { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, status: "running", runtimeMode: mode, + contextSnapshot: { issueId: value }, runnerProfileJson: { adapterDispatch: { adapterType: value } }, + }).returning(); + const narrow = await state.getRunIssueControl(run.id), full = await state.getRun(run.id); + expect(narrow?.contextSnapshot?.issueId).toEqual(value); + expect(narrow?.runnerProfileJson).toEqual({ adapterDispatch: { adapterType: value } }); + for (const queueState of ["queued", "deferred", null] as const) { + const facts = { state: queueState, queueRunRuntimeMode: mode, assignedAgentAdapterType: "paperclip_runner", queuedCommentCount: 1 }; + expect(decideQueuedCommentQueueSteering({ ...facts, activeRun: narrow })) + .toEqual(decideQueuedCommentQueueSteering({ ...facts, activeRun: full })); + } + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: issue.id }, runtimeModeResolvedAt: new Date("2020-01-01T00:00:00Z") }) + .where(eq(heartbeatRuns.id, run.id)); + const updated = await state.getRunIssueControl(run.id); + expect(updated?.contextSnapshot).toEqual({ issueId: issue.id }); + expect(updated?.runtimeModeResolvedAt).toEqual(new Date("2020-01-01T00:00:00Z")); + } + } + }); + it("reads cleanup and workspace settlement without transferring saved input or changing the run", async () => { const { companyId, agent, issue, state } = await fixture(); const bulk = "cleanup-saved-input-sentinel🙂".repeat(40_000); diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts index 1911ad1a25c..3bf92684a1f 100644 --- a/server/src/services/heartbeat/run-state.ts +++ b/server/src/services/heartbeat/run-state.ts @@ -354,6 +354,18 @@ const heartbeatRunLogAccessColumns = { logRef: heartbeatRuns.logRef, } as const; +const heartbeatRunIssueControlColumns = { + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode, + runtimeModeResolvedAt: heartbeatRuns.runtimeModeResolvedAt, + startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, + contextSnapshot: sql | null>`case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object' + then jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId') else null end`, + runnerProfileJson: sql | null>`case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' + then jsonb_build_object('adapterDispatch', jsonb_build_object('adapterType', ${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->'adapterType')) + else null end`, +} as const; + const heartbeatRunIssueSummaryColumns = { id: heartbeatRuns.id, runtimeMode: heartbeatRuns.runtimeMode, @@ -1955,6 +1967,19 @@ export function createHeartbeatRunState(db: Db) { return run ?? null; }, + getRunIssueControl: async (runId: string) => { + const [run] = await db.select(heartbeatRunIssueControlColumns).from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)).limit(1); + return run ?? null; + }, + + getActiveRunIssueControlForAgent: async (agentId: string) => { + const [run] = await db.select(heartbeatRunIssueControlColumns).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.agentId, agentId), eq(heartbeatRuns.status, "running"))) + .orderBy(desc(heartbeatRuns.startedAt)).limit(1); + return run ?? null; + }, + getActiveRunForAgent: async (agentId: string) => { const [run] = await db .select()