From b7dc594bfaec15c8c87eb97a61931faf2d0aa518 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 25 Oct 2026 03:54:59 -0700 Subject: [PATCH] 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 --- doc/DATABASE.md | 5 ++ server/src/services/heartbeat.ts | 2 + server/src/services/heartbeat/run-control.ts | 9 +-- .../src/services/heartbeat/run-state.test.ts | 61 +++++++++++++++++++ server/src/services/heartbeat/run-state.ts | 28 +++++++++ .../src/services/native-local-process-stop.ts | 3 +- 6 files changed, 103 insertions(+), 5 deletions(-) diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 06cf24c9368..a814c50cfad 100644 --- a/doc/DATABASE.md +++ b/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, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index adb10d852c9..354f680eb1d 100644 --- a/server/src/services/heartbeat.ts +++ b/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, diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index b4ba5f0ddea..7387da1f74d 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -165,7 +165,7 @@ async function withBudgetStopDelivery(db: Db, work: () => Promise): Promis /** Cancellation shares executor ownership and delegates lifecycle and admission effects. */ export interface HeartbeatRunControlDependencies extends Pick, - "getRun" | "getAgent" | "getRunCleanupState">, + "getRun" | "getAgent" | "getRunCleanupState" | "getRunExecutionWait">, Pick, "mergeRunStopMetadataForAgent" | "setRunStatusFromLive" | "setRunStatus" | "setWakeupStatus" | "appendRunEvent">, Pick { @@ -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>>, 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 => { diff --git a/server/src/services/heartbeat/run-state.test.ts b/server/src/services/heartbeat/run-state.test.ts index 702f842e464..1b9c4276a61 100644 --- a/server/src/services/heartbeat/run-state.test.ts +++ b/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> = [], 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(); diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts index f870c3957fb..7ac3a3816a2 100644 --- a/server/src/services/heartbeat/run-state.ts +++ b/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 | 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, diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 7e74b8eca1c..422d2c24f0f 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/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) { 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),