From 64a8b927c1f199c612bcd36dccb6190e6e250eb3 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 24 Oct 2026 17:48:47 -0700 Subject: [PATCH] fix(native): narrow committed finalization and cleanup reads (#17525) Select fresh metadata for committed finalization and cleanup. Preserve full saved evidence for new finalization, cancelled workspace retention, and legacy Stop. PostgreSQL byte-budget and unchanged-row regressions pass, full CI is green, and Greptile scored the exact head 5/5. Co-Authored-By: Paperclip --- doc/DATABASE.md | 9 +++ server/src/services/heartbeat.ts | 4 ++ .../__tests__/run-control-fixtures.ts | 2 + .../src/services/heartbeat/recovery.test.ts | 2 + server/src/services/heartbeat/recovery.ts | 4 +- .../services/heartbeat/run-control.test.ts | 6 +- server/src/services/heartbeat/run-control.ts | 22 ++++--- .../src/services/heartbeat/run-state.test.ts | 57 +++++++++++++++++++ server/src/services/heartbeat/run-state.ts | 24 ++++++++ .../__tests__/helpers/committed-replay.ts | 34 +++++++++++ .../native-run-finalizer-telemetry.test.ts | 29 ++++++---- .../native-runtime/native-run-finalizer.ts | 55 ++++++++++++------ 12 files changed, 208 insertions(+), 40 deletions(-) create mode 100644 server/src/services/native-runtime/__tests__/helpers/committed-replay.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index e6211773d87..717ec1ca0ed 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -68,6 +68,15 @@ Runner profiles retain the adapter type and complete native execution input and workspace-sync receipts. Full result evidence stays available to the existing retry and workspace checks. Stored context, profiles, and output are unchanged. +Committed native finalization reads run identity and ownership state before it +claims the coordinator. It loads complete saved input only when finalization +still needs to process an accepted result. Review presentation retains the full +result. Muse settlement reads its exact provider binding. Cleanup reads fresh +status and ownership metadata; cancelled native workspace retention and legacy +Stop settlement still load complete run evidence. Workspace settlement reads the +saved sync descriptor, export retry, and first-heartbeat flag. These projections +leave stored prompts, logs, results, and recovery state unchanged. + Conversation ownership checks select process identity and full recovery result evidence without execution prompts or profiles. Committed native reply repair selects its persisted result and identity bindings plus typed presentation flags. diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 543d96a97f8..9a18721aaa7 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -811,6 +811,8 @@ export function heartbeatService( getAgent, resolveSessionBeforeForWakeup, getRun, + getRunCleanupState, + getRunWorkspaceSettlement, getRunStatus, ensureRuntimeState, getTaskSession, @@ -1014,6 +1016,7 @@ export function heartbeatService( environmentRuntime, scheduleNativeSessionResumeDispatch, getRun, + getRunWorkspaceSettlement, executeRun, activeRunExecutionPromises, cancelHeartbeatNativeRun, @@ -1386,6 +1389,7 @@ export function heartbeatService( const runControl = createHeartbeatRunControl(db, { getRun, + getRunCleanupState, options, envOrchestrator, activeRunExecutions, diff --git a/server/src/services/heartbeat/__tests__/run-control-fixtures.ts b/server/src/services/heartbeat/__tests__/run-control-fixtures.ts index 42f6bea5150..2adce9077c1 100644 --- a/server/src/services/heartbeat/__tests__/run-control-fixtures.ts +++ b/server/src/services/heartbeat/__tests__/run-control-fixtures.ts @@ -13,6 +13,7 @@ import type { HeartbeatRunControlDependencies } from "../run-control.js"; type TestCallbacks = HeartbeatRunControlDependencies & ReturnType & { getRun: Mock; getAgent: Mock; + getRunCleanupState: Mock; setRunStatusFromLive: Mock; getSchedulingSuppression: Mock; options: { closeWarmNativeSessionsForRun: Mock> }; @@ -43,6 +44,7 @@ export function callbacks(db: Db): TestCallbacks { ...state, ...lifecycle, getRun: vi.fn(state.getRun), + getRunCleanupState: vi.fn(state.getRunCleanupState), getAgent: vi.fn(state.getAgent), setRunStatusFromLive: vi.fn(lifecycle.setRunStatusFromLive), setRunStatus: vi.fn(lifecycle.setRunStatus), diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index 8faa8c7e865..34139757468 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -7,6 +7,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../environment-runtime.js"; import { environmentService } from "../environments.js"; import { createHeartbeatRecovery, type HeartbeatRecoveryDependencies } from "./recovery.js"; +import { createHeartbeatRunState } from "./run-state.js"; import type { HotRestartIntent } from "../hot-restart.js"; import type { NativeRestartRecoveryClaim, NativeRestartRecoveryDisposition } from "../native-runtime/index.js"; import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "../native-runtime/native-runner-ownership.js"; @@ -48,6 +49,7 @@ function callbacks(db: Db) { return { enterShutdown: vi.fn(), getRun: vi.fn(async () => null), + getRunWorkspaceSettlement: vi.fn(createHeartbeatRunState(db).getRunWorkspaceSettlement), getAgent: vi.fn(async () => null), appendRunEvent: vi.fn(async () => undefined), setRunStatusIfRunning: vi.fn(async () => ({ updated: false, run: null })), diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 73c586bdff3..a5197485881 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -101,6 +101,7 @@ export interface HeartbeatRecoveryDependencies extends Pick["getRun"]>[1]) => Promise< Awaited["getRun"]>> | null >; + getRunWorkspaceSettlement: ReturnType["getRunWorkspaceSettlement"]; environmentRuntime: ReturnType; pluginWorkerManager?: WorkspaceWorkers; environmentsSvc: Pick, "getById" | "getLeaseById" | "releaseLease">; @@ -360,6 +361,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD environmentRuntime, scheduleNativeSessionResumeDispatch, getRun, + getRunWorkspaceSettlement, executeRun, activeRunExecutionPromises, cancelHeartbeatNativeRun, @@ -1858,7 +1860,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD agentId: string; succeeded: boolean; }) { - const settledRun = await getRun(input.runId); + const settledRun = await getRunWorkspaceSettlement(input.runId); const workspaceSyncReference = readNativeWorkspaceSyncReference( parseObject(settledRun?.runnerProfileJson).nativeWorkspaceSync, ); diff --git a/server/src/services/heartbeat/run-control.test.ts b/server/src/services/heartbeat/run-control.test.ts index 7f3991eaa0c..d0d08896538 100644 --- a/server/src/services/heartbeat/run-control.test.ts +++ b/server/src/services/heartbeat/run-control.test.ts @@ -98,7 +98,7 @@ describe("heartbeat run-control module boundary", () => { it("retains leases while native execution ownership is unverified", async () => { const database = guardedDatabase(); const deps = callbacks(database.db); - deps.getRun.mockResolvedValue({ runtimeMode: "native", status: "running", nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" } as Run); + deps.getRunCleanupState.mockResolvedValue({ runtimeMode: "native", status: "running", nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" } as Run); await createHeartbeatRunControl(database.db, deps).releaseEnvironmentLeasesForRun({ runId: "held", companyId: "company", agentId: "agent", status: "failed", providerResourceDisposition: "destroy" }); expect(database.access).not.toHaveBeenCalled(); expect(deps.options.closeWarmNativeSessionsForRun).not.toHaveBeenCalled(); @@ -108,7 +108,7 @@ describe("heartbeat run-control module boundary", () => { it("uses the persisted terminal outcome when choosing warm resource retention", async () => { const database = guardedDatabase(); const deps = callbacks(database.db); - deps.getRun.mockResolvedValue({ status: "failed", runtimeMode: "legacy" } as Run); + deps.getRunCleanupState.mockResolvedValue({ status: "failed", runtimeMode: "legacy" } as Run); await createHeartbeatRunControl(database.db, deps).releaseEnvironmentLeasesForRun({ runId: "failed", companyId: "company", agentId: "agent", status: "succeeded", providerResourceDisposition: "keep_running" }); expect(deps.envOrchestrator.releaseForRun).toHaveBeenCalledWith(expect.objectContaining({ heartbeatRunId: "failed", status: "failed", providerResourceDisposition: "stop_and_retain" })); expect(database.access).not.toHaveBeenCalled(); @@ -117,7 +117,7 @@ describe("heartbeat run-control module boundary", () => { it("defers destruction when a warm session remains busy", async () => { const database = guardedDatabase(); const deps = callbacks(database.db); - deps.getRun.mockResolvedValue({ status: "failed", runtimeMode: "legacy" } as Run); + deps.getRunCleanupState.mockResolvedValue({ status: "failed", runtimeMode: "legacy" } as Run); deps.options.closeWarmNativeSessionsForRun.mockResolvedValue({ closed: 0, busy: 1, failed: 0 }); await createHeartbeatRunControl(database.db, deps).releaseEnvironmentLeasesForRun({ runId: "busy", companyId: "company", agentId: "agent", status: "failed", providerResourceDisposition: "destroy" }); expect(deps.options.closeWarmNativeSessionsForRun).toHaveBeenCalledWith({ runId: "busy", reason: "terminal heartbeat run destroyed its environment lease" }); diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 7f80f99321a..d7086c52508 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -164,7 +164,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">, + "getRun" | "getAgent" | "getRunCleanupState">, Pick, "mergeRunStopMetadataForAgent" | "setRunStatusFromLive" | "setRunStatus" | "setWakeupStatus" | "appendRunEvent">, Pick { @@ -324,6 +324,7 @@ function runCancellationFenceSql(input: { export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunControlDependencies) { const { getRun, + getRunCleanupState, options, envOrchestrator, activeRunExecutions, @@ -359,7 +360,11 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont sandboxResource: "keep_running" | "stop_and_reuse" | "destroy_after_turn"; }; }) { - const leaseOwnerRun = await getRun(input.runId); + let leaseOwnerRun = await getRunCleanupState(input.runId); + if (leaseOwnerRun && isNativeRunnerOwnershipHeld(leaseOwnerRun)) return; + const cancelledRun = leaseOwnerRun?.runtimeMode === "native" && leaseOwnerRun.status === "cancelled" + ? await getRun(input.runId) : null; + if (cancelledRun) leaseOwnerRun = cancelledRun; if (leaseOwnerRun && isNativeRunnerOwnershipHeld(leaseOwnerRun)) return; // Recovery can finish workspace copy-back outside the executor's finally. // Successful copy-back does not earn warm retention for a failed turn. @@ -368,11 +373,11 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont input.providerResourceDisposition, status, ); - if (leaseOwnerRun?.runtimeMode === "native" && status === "cancelled" - && readNativeWorkspaceSyncReference(leaseOwnerRun.runnerProfileJson?.nativeWorkspaceSync)?.state === "prepared" - && !nativeRunHasNoWorkspaceAccess(leaseOwnerRun)) { + if (cancelledRun?.runtimeMode === "native" && status === "cancelled" + && readNativeWorkspaceSyncReference(cancelledRun.runnerProfileJson?.nativeWorkspaceSync)?.state === "prepared" + && !nativeRunHasNoWorkspaceAccess(cancelledRun)) { try { - const preserved = await preserveNativeFailedWorkspace(db, leaseOwnerRun, { allowPreterminal: true }); + const preserved = await preserveNativeFailedWorkspace(db, cancelledRun, { allowPreterminal: true }); if (!preserved || !hasRequiredWorkspaceRecovery(preserved.resultJson)) return; } catch { // Keep the active source eligible for the bounded retention sweep. @@ -478,13 +483,14 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont async function acknowledgeRemoteStop(runId: string, companyId: string) { // The provider receipt arrives after adapter settlement. A remote ACP child // has no host PID, so only this target-aware boundary can acknowledge Stop. - const stopped = await getRun(runId); - if (stopped?.runtimeMode === "native") { + const state = await getRunCleanupState(runId); + if (state?.runtimeMode === "native") { const scopes = await stoppedRemoteCleanupScopes(db, companyId, runId); for (const remoteCleanupScope of scopes ?? []) { completeTerminatedRemoteNativeSessionCleanup({ companyId, runId, remoteCleanupScope }); } } + const stopped = state?.runtimeMode === "legacy" && state.status === "cancelled" ? await getRun(runId) : null; if (stopped?.runtimeMode === "legacy" && stopped.status === "cancelled" && parseObject(stopped.resultJson?.executionCancellation).state === "requested" && await runUsedConversationAdapter(db, stopped) && diff --git a/server/src/services/heartbeat/run-state.test.ts b/server/src/services/heartbeat/run-state.test.ts index 4a729328027..ccc91735df4 100644 --- a/server/src/services/heartbeat/run-state.test.ts +++ b/server/src/services/heartbeat/run-state.test.ts @@ -1,4 +1,5 @@ import { randomBytes, randomUUID } from "node:crypto"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; import type { Db } from "@paperclipai/db"; import { agents, @@ -86,6 +87,9 @@ describe("heartbeat run state module", () => { expect(projectedSql(utf8.columns[0].resultJson)).toContain("pg_column_size"); await asciiState.getRun("evidence", { includeExecutionEvidence: true }); expect(projectedSql(ascii.columns[2].resultJson)).toContain("startupPreparationSettledAt"); + await asciiState.getRunWorkspaceSettlement("settlement"); + expect(projectedSql(ascii.columns[3].error)).toBe("NULL"); + expect(ascii.execute).toHaveBeenCalledTimes(1); const raw = readingDatabase("SQL_ASCII"); await extracted.createHeartbeatRunState(raw.db).getRun("raw", { unsafeFullResultJson: true }); expect(raw.execute).not.toHaveBeenCalled(); @@ -154,6 +158,59 @@ describePostgres("heartbeat run state database wiring", () => { return { companyId, agent, issue, state: extracted.createHeartbeatRunState(db) }; } + 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); + const sync = { schema: "paperclip.native-workspace-sync.v1", state: "prepared", resourceDisposition: "destroy" }; + const [run] = await db.insert(heartbeatRuns).values({ + companyId, agentId: agent.id, issueId: issue.id, runtimeMode: "native", status: "failed", + errorCode: "provider_transport_failed", error: "Retain saved files", nativePhase: "terminal_failure", + runnerProfileJson: { nativeWorkspaceSync: sync, savedInput: bulk }, + resultJson: { workspaceExportRetry: { requestId: "retry-source" }, stdout: bulk }, + contextSnapshot: { timerClaimWasFirstHeartbeat: true, savedInput: bulk }, + stdoutExcerpt: bulk, stderrExcerpt: bulk, + }).returning(); + await state.getRunWorkspaceSettlement(run.id); + const events: Array> = []; + const now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRuns }, capture); + expect(await state.getRunCleanupState(run.id)).toEqual({ + id: run.id, companyId, agentId: agent.id, runtimeMode: run.runtimeMode, + status: run.status, errorCode: run.errorCode, nativePhase: run.nativePhase, + }); + expect(await state.getRunWorkspaceSettlement(run.id)).toEqual({ + id: run.id, status: run.status, error: run.error, + runnerProfileJson: { nativeWorkspaceSync: sync }, + resultJson: { workspaceExportRetry: { requestId: "retry-source" } }, + contextSnapshot: { timerClaimWasFirstHeartbeat: true }, + }); + capture.flush(); + const summary = events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!; + expect(summary.completedQueries).toBe(2); + expect(summary.responseValueBytes).toBeLessThan(2000); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toEqual(run); + expect(JSON.stringify(events)).not.toContain("cleanup-saved-input-sentinel"); + expect(await state.getRunCleanupState(randomUUID())).toBeNull(); + expect(await state.getRunWorkspaceSettlement(randomUUID())).toBeNull(); + }); + + it("preserves JSON types in workspace settlement decisions", async () => { + const { companyId, agent, state } = await fixture(); + for (const value of [null, false, true, "true", 0, [], { future: [1, "retained"] }]) { + const [run] = await db.insert(heartbeatRuns).values({ + companyId, agentId: agent.id, + runnerProfileJson: { nativeWorkspaceSync: value }, + resultJson: { workspaceExportRetry: value }, + contextSnapshot: { timerClaimWasFirstHeartbeat: value }, + }).returning({ id: heartbeatRuns.id }); + const actual = await state.getRunWorkspaceSettlement(run.id); + expect(actual?.runnerProfileJson?.nativeWorkspaceSync).toEqual(value); + expect(actual?.resultJson?.workspaceExportRetry).toEqual(value); + expect(actual?.contextSnapshot?.timerClaimWasFirstHeartbeat).toEqual(value); + } + }); + it("retains the consumed queue identity in issue polling summaries", async () => { const { companyId, agent, issue, state } = await fixture(); const [wake] = await db.insert(agentWakeupRequests).values({ diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts index 30f9ef0b001..1911ad1a25c 100644 --- a/server/src/services/heartbeat/run-state.ts +++ b/server/src/services/heartbeat/run-state.ts @@ -1119,6 +1119,28 @@ export function createHeartbeatRunState(db: Db) { .then((rows) => rows[0] ?? null); } + async function getRunCleanupState(runId: string) { + return db.select({ + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, nativePhase: heartbeatRuns.nativePhase, + }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0] ?? null); + } + + async function getRunWorkspaceSettlement(runId: string) { + const safeForLegacyEncoding = await hasUnsafeTextProjectionDatabase(); + return db.select({ + id: heartbeatRuns.id, status: heartbeatRuns.status, + error: safeForLegacyEncoding ? heartbeatRunSqlAsciiSafeColumns.error : heartbeatRuns.error, + runnerProfileJson: sql | null>`case when ${heartbeatRuns.runnerProfileJson} is null then null + else jsonb_build_object('nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync') end`, + resultJson: sql | null>`case when ${heartbeatRuns.resultJson} is null then null + else jsonb_build_object('workspaceExportRetry', ${heartbeatRuns.resultJson}->'workspaceExportRetry') end`, + contextSnapshot: sql | null>`case when ${heartbeatRuns.contextSnapshot} is null then null + else jsonb_build_object('timerClaimWasFirstHeartbeat', ${heartbeatRuns.contextSnapshot}->'timerClaimWasFirstHeartbeat') end`, + }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0] ?? null); + } + async function getRunStatus(runId: string) { return db .select({ @@ -1688,6 +1710,8 @@ export function createHeartbeatRunState(db: Db) { getAgent, resolveSessionBeforeForWakeup, getRun, + getRunCleanupState, + getRunWorkspaceSettlement, ensureRuntimeState, getTaskSession, getLatestAgentConfigRevision, diff --git a/server/src/services/native-runtime/__tests__/helpers/committed-replay.ts b/server/src/services/native-runtime/__tests__/helpers/committed-replay.ts new file mode 100644 index 00000000000..7caef7cf216 --- /dev/null +++ b/server/src/services/native-runtime/__tests__/helpers/committed-replay.ts @@ -0,0 +1,34 @@ +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; +import { eq } from "drizzle-orm"; +import { expect } from "vitest"; + +export async function verifyCommittedReplayPayloadReads( + db: Db, + binding: { runId: string; companyId: string; agentId: string }, + replay: () => Promise, +) { + const { runId, companyId, agentId } = binding; + const bulk = "saved-native-input-sentinel🙂".repeat(40_000); + const [original] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + await db.update(heartbeatRuns).set({ + contextSnapshot: { ...original.contextSnapshot, savedInput: bulk }, + runnerProfileJson: { savedInput: bulk, nativeExecutionInput: { + schema: "paperclip.native-execution-input.v8", savedInput: bulk, + provider: { kind: "muse", binding: { companyId, agentId, bindingId: runId } }, + } }, stdoutExcerpt: bulk, stderrExcerpt: bulk, + }).where(eq(heartbeatRuns.id, runId)); + const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const events: Array> = []; + const now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRuns }, capture); + try { await replay(); } + finally { capture.flush(); } + const summary = events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!; + expect(summary.completedQueries).toBeGreaterThan(0); + expect(summary.responseValueBytes).toBeLessThan(100_000); + const [after] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(after).toEqual(before); + expect(JSON.stringify(events)).not.toContain("saved-native-input-sentinel"); +} diff --git a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts b/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts index 0a8b6ce4ed1..2c2e534081b 100644 --- a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts +++ b/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts @@ -1,5 +1,6 @@ import { enableRunFailureReporting } from "../../__tests__/helpers/run-failure-reporting.js"; import { randomUUID } from "node:crypto"; +import { verifyCommittedReplayPayloadReads } from "./__tests__/helpers/committed-replay.js"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { @@ -34,6 +35,12 @@ const mockTelemetryClient = vi.hoisted(() => ({ })); vi.mock("../../telemetry.ts", () => ({ getTelemetryClient: () => mockTelemetryClient })); +const reconcileMuseAssignments = vi.hoisted(() => vi.fn(async () => undefined)); +vi.mock("../muse-runner-broker.js", async importOriginal => ({ + ...await importOriginal(), + museRunnerBroker: () => ({ reconcileTerminalAssignments: reconcileMuseAssignments }), +})); + const mockCaptureRunFailure = vi.hoisted(() => vi.fn()); vi.mock("../../sentry.ts", async () => { const actual = await vi.importActual("../../sentry.ts"); @@ -261,18 +268,18 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a projectRunStatus: true, }); - const callsBefore = mockTelemetryClient.track.mock.calls.length; - // The coordinator is already "committed", so this second call takes the - // projectCommittedRun short-circuit. A succeeded row is eligible so stale - // cleanup errors can be projected separately; an unchanged terminal state - // must still never emit a second agent.task_run event. - await finalizeNativeRun({ - db, - runId: fixture.runId, - workspaceFinalizeStatus: "succeeded", - projectRunStatus: true, + await verifyCommittedReplayPayloadReads(db, { runId: fixture.runId, companyId, agentId }, async () => { + const callsBefore = mockTelemetryClient.track.mock.calls.length; + reconcileMuseAssignments.mockClear(); + await finalizeNativeRun({ + db, + runId: fixture.runId, + workspaceFinalizeStatus: "succeeded", + projectRunStatus: true, + }); + expect(agentTaskRunCalls(callsBefore)).toHaveLength(0); + expect(reconcileMuseAssignments).toHaveBeenCalledExactlyOnceWith(companyId, agentId, fixture.runId); }); - expect(agentTaskRunCalls(callsBefore)).toHaveLength(0); }); it("repairs a stale succeeded-run error once without duplicate telemetry or destroying retained owner evidence", async () => { diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 8b2ae73c349..7d60d73e0d7 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -568,7 +568,7 @@ export async function recordNativeFinalizationFailure(input: { }); } -async function resolveCommittedFinalizationRecovery(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string) { +async function resolveCommittedFinalizationRecovery(db: Db, run: Pick, issueId: string) { const actions = await db.select().from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId), eq(issueRecoveryActions.kind, "active_run_watchdog"), inArray(issueRecoveryActions.status, ["active", "escalated"]), @@ -704,7 +704,8 @@ async function projectCommittedRun(input: { async function materializeCommittedReviewResponse(db: Db, runId: string) { try { const [run] = await db - .select() + .select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + nativeIssueId: heartbeatRuns.nativeIssueId, resultJson: heartbeatRuns.resultJson }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, runId)) .limit(1); @@ -1146,13 +1147,15 @@ export async function finalizeNativeRun(input: { preserveProviderAttempt?: boolean; failpoint?: NativeStatusCommitFailpoint; }) { - const run = await input.db - .select() + const admittedRun = await input.db + .select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode, + nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .limit(1) .then((rows) => rows[0] ?? null); - if (!run || run.runtimeMode !== "native") + if (!admittedRun || admittedRun.runtimeMode !== "native") throw new Error("native_finalization_run_missing"); const claim = await claimCoordinator({ db: input.db, @@ -1161,12 +1164,12 @@ export async function finalizeNativeRun(input: { }); const coordinator = claim.coordinator; if (!claim.leaseOwner && coordinator.phase === "committed") { - if (isNativeRunnerOwnershipHeld(run)) return coordinator; + if (isNativeRunnerOwnershipHeld(admittedRun)) return coordinator; const presentationAlreadyMaterialized = coordinator.decisionId ? await hasMaterializedNativeReviewResponse(input.db, { - companyId: run.companyId, + companyId: admittedRun.companyId, issueId: coordinator.issueId, - runId: run.id, + runId: admittedRun.id, decisionId: coordinator.decisionId, }) : false; @@ -1178,9 +1181,9 @@ export async function finalizeNativeRun(input: { ) { await input.db.transaction((tx) => restoreNativeChatReviewPresentationInTransaction(tx as unknown as Db, { - companyId: run.companyId, + companyId: admittedRun.companyId, issueId: coordinator.issueId, - runId: run.id, + runId: admittedRun.id, decisionId: coordinator.decisionId!, resultId: coordinator.resultId!, assessmentId: coordinator.assessmentId!, @@ -1188,27 +1191,45 @@ export async function finalizeNativeRun(input: { ); } if (input.projectRunStatus) - await projectCommittedRun({ db: input.db, run, coordinator }); + await projectCommittedRun({ db: input.db, run: admittedRun, coordinator }); if (input.projectRunStatus && !presentationAlreadyMaterialized) await materializeCommittedReviewResponse(input.db, input.runId); if (input.projectRunStatus) await repairCommittedNativeChatResponse(input.db, { - companyId: run.companyId, + companyId: admittedRun.companyId, issueId: coordinator.issueId, - runId: run.id, + runId: admittedRun.id, }); if (input.projectRunStatus) { - await settleExternalConversation(input.db, run.companyId, coordinator.issueId).catch((err) => { - logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation"); + await settleExternalConversation(input.db, admittedRun.companyId, coordinator.issueId).catch((err) => { + logger.warn({ err, runId: admittedRun.id }, "Slack conversation settlement deferred to reconciliation"); }); } - await resolveCommittedFinalizationRecovery(input.db, run, coordinator.issueId); + await resolveCommittedFinalizationRecovery(input.db, admittedRun, coordinator.issueId); if (input.projectRunStatus) { - const [projectedRun] = await input.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + const [projectedRun] = await input.db.select({ + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, + runnerProfileJson: sql | null>`case when ${heartbeatRuns.runnerProfileJson} is null then null + else jsonb_build_object('nativeExecutionInput', jsonb_build_object( + 'schema', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,schema}', + 'provider', jsonb_build_object( + 'kind', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,provider,kind}', + 'binding', jsonb_build_object( + 'companyId', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,provider,binding,companyId}', + 'agentId', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,provider,binding,agentId}', + 'bindingId', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,provider,binding,bindingId}' + ) + ) + )) end`, + }).from(heartbeatRuns).where(eq(heartbeatRuns.id, admittedRun.id)); if (projectedRun) await reconcileMuseNativeTerminalRun(input.db, projectedRun); } return coordinator; } + const run = await input.db.select().from(heartbeatRuns) + .where(eq(heartbeatRuns.id, input.runId)).limit(1).then(rows => rows[0] ?? null); + if (!run || run.runtimeMode !== "native") throw new Error("native_finalization_run_missing"); const [resultRow, contractRow] = await Promise.all([ input.db .select()