perf(recovery): narrow saved-message execution-wait reads (#17588)

Read current ownership, startup, cancellation, task-binding, and adapter evidence without transferring unused saved profiles, launch context, or output. Preserve normal continuation admission, recovery holds, and locked mutation reads.

Verified with full build and typecheck, 537 affected tests across 12 suites, a counterfactual real PostgreSQL byte regression, complete exact-head CI, and Greptile 5/5 Medium with no unresolved comments.

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

doc/DATABASE.md

@@ -140,6 +140,11 @@ Historical root-reader authorization has been retired; no retained-reader batch
 loaded at admission or publication. Workspace-record disclosure still uses task
 privacy checks, independently of filesystem access.
 
+Saved-message execution-wait scans read current run state, process ownership,
+startup and cancellation receipts, task binding, and the recorded adapter kind.
+Saved profiles, output excerpts, and unrelated launch context stay in storage.
+Continuation admission and its locked mutation reads retain their existing gates.
+
 The initial agent JWT run lookup reads only the cancellation state and
 conversation flag from saved run JSON. It does not transfer the full context or
 result. Run status, responsible-user identity, company and agent boundaries,

server/src/services/heartbeat.ts

@@ -812,6 +812,7 @@ export function heartbeatService(
     resolveSessionBeforeForWakeup,
     getRun,
     getRunCleanupState,
+    getRunExecutionWait,
     getRunWorkspaceSettlement,
     getRunStatus,
     ensureRuntimeState,
@@ -1393,6 +1394,7 @@ export function heartbeatService(
   const runControl = createHeartbeatRunControl(db, {
     getRun,
     getRunCleanupState,
+    getRunExecutionWait,
     options,
     envOrchestrator,
     activeRunExecutions,

server/src/services/heartbeat/run-control.ts

@@ -165,7 +165,7 @@ async function withBudgetStopDelivery<T>(db: Db, work: () => Promise<T>): Promis
 
 /** Cancellation shares executor ownership and delegates lifecycle and admission effects. */
 export interface HeartbeatRunControlDependencies extends Pick<ReturnType<typeof createHeartbeatRunState>,
-  "getRun" | "getAgent" | "getRunCleanupState">,
+  "getRun" | "getAgent" | "getRunCleanupState" | "getRunExecutionWait">,
   Pick<ReturnType<typeof createHeartbeatLifecycle>,
     "mergeRunStopMetadataForAgent" | "setRunStatusFromLive" | "setRunStatus" | "setWakeupStatus" | "appendRunEvent">,
   Pick<HeartbeatRetryDependencies, "finalizeAgentStatus"> {
@@ -326,6 +326,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
   const {
     getRun,
     getRunCleanupState,
+    getRunExecutionWait,
     options,
     envOrchestrator,
     activeRunExecutions,
@@ -509,7 +510,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
     }
   }
 
-  async function resumeRemoteStopComments(run: typeof heartbeatRuns.$inferSelect, requestId?: string) {
+  async function resumeRemoteStopComments(run: NonNullable<Awaited<ReturnType<typeof getRunExecutionWait>>>, requestId?: string) {
     if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id)) return;
     const [preparationCoordinator] = run.runtimeMode === "legacy" && run.status === "cancelled" && !run.runtimeModeResolvedAt
       ? await db.select().from(nativeRunFinalizations).where(and(
@@ -536,7 +537,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
       await acknowledgedNativeStopExecutionHasStopped(db, currentRun) &&
       !(await getExecutionBlocker(db, run.companyId, issueId));
     const legacyContinuation = run.runtimeMode === "legacy" &&
-      hasConversationContinuationPolicy((await getRun(run.id, { includeExecutionEvidence: true }))?.resultJson) &&
+      hasConversationContinuationPolicy((await getRunExecutionWait(run.companyId, run.id))?.resultJson) &&
       !(await getExecutionBlocker(db, run.companyId, issueId));
     if (run.runtimeMode !== "native" && run.runtimeMode !== "legacy") return;
     const pending = await db.select().from(agentWakeupRequests).where(and(
@@ -803,7 +804,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
       const blocker = await getExecutionBlocker(db, wake.companyId, issueId);
       const sourceId = blocker?.runId;
       if (!sourceId || !isUuidLike(sourceId)) continue;
-      const run = await getRun(sourceId, { includeExecutionEvidence: true });
+      const run = await getRunExecutionWait(wake.companyId, sourceId);
       if (!run || run.companyId !== wake.companyId || run.agentId !== wake.agentId) continue;
       if (canContinueCancelledRun(run)) {
         await resumeSavedLegacyComments(wake.companyId, wake.id).catch(err => {

server/src/services/heartbeat/run-state.test.ts

@@ -97,6 +97,17 @@ describe("heartbeat run state module", () => {
     expect(raw.columns[0].resultJson).toBe(heartbeatRuns.resultJson);
   });
 
+  it.each(["UTF8", "SQL_ASCII"])("keeps execution-wait reads company scoped with %s text handling", async encoding => {
+    const reader = readingDatabase(encoding), state = extracted.createHeartbeatRunState(reader.db);
+    expect(await state.getRunExecutionWait("company", "run")).toBeNull();
+    expect(await state.getRunExecutionWait("other-company", "run")).toBeNull();
+    expect(reader.scopes).toEqual([["company", "run"], ["other-company", "run"]]);
+    expect(reader.execute).toHaveBeenCalledTimes(1);
+    if (encoding === "UTF8") expect(reader.columns[0].error).toBe(heartbeatRuns.error);
+    else expect(new PgDialect().sqlToQuery((reader.columns[0].error as SQL.Aliased).sql).sql).toBe("NULL");
+    expect(new PgDialect().sqlToQuery((reader.columns[0].resultJson as SQL.Aliased).sql).sql).toContain("startupPreparationSettledAt");
+  });
+
   it("loads current run status without large fields or an encoding probe", async () => {
     const binding = { id: "run", companyId: "company", agentId: "agent", status: "running" };
     const reader = readingDatabase("SQL_ASCII", [binding], []);
@@ -159,6 +170,56 @@ describePostgres("heartbeat run state database wiring", () => {
     return { companyId, agent, issue, state: extracted.createHeartbeatRunState(db) };
   }
 
+  it("reads current execution-wait evidence without transferring saved input or output", async () => {
+    const { companyId, agent, issue } = await fixture();
+    const payload = randomBytes(128 * 1024).toString("base64");
+    const context = { issueId: issue.id, connectionIntentResolution: "existing_connection", source: "connection_intent.resolved",
+      interactionKind: "connection_intent", interactionStatus: "expired", mutation: "interaction", interactionId: randomUUID() };
+    const result = { startupCancellation: { beforeNativeSelection: true, futureProof: { version: 2 } },
+      startupPreparationSettledAt: "2020-01-01T00:00:00Z", cancellation: { source: "provider", expected: false,
+        initiator: { type: "provider" }, reason: "Cancelled", recordedAt: "2020-01-01T00:00:00Z" },
+      acpToolInventoryComplete: true, acpPendingToolCount: 0, stdout: payload };
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, status: "cancelled", runtimeMode: "legacy",
+      startedAt: new Date("2020-01-01T00:00:00Z"), finishedAt: new Date("2020-01-01T00:01:00Z"),
+      contextSnapshot: { ...context, prompt: payload }, resultJson: result,
+      runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner", adapterConfig: { prompt: payload } }, savedInput: payload },
+      stdoutExcerpt: payload, stderrExcerpt: payload,
+    }).returning();
+    const observed = createDb(database.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 state = extracted.createHeartbeatRunState(observed);
+      const read = await state.getRunExecutionWait(companyId, run.id);
+      capture.flush();
+      const summary = events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!;
+      expect(summary.completedQueries).toBe(2);
+      expect(summary.responseValueBytes).toBeLessThan(4096);
+      expect(read).toMatchObject({ id: run.id, companyId, agentId: agent.id, status: "cancelled", contextSnapshot: context,
+        runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
+        resultJson: { startupCancellation: result.startupCancellation, cancellation: result.cancellation,
+          acpToolInventoryComplete: true, acpPendingToolCount: 0 } });
+      expect(JSON.stringify(read)).not.toContain(payload.slice(0, 64));
+      expect(read).not.toHaveProperty("stdoutExcerpt");
+      expect(read).not.toHaveProperty("stderrExcerpt");
+      expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toEqual(run);
+      await db.update(heartbeatRuns).set({ status: "running", resultJson: { ...result, acpPendingToolCount: 1 },
+        runnerProfileJson: { adapterDispatch: { adapterType: "process" }, savedInput: payload },
+      }).where(eq(heartbeatRuns.id, run.id));
+      expect(await state.getRunExecutionWait(companyId, run.id)).toMatchObject({ status: "running",
+        resultJson: { acpPendingToolCount: 1 }, runnerProfileJson: { adapterDispatch: { adapterType: "process" } } });
+      expect(await state.getRunExecutionWait(randomUUID(), run.id)).toBeNull();
+      expect(await state.getRunExecutionWait(companyId, randomUUID())).toBeNull();
+      for (const value of [null, false, 42, [], {}, { nested: true }, "task"]) {
+        await db.update(heartbeatRuns).set({ contextSnapshot: { ...context, issueId: value },
+          runnerProfileJson: { adapterDispatch: { adapterType: value } },
+        }).where(eq(heartbeatRuns.id, run.id));
+        expect(await state.getRunExecutionWait(companyId, run.id)).toMatchObject({ contextSnapshot: { issueId: value },
+          runnerProfileJson: { adapterDispatch: { adapterType: value } } });
+      }
+    } finally { capture.flush(); await observed.$client.end(); }
+  });
+
   it("reads current workspace access without execution snapshots and retains JSON types", async () => {
     const { companyId, agent, issue } = await fixture();
     const payload = randomBytes(128 * 1024).toString("base64"), workspaceId = randomUUID();

server/src/services/heartbeat/run-state.ts

@@ -1131,6 +1131,33 @@ export function createHeartbeatRunState(db: Db) {
       .then((rows) => rows[0] ?? null);
   }
 
+  async function getRunExecutionWait(companyId: string, runId: string) {
+    const safeForLegacyEncoding = await hasUnsafeTextProjectionDatabase();
+    return db.select({
+      ...heartbeatRunIssueControlColumns,
+      finishedAt: heartbeatRuns.finishedAt,
+      errorCode: heartbeatRuns.errorCode,
+      error: safeForLegacyEncoding ? heartbeatRunSqlAsciiSafeColumns.error : heartbeatRuns.error,
+      nativeIssueId: heartbeatRuns.nativeIssueId,
+      nativeSessionId: heartbeatRuns.nativeSessionId,
+      processPid: heartbeatRuns.processPid,
+      processGroupId: heartbeatRunProcessGroupIdColumn,
+      processStartedAt: heartbeatRuns.processStartedAt,
+      sessionIdAfter: heartbeatRuns.sessionIdAfter,
+      executionStage: heartbeatRuns.executionStage,
+      controllerBootId: heartbeatRuns.controllerBootId,
+      controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt,
+      resultJson: heartbeatRunExecutionEvidenceColumn,
+      contextSnapshot: sql<Record<string, unknown> | null>`case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object'
+        then (select coalesce(jsonb_object_agg(key, value), '{}'::jsonb)
+          from jsonb_each(${heartbeatRuns.contextSnapshot})
+          where key = any(array['issueId', 'connectionIntentResolution', 'source', 'interactionKind',
+            'interactionStatus', 'mutation', 'interactionId'])) else null end`,
+    }).from(heartbeatRuns).where(and(
+      eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId),
+    )).then(rows => rows[0] ?? null);
+  }
+
   async function getRunCleanupState(runId: string) {
     return db.select({
       id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
@@ -1730,6 +1757,7 @@ export function createHeartbeatRunState(db: Db) {
     resolveSessionBeforeForWakeup,
     getRun,
     getRunCleanupState,
+    getRunExecutionWait,
     getRunWorkspaceSettlement,
     ensureRuntimeState,
     getTaskSession,

server/src/services/native-local-process-stop.ts

@@ -22,7 +22,8 @@ function absent(pid: number): boolean {
 
 /** A fresh workspace is not process containment. Bind an OS observation to
  * the original host, process birth, and launch event before accepting it. */
-export async function recordedLocalExecutionStopped(db: Db, run: typeof heartbeatRuns.$inferSelect) {
+export async function recordedLocalExecutionStopped(db: Db, run: Pick<typeof heartbeatRuns.$inferSelect,
+  "id" | "companyId" | "processPid" | "processGroupId" | "processStartedAt">) {
   const [identity] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and(
     eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
     eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED), isNull(heartbeatRunEvents.sourceEventId),