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 <noreply@paperclip.ing>

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.

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,

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

@@ -13,6 +13,7 @@ import type { HeartbeatRunControlDependencies } from "../run-control.js";
 type TestCallbacks = HeartbeatRunControlDependencies & ReturnType<typeof createHeartbeatLifecycle> & {
   getRun: Mock<HeartbeatRunControlDependencies["getRun"]>;
   getAgent: Mock<HeartbeatRunControlDependencies["getAgent"]>;
+  getRunCleanupState: Mock<HeartbeatRunControlDependencies["getRunCleanupState"]>;
   setRunStatusFromLive: Mock<HeartbeatRunControlDependencies["setRunStatusFromLive"]>;
   getSchedulingSuppression: Mock<HeartbeatRunControlDependencies["getSchedulingSuppression"]>;
   options: { closeWarmNativeSessionsForRun: Mock<NonNullable<HeartbeatRunControlDependencies["options"]["closeWarmNativeSessionsForRun"]>> };
@@ -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),

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<HeartbeatRecoveryDependencies["getRun"]>(async () => null),
+    getRunWorkspaceSettlement: vi.fn(createHeartbeatRunState(db).getRunWorkspaceSettlement),
     getAgent: vi.fn<HeartbeatRecoveryDependencies["getAgent"]>(async () => null),
     appendRunEvent: vi.fn<HeartbeatRecoveryDependencies["appendRunEvent"]>(async () => undefined),
     setRunStatusIfRunning: vi.fn<HeartbeatRecoveryDependencies["setRunStatusIfRunning"]>(async () => ({ updated: false, run: null })),

server/src/services/heartbeat/recovery.ts

@@ -101,6 +101,7 @@ export interface HeartbeatRecoveryDependencies extends Pick<HeartbeatRetryDepend
   getRun: (runId: string, options?: Parameters<ReturnType<typeof createHeartbeatRunState>["getRun"]>[1]) => Promise<
     Awaited<ReturnType<ReturnType<typeof createHeartbeatRunState>["getRun"]>> | null
   >;
+  getRunWorkspaceSettlement: ReturnType<typeof createHeartbeatRunState>["getRunWorkspaceSettlement"];
   environmentRuntime: ReturnType<typeof environmentRuntimeService>;
   pluginWorkerManager?: WorkspaceWorkers;
   environmentsSvc: Pick<ReturnType<typeof environmentService>, "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,
     );

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" });

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

@@ -164,7 +164,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">,
+  "getRun" | "getAgent" | "getRunCleanupState">,
   Pick<ReturnType<typeof createHeartbeatLifecycle>,
     "mergeRunStopMetadataForAgent" | "setRunStatusFromLive" | "setRunStatus" | "setWakeupStatus" | "appendRunEvent">,
   Pick<HeartbeatRetryDependencies, "finalizeAgentStatus"> {
@@ -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) &&

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<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);
+    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({

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<Record<string, unknown> | null>`case when ${heartbeatRuns.runnerProfileJson} is null then null
+        else jsonb_build_object('nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync') end`,
+      resultJson: sql<Record<string, unknown> | null>`case when ${heartbeatRuns.resultJson} is null then null
+        else jsonb_build_object('workspaceExportRetry', ${heartbeatRuns.resultJson}->'workspaceExportRetry') end`,
+      contextSnapshot: sql<Record<string, unknown> | 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,

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<void>,
+) {
+  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<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 { 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");
+}

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<typeof import("../muse-runner-broker.js")>(),
+  museRunnerBroker: () => ({ reconcileTerminalAssignments: reconcileMuseAssignments }),
+}));
+
 const mockCaptureRunFailure = vi.hoisted(() => vi.fn());
 vi.mock("../../sentry.ts", async () => {
   const actual = await vi.importActual<typeof import("../../sentry.ts")>("../../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 () => {

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<typeof heartbeatRuns.$inferSelect, "id" | "companyId">, 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<Record<string, unknown> | 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()