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),