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 <noreply@paperclip.ing>

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.

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<Record<string, unknown>> = [], 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" } } })

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(

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,

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<Record<string, unknown>> = [], 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);

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<Record<string, unknown> | null>`case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object'
+    then jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId') else null end`,
+  runnerProfileJson: sql<Record<string, unknown> | 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()