perf(wake-queue): narrow native release reads (#17539)

Native wake-queue release reads omit saved profiles and excerpts. Legacy release reads keep their complete inputs. Ownership, cleanup, and file retention checks remain unchanged.

Verified with 275 wake-queue tests, the full build, and typecheck. All CI checks pass. Greptile scored the exact head 5/5 with Medium impact and no unresolved comments.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/modules/wake-queue/adapters/postgres.test.ts

@@ -206,7 +206,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
     return { recorded, rows };
   }
 
-  it("keeps in-flight work and every recovery receipt while omitting output from release reads", async () => {
+  it.each(["legacy", "native"])("keeps %s in-flight work and every recovery receipt while omitting output from release reads", async runtimeMode => {
     const companyId = await seedCompany();
     const agentId = await seedAgent({ companyId });
     const issueId = await seedIssue({ companyId, assigneeAgentId: agentId });
@@ -218,7 +218,9 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
       nativeCancellation: { schema: "paperclip.native-cancellation.v1", runId, companyId, issueId },
       executionCancellation: { state: "acknowledged" }, executionRecovery: { kind: "interrupted", providerStopped: true },
       configurationIncomplete: { reason: "workspace_unavailable" }, unknownFutureReceipt: { retained: true } };
-    await db.update(heartbeatRuns).set({ resultJson: { ...evidence, stdout: payload.repeat(3), stderr: payload } }).where(eq(heartbeatRuns.id, runId));
+    await db.update(heartbeatRuns).set({ runtimeMode,
+      ...(runtimeMode === "native" ? { runnerProfileJson: { savedInput: payload }, stdoutExcerpt: payload, stderrExcerpt: payload } : {}),
+      resultJson: { ...evidence, stdout: payload.repeat(3), stderr: payload } }).where(eq(heartbeatRuns.id, runId));
     await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId));
     const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
     const [issue] = await db.select().from(issues).where(eq(issues.id, issueId));
@@ -230,7 +232,8 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
     );
     expect(result.outcome.kind).toBe("released");
     expect(rows).toHaveLength(1);
-    expect(rows[0]).toEqual([{ ...run, resultJson: evidence }]);
+    expect(rows[0]).toEqual([{ ...run, resultJson: evidence,
+      ...(runtimeMode === "native" ? { runnerProfileJson: null, stdoutExcerpt: null, stderrExcerpt: null } : {}) }]);
     expect(Buffer.byteLength(JSON.stringify(rows))).toBeLessThan(8 * 1024);
     expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]).toEqual(run);
     expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toEqual(issue);

server/src/modules/wake-queue/adapters/postgres.ts

@@ -5,7 +5,8 @@ import { isCompletedOnboardingHandoffWake } from "../../../services/chat-complet
 import { instanceSettingsService } from "../../../services/instance-settings.js";
 import { currentConversationCommentCondition } from "../../../services/agent-conversations.js";
 import { getExecutionBlocker, hasPendingRunCleanup } from "../../../services/execution-blocker.js";
-import { and, asc, eq, getTableColumns, inArray, isNull, notInArray, or, sql } from "drizzle-orm";
+import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm";
+import { readReleaseRun, type ReleaseRun } from "./release-run-read.js";
 import type { Db } from "@paperclipai/db";
 import { extractIssueReferenceIdentifiers, isUuidLike } from "@paperclipai/shared";
 import {
@@ -74,14 +75,6 @@ const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_r
 type HeartbeatRunRow = typeof heartbeatRuns.$inferSelect;
 type IssueRow = typeof issues.$inferSelect;
 
-const releaseRunFields = {
-  ...getTableColumns(heartbeatRuns),
-  resultJson: sql<Record<string, unknown> | null>`case
-    when jsonb_typeof(${heartbeatRuns.resultJson}) = 'object'
-      then ${heartbeatRuns.resultJson} - 'stdout' - 'stderr'
-    else ${heartbeatRuns.resultJson} end`,
-};
-
 function normalizeAgentNameKey(value: string | null | undefined): string | null {
   if (typeof value !== "string") return null;
   const normalized = value.trim().toLowerCase();
@@ -94,7 +87,7 @@ function toRequestedByActorType(value: string | null): "user" | "agent" | "syste
   return value === "user" || value === "agent" || value === "system" ? value : null;
 }
 
-function toRunSnapshot(row: HeartbeatRunRow): RunSnapshot {
+function toRunSnapshot(row: ReleaseRun): RunSnapshot {
   const configurationIncompletePayload = parseObject(parseObject(row.resultJson).configurationIncomplete);
   return {
     id: row.id,
@@ -206,7 +199,7 @@ function buildHost(_tx: Db, deps: WakeQueuePostgresAdapterDeps): WakeQueueHost {
   };
 }
 
-function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, run: HeartbeatRunRow): WakeQueueTransaction {
+function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, run: ReleaseRun): WakeQueueTransaction {
   const treeControlSvc = issueTreeControlService(tx);
   const issuesSvc = issueService(tx);
   const interruptQueueId = run.runtimeMode !== "native" && run.status === "cancelled"
@@ -750,7 +743,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
   };
 }
 
-async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: HeartbeatRunRow, issue: IssueRow, now: Date): Promise<boolean> {
+async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: ReleaseRun, issue: IssueRow, now: Date): Promise<boolean> {
   const applies =
     run.runtimeMode === "native" &&
     ["failed", "timed_out", "interrupted", "cancelled"].includes(run.status) &&
@@ -1058,11 +1051,8 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd
     async withIssueExecutionLock(input, fn): Promise<ReleaseTransactionResult & { run: RunSnapshot }> {
       return db.transaction(async (rawTx) => {
         const tx = rawTx as unknown as Db;
-        const run = await tx
-          .select(releaseRunFields)
-          .from(heartbeatRuns)
-          .where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId)))
-          .then((rows) => rows[0] ?? null);
+        const evidence = await readReleaseRun(tx, input);
+        const run = evidence.run;
         if (!run) {
           throw new Error(`wake-queue: run ${input.runId} was not found while releasing issue execution`);
         }
@@ -1164,7 +1154,8 @@ export function createPostgresWakeQueueAdapter(db: Db, deps: WakeQueuePostgresAd
           issueStatus: issueRow?.status ?? "",
           hasAssigneeUser: Boolean(issueRow?.assigneeUserId),
           assigneeAgentMatchesRunAgent: issueRow?.assigneeAgentId === run.agentId,
-          legacyExecutionNeedsReconciliation: await legacyExecutionNeedsReconciliationWithEvidence(tx as unknown as Db, run),
+          legacyExecutionNeedsReconciliation: evidence.kind === "complete" &&
+            await legacyExecutionNeedsReconciliationWithEvidence(tx as unknown as Db, evidence.run),
           // An operator stop never promotes old queued work by itself. The
           // next explicit wake adopts those messages atomically when it
           // queues a run.

server/src/modules/wake-queue/adapters/release-run-read.test.ts

@@ -0,0 +1,74 @@
+import { randomBytes, randomUUID } from "node:crypto";
+import { agents, companies, createDb, heartbeatRuns } from "@paperclipai/db";
+import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture";
+import { eq } from "drizzle-orm";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js";
+import { readReleaseRun } from "./release-run-read.js";
+
+const support = await getEmbeddedPostgresTestSupport();
+
+describe.skipIf(!support.supported)("wake release run reads", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  let companyId: string;
+  let agentId: string;
+
+  beforeAll(async () => {
+    database = await startEmbeddedPostgresTestDatabase("wake-release-read-");
+    db = createDb(database.connectionString);
+    const [company] = await db.insert(companies).values({ name: "Wake release fixture", issuePrefix: "WRR" }).returning();
+    const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Release agent", adapterType: "process" }).returning();
+    companyId = company.id;
+    agentId = agent.id;
+  }, 60_000);
+
+  afterAll(async () => {
+    await db?.$client.end();
+    await database?.cleanup();
+  });
+
+  it("omits native bulk at the driver boundary and preserves the complete legacy fallback", async () => {
+    const bulk = randomBytes(128 * 1024).toString("base64");
+    const context = { issueId: randomUUID(), opaqueFutureContext: { savedInput: "keep this input" }, explicitUserContinuation: true };
+    const evidence = { nativeCancellation: { schema: "paperclip.native-cancellation.v1", acknowledged: true }, unknownFutureReceipt: [true, null, 42] };
+    const [before] = await db.insert(heartbeatRuns).values({
+      companyId, agentId, runtimeMode: "native", status: "running", contextSnapshot: context,
+      runnerProfileJson: { nativeExecutionInput: { savedInput: bulk } }, stdoutExcerpt: bulk, stderrExcerpt: bulk,
+      resultJson: { ...evidence, stdout: bulk, stderr: bulk },
+    }).returning();
+    const scope = { companyId, runId: before.id };
+    const events: Array<Record<string, unknown>> = [];
+    const now = Date.now();
+    const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event));
+    installDatabaseResponseCapture(db, { heartbeatRuns }, capture);
+    try {
+      const native = await db.transaction(tx => readReleaseRun(tx as unknown as typeof db, scope));
+      expect(native.kind).toBe("native");
+      expect(native.run?.contextSnapshot).toEqual(context);
+      expect(native.run?.resultJson).toEqual(evidence);
+      expect(native.run).not.toHaveProperty("runnerProfileJson");
+      expect(native.run).not.toHaveProperty("stdoutExcerpt");
+      expect(native.run).not.toHaveProperty("stderrExcerpt");
+    } finally { capture.flush(); }
+    const summary = events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!;
+    expect(summary.completedQueries).toBe(1);
+    expect(summary.responseValueBytes).toBeLessThan(8192);
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, before.id)))[0]).toEqual(before);
+    expect(JSON.stringify(events)).not.toContain(bulk.slice(0, 64));
+    await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, before.id));
+    const legacy = await readReleaseRun(db, scope);
+    expect(legacy.kind).toBe("complete");
+    expect(legacy.run).toEqual({ ...before, runtimeMode: "legacy", resultJson: evidence });
+    expect((await readReleaseRun(db, { ...scope, companyId: randomUUID() })).run).toBeUndefined();
+    expect((await readReleaseRun(db, { companyId, runId: randomUUID() })).run).toBeUndefined();
+  });
+
+  it.each([null, ["stdout", { stderr: "evidence" }], "scalar", 42].map(result => ({ result })))(
+    "preserves non-object native result evidence $result", async ({ result }) => {
+      const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, runtimeMode: "native",
+        resultJson: result as Record<string, unknown> | null }).returning();
+      expect((await readReleaseRun(db, { companyId, runId: run.id })).run?.resultJson).toEqual(result);
+    },
+  );
+});

server/src/modules/wake-queue/adapters/release-run-read.ts

@@ -0,0 +1,22 @@
+import { heartbeatRuns, type Db } from "@paperclipai/db";
+import { and, eq, getTableColumns, sql } from "drizzle-orm";
+
+type Run = typeof heartbeatRuns.$inferSelect;
+export type ReleaseRun = Omit<Run, "runnerProfileJson" | "stdoutExcerpt" | "stderrExcerpt">;
+
+export async function readReleaseRun(db: Db, scope: { companyId: string; runId: string }) {
+  const [selected] = await db.select({
+    ...getTableColumns(heartbeatRuns),
+    runnerProfileJson: sql<Run["runnerProfileJson"]>`case when ${heartbeatRuns.runtimeMode} = 'native'
+      then null else ${heartbeatRuns.runnerProfileJson} end`,
+    stdoutExcerpt: sql<Run["stdoutExcerpt"]>`case when ${heartbeatRuns.runtimeMode} = 'native'
+      then null else ${heartbeatRuns.stdoutExcerpt} end`,
+    stderrExcerpt: sql<Run["stderrExcerpt"]>`case when ${heartbeatRuns.runtimeMode} = 'native'
+      then null else ${heartbeatRuns.stderrExcerpt} end`,
+    resultJson: sql<Run["resultJson"]>`case when jsonb_typeof(${heartbeatRuns.resultJson}) = 'object'
+      then ${heartbeatRuns.resultJson} - 'stdout' - 'stderr' else ${heartbeatRuns.resultJson} end`,
+  }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, scope.runId), eq(heartbeatRuns.companyId, scope.companyId)));
+  if (!selected || selected.runtimeMode !== "native") return { kind: "complete" as const, run: selected };
+  const { runnerProfileJson: _profile, stdoutExcerpt: _stdout, stderrExcerpt: _stderr, ...run } = selected;
+  return { kind: "native" as const, run };
+}