fix(server): wake committed native run projection from durable changes (#17752)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A native run has separate result acceptance, task decision, run status, and resource settlement steps. > - A task decision can commit before its run status is saved. > - Recovery currently revisits committed decisions through the periodic orphan pass. > - This PR gives the existing run projection repair its own transaction-driven worker. > - Existing records reconstruct unfinished work after restart. Completed history becomes quiet. > - Other completion and resource obligations keep their existing recovery paths. ## Linked Issues or Issue Description Refs #17748 (stack parent: accepted-result commit), #17734 (workspace finalization), and #17717 (retained session cleanup). Related #17614 preserves terminal notifications. This slice is prepared on current parent #17748 at `b2efdb2dbbbef5b2227ffa2bd2a9a6cfe3867fe2`; the parent has merged as `28e38da7e6bdd6af4092dda56bc2be05bed47af5`. **Current behavior** The native recovery pass revisits committed decisions to repair the heartbeat-run status and metadata. A crash between decision and run writes creates real work, but periodic scanning also revisits retained history. **Proposed behavior** Wake projection repair after committed producer changes. Reconstruct mismatches on startup and settings activation. Retry concrete failures through the shared scheduler. Keep the run projection before resource settlement. **Reason and benefit** The repair worker performs no recurring SQL after its work settles. A real decision, effect settlement, or ownership release reactivates it. This is one worker migration; it does not establish host or database suspension. ## What Changed - Extract the existing projection predicate and write into their native-runtime owner. The finalizer keeps initial metadata enrichment and task decision authority. - Discover accepted, committed, mismatched run projections with company, issue and run bindings. Exclude pending status effects and held runner ownership. Read narrow fields instead of whole result payloads. - Use existing transaction hints, scoped delivery scheduling, failure backoff and lifecycle handling. Add no tables or per-feature timers. - Require issue/run scope for ordinary finalization fanout. Routine company-only hints leave dormant projection quiet; explicit settings and dedicated producer signals keep their existing contract. - Connect finalization, pending effects, lifecycle status writes, issue changes and worktree execution activation. Preserve outer transaction commit and rollback behavior. - Lock the current run before deciding which terminal event and completion time to save. Concurrent live and recovery projection must not announce the same transition twice. - Delegate only already-committed projection from the periodic pass. Its candidate scan uses the same pending predicate to defer remaining consequences and cleanup until projection finishes. - Add real PostgreSQL producer, race and lifecycle tests plus an owner audit. ## Verification - Current local source `a83b9449233a275176e637c765059f414305d7c1` incorporates the parent with a normal merge. All 1,208 parent-only, seven owned-only and ten automatic shared paths are preserved. The only manual conflict keeps new projection fanout inside the ordinary issue/run guard. The complete package tree matches parent profile 80 byte-for-byte. - Controlled real-database regression: the original projection-only unscoped fanout produces three candidate SELECTs for routine company hints. The same zero-query assertion passes with scoped fanout. A real committed result then repairs the run and emits one terminal event before returning to quiet. Production bytes restore exactly in both phases. Twenty cases outside that focused pattern are not counted as passing there. - The complete main command passes **1,830 cases in 19 files**, without skips, in 301.14 seconds. It includes all 21 projection cases, 85 startup cases, required quality review and heartbeat run lifecycle, plus physical-work and result-commit consumers. Separate process recovery passes 410 in 163.58 seconds and external-chat response passes 273 in 53.26 seconds. Total: **2,513 distinct passes in 21 complete files**, with no skips. Normal merge `d4b46fd29cb23126d556cc3aeb883e6b41f1a8a5` adopts the actual parent squash with this entire tested child tree unchanged. - Production server typecheck, build, module boundaries, migration order and source identity checks pass. Seventeen dedicated PostgreSQL 17 databases use the unchanged migration 413 schema. Embedded-only calls use the existing fresh-per-call recovery helper. All temporary imports restore byte-exact and all 21 selected test hashes match after completion. No assertion or timeout is relaxed. - Parent package acceptance remains applicable: 130 passing installation/compatibility/transport cases, 22 installed ACPX/Grok contract cases and 26 native-provider Rust cases. One pre-existing opt-in closed Pi snapshot case is skipped because its distribution root is not configured. No live provider qualification is claimed. - The owner audit at `doc/plans/2026-10-26-native-run-projection-sleep-audit.md` retains prior failures and their disposition. Current-head acceptance: all 57 checks are clean, including protected CI and security. Greptile gives 5/5 on `a49a2dac13b0e1c19da5d3d6b435c781703b1c0b`, with no actionable findings. Old scores do not substitute. No Ponytail verdict is present under the authorized missing-review policy. - These checks do not claim a full local repository suite, deployed suspension, expanded Cloud wake coverage or measured cost savings. ## Risks - Process-local signals cover instrumented application writers. Independent SQL writers and replicas without supported ingress are not newly covered. Durable startup reconstruction remains enabled. - Projection must precede remaining committed resource settlement. The real reconciler test proves the periodic path defers cleanup while projection is pending and resumes it after repair. - Status decisions, external conversation settlement, watchdog resolution, Muse reconciliation and resource release keep their own authority and recovery. A committed result alone is not proof of completed cleanup. - Candidate discovery uses narrow bounded pages. There is no large-fleet performance or deployed sleep claim. Existing Cloud wake coverage and sleep blockers remain unchanged. - Parent #17748 has merged. This PR targets master at `28e38da7e6bdd6af4092dda56bc2be05bed47af5`. Current-head CI and review must pass before the authorized protected merge. ## Model Used OpenAI Codex, GPT-6 family, with repository tools and code execution. The exact serving model ID and context-window size are not exposed in this session. No subagent was used for this slice. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Codex <noreply@openai.com>
2026-11-04 11:01:16 +01:00 · 2026-10-29 05:10:16 -05:00
1 parent 28e38da7e6
commit 50fe32684d
@@ -0,0 +1,337 @@
+import { heartbeatService } from "../services/heartbeat.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
+import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "../services/heartbeat/run-lifecycle.js";
+import { reconcileNativeFinalizations } from "../services/native-runtime/native-finalization-reconciler.js";
+import { statusDecisionEffects } from "@paperclipai/db";
+import { randomUUID } from "node:crypto";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { setTimeout as delay } from "node:timers/promises";
+import { eq } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { nativeRunProjectionWork } from "../services/native-runtime/native-run-projection-work.js";
+import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js";
+import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js";
+import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js";
+import * as statusCommitter from "../services/native-runtime/status-decision-committer.js";
+import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js";
+import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js";
+import { workspaceOperationService } from "../services/workspace-operations.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js";
+import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { emitAgentTaskRun } from "../services/agent-task-run-telemetry.js";
+vi.mock("../services/agent-task-run-telemetry.js", () => ({ emitAgentTaskRun: vi.fn(async () => undefined) }));
+vi.mock("../services/run-failure-report.js", async importOriginal => ({
+  ...await importOriginal<typeof import("../services/run-failure-report.js")>(), reportRunFailure: vi.fn(async () => undefined),
+}));
+
+describe("committed native run projection lifecycle", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: Db, root: string;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let worker: ReturnType<ReturnType<typeof nativeRunProjectionWork>["register"]> | undefined;
+  const companyId = randomUUID(), agentId = randomUUID();
+  const priorLogs = process.env.WORKSPACE_OPERATION_LOG_BASE_PATH;
+  let admitted = true;
+  const errors: unknown[] = [];
+  beforeAll(async () => {
+    root = await mkdtemp(join(tmpdir(), "native-run-projection-"));
+    process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = root;
+    const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RUN_PROJECTION_TEST_DATABASE_URL");
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-run-projection-");
+    db = createDb(external ?? temporary!.connectionString);
+    await db.insert(companies).values({ id: companyId, name: "Result commits", issuePrefix: `RC${randomUUID().slice(0, 6)}` });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Result worker", adapterType: "paperclip_runner" });
+  }, 30_000);
+  beforeEach(() => {
+    admitted = true; errors.length = 0; vi.mocked(emitAgentTaskRun).mockClear();
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+  });
+  afterEach(async () => {
+    await coordinator?.stop(); coordinator = undefined; worker = undefined;
+    vi.useRealTimers(); vi.restoreAllMocks();
+    await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.companyId, companyId));
+    expect(errors).toEqual([]);
+  });
+  afterAll(async () => {
+    await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); await rm(root, { recursive: true, force: true });
+    if (priorLogs === undefined) delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH;
+    else process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = priorLogs;
+  });
+  async function start(project = projectCommittedRun) {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted,
+      onError: error => errors.push(error) });
+    worker = nativeRunProjectionWork(db, project).register(coordinator, () => admitted);
+    await worker.ready;
+  }
+  async function flush() {
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(idleWorkSnapshot().active).toBe(0);
+      expect(coordinator!.nextWakeAt()).toBeNull();
+    }, { timeout: 5000 });
+  }
+  async function quiet() {
+    const previous = db.$client.options.debug;
+    const queries: string[] = [];
+    db.$client.options.debug = (_id, query) => { queries.push(query); };
+    try { await vi.advanceTimersByTimeAsync(120_000); await delay(10); expect(queries).toEqual([]); }
+    finally { db.$client.options.debug = previous; }
+  }
+  async function seed() {
+    const runId = randomUUID(), issueId = randomUUID(), contractId = randomUUID(), sourceInstanceId = randomUUID();
+    await db.insert(issues).values({ id: issueId, companyId, title: "Commit accepted work", status: "in_progress", assigneeAgentId: agentId });
+    await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1,
+      schemaVersion: "paperclip.completion-contract.v1", policyVersion: "result-worker-v1", risk: "standard",
+      completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
+      contractJson: { revision: "result-worker-v1", objective: "Finish", criteria: [{ id: "objective", requirement: "Complete the task" }] },
+      canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" });
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, runtimeMode: "native", status: "running",
+      nativeIssueId: issueId, nativeSessionId: runId, runnerInstanceId: sourceInstanceId,
+      completionContractId: contractId, completionContractSha256: contractId, contextSnapshot: { issueId } });
+    await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
+    return { companyId, issueId, runId, agentId, sessionId: runId, completionContractId: contractId,
+      completionContractSha256: contractId, sourceInstanceId, controlPlaneSourceInstanceId: "result-worker-test" };
+  }
+  async function complete(binding: Awaited<ReturnType<typeof seed>>, producer: "control_plane" | "coordinator", writer: Db = db, terminalStatus: "succeeded" | "failed" | "cancelled" = "succeeded") {
+    const port = producer === "control_plane" ? new PaperclipControlPlanePort(writer, binding)
+      : new NativeRunCoordinatorStore(writer, { ...binding, normalizedSessionId: binding.sessionId,
+        runnerSourceInstanceId: binding.sourceInstanceId,
+        completionContractRevision: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.contractRevision,
+        completionContractCriterionIds: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.criteria.map(item => item.criterionId) });
+    await port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: { ...CONTROL_PLANE_CONFORMANCE_TERMINAL, runTerminalState: terminalStatus },
+      callerResultId: `${binding.runId}:result` });
+  }
+  async function barrier(binding: Awaited<ReturnType<typeof seed>>) {
+    await workspaceOperationService(db).createRecorder({ companyId, issueId: binding.issueId, heartbeatRunId: binding.runId })
+      .recordOperation({ phase: "workspace_finalize", cwd: root, run: async () => ({ status: "succeeded", exitCode: 0 }) });
+  }
+  async function state(runId: string) {
+    return db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)).then(rows => rows[0]);
+  }
+  async function read(runId: string) {
+    return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0]);
+  }
+  async function accepted(status: "succeeded" | "failed" | "cancelled" = "succeeded") {
+    const binding = await seed(); await complete(binding, "control_plane", db, status); await barrier(binding); return binding;
+  }
+  it("reconstructs a committed decision with an unfinished run projection and stays quiet after repair", async () => {
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    expect((await read(binding.runId)).status).toBe("running");
+    await start(); await flush();
+    expect((await read(binding.runId)).status).toBe("succeeded");
+    expect(emitAgentTaskRun).toHaveBeenCalledTimes(1); await quiet();
+    const settled = await read(binding.runId);
+    await coordinator!.stop(); await start(); await flush(); await quiet();
+    expect((await read(binding.runId)).executionStatusDeliveryId).toBe(settled.executionStatusDeliveryId);
+    expect(emitAgentTaskRun).toHaveBeenCalledTimes(1);
+  });
+  it("ignores routine company-only finalization hints and still repairs a committed run", async () => {
+    await start(); await flush(); await quiet();
+    const previous = db.$client.options.debug;
+    const queries: string[] = [];
+    db.$client.options.debug = (_id, query) => {
+      if (/from\s+"native_run_finalizations"/i.test(query)) queries.push(query);
+    };
+    try {
+      for (let index = 0; index < 3; index += 1) {
+        await db.transaction(async tx => { await notifyNativeFinalizationWork(tx, { companyId }); });
+        await flush();
+      }
+      expect(queries).toEqual([]);
+    } finally { db.$client.options.debug = previous; }
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    await flush();
+    expect((await read(binding.runId)).status).toBe("succeeded");
+    expect(emitAgentTaskRun).toHaveBeenCalledTimes(1);
+    await quiet();
+  });
+
+  it("wakes from the real status decision transaction after startup has become quiet", async () => {
+    await start(); await flush(); await quiet();
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    await flush();
+    expect((await read(binding.runId)).status).toBe("succeeded"); await quiet();
+  });
+  it.each(["commit", "rollback"])("honors the status writer's outer %s", async outcome => {
+    await start(); await flush();
+    const binding = await accepted();
+    const transaction = db.transaction(async tx => {
+      await finalizeNativeRun({ db: tx as unknown as Db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+      await vi.advanceTimersByTimeAsync(5000);
+      expect((await read(binding.runId)).status).toBe("running");
+      if (outcome === "rollback") throw new Error("rollback fixture");
+    });
+    if (outcome === "rollback") await expect(transaction).rejects.toThrow("rollback fixture"); else await transaction;
+    await flush();
+    expect((await read(binding.runId)).status).toBe(outcome === "commit" ? "succeeded" : "running"); await quiet();
+  });
+  it("does not emit another terminal event when recovery projects before the live finalizer", async () => {
+    const binding = await accepted();
+    const commit = statusCommitter.commitNativeStatusDecision;
+    const hook = vi.spyOn(statusCommitter, "commitNativeStatusDecision").mockImplementation(async input => {
+      const outcome = await commit(input);
+      await projectCommittedRun({ db, run: await read(binding.runId), coordinator: await state(binding.runId) });
+      expect((await read(binding.runId)).status).toBe("succeeded");
+      return outcome;
+    });
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    expect(hook).toHaveBeenCalled();
+    expect(emitAgentTaskRun).toHaveBeenCalledTimes(1);
+  });
+  it("uses fresh completion time when the caller's projection snapshot is stale", async () => {
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    const stale = await read(binding.runId);
+    const finishedAt = new Date("2020-01-01T00:00:00Z");
+    await db.update(heartbeatRuns).set({ finishedAt }).where(eq(heartbeatRuns.id, binding.runId));
+    await projectCommittedRun({ db, run: stale, coordinator: await state(binding.runId) });
+    expect((await read(binding.runId)).finishedAt).toEqual(finishedAt);
+  });
+  it.each(["succeeded", "failed", "cancelled"] as const)("stays quiet with retained %s history", async status => {
+    const binding = await accepted(status);
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    const before = await read(binding.runId);
+    expect(before.status).toBe(status);
+    const calls = vi.mocked(emitAgentTaskRun).mock.calls.length;
+    await start(); await flush(); await quiet();
+    expect(await read(binding.runId)).toEqual(before);
+    expect(emitAgentTaskRun).toHaveBeenCalledTimes(calls);
+  });
+  it("serializes competing recovery projections without duplicate terminal events", async () => {
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    const input = { db, run: await read(binding.runId), coordinator: await state(binding.runId) };
+    await Promise.all(Array.from({ length: 4 }, () => projectCommittedRun(input)));
+    expect((await read(binding.runId)).status).toBe("succeeded");
+    expect(emitAgentTaskRun).toHaveBeenCalledTimes(1);
+  });
+  it("repairs projection metadata and preserves unrelated diagnostics", async () => {
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    const before = await read(binding.runId);
+    await db.update(heartbeatRuns).set({ nativePhase: "retryable_failure", error: "old failure", errorCode: "transport_error",
+      resultJson: { ...before.resultJson, failureCode: "old", nextAttemptAt: "old", unrelatedReceipt: { key: "preserve" } },
+    }).where(eq(heartbeatRuns.id, binding.runId));
+    const calls = vi.mocked(emitAgentTaskRun).mock.calls.length;
+    await start(); await flush();
+    expect(await read(binding.runId)).toMatchObject({ status: "succeeded", nativePhase: "committed", error: null, errorCode: null,
+      finishedAt: before.finishedAt, resultJson: { failureCode: null, nextAttemptAt: null,
+        unrelatedReceipt: { key: "preserve" }, recoveredExecutionFailure: { error: "old failure", errorCode: "transport_error" } } });
+    expect(emitAgentTaskRun).toHaveBeenCalledTimes(calls); await quiet();
+  });
+  it.each(["restore", "worktree"])("respects %s suppression through the public heartbeat registration", async mode => {
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    const runtimeEnv = mode === "restore" ? { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true" }
+      : { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "projection-suppression-test" };
+    const settings = instanceSettingsService(db, { runtimeEnv });
+    if (mode === "worktree") await settings.updateExperimental({ enableWorktreeRunExecution: false });
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted,
+      onError: error => errors.push(error) });
+    worker = heartbeatService(db, { runtimeEnv }).registerNativeRunProjectionWork(coordinator, () => admitted);
+    await worker.ready; await flush(); await quiet();
+    expect((await read(binding.runId)).status).toBe("running");
+    if (mode === "worktree") {
+      try {
+        await settings.updateExperimental({ enableWorktreeRunExecution: true });
+        await flush(); expect((await read(binding.runId)).status).toBe("succeeded"); await quiet();
+      } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); }
+    }
+  });
+  it("retains the first committed decision during drain and projects on resume", async () => {
+    await start(); await flush(); admitted = false;
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    await vi.advanceTimersByTimeAsync(120_000); await quiet();
+    expect((await read(binding.runId)).status).toBe("running");
+    admitted = true; await vi.advanceTimersByTimeAsync(30_000); await flush();
+    expect((await read(binding.runId)).status).toBe("succeeded"); await quiet();
+  });
+  it("retains a failed projection across drain and restart", async () => {
+    let fail = true;
+    const project: typeof projectCommittedRun = input => {
+      if (fail) { fail = false; throw new Error("projection unavailable"); }
+      return projectCommittedRun(input);
+    };
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    await start(project);
+    expect(errors).toHaveLength(1); errors.length = 0;
+    expect((await read(binding.runId)).status).toBe("running");
+    expect(coordinator!.nextWakeAt()).not.toBeNull();
+    admitted = false; await vi.advanceTimersByTimeAsync(120_000); await quiet();
+    await coordinator!.stop(); admitted = true; await start(); await flush();
+    expect((await read(binding.runId)).status).toBe("succeeded"); await quiet();
+  });
+  it("joins active projection on shutdown and reconstructs later work", async () => {
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const entered = vi.fn();
+    const project: typeof projectCommittedRun = async input => { entered(); await gate; return projectCommittedRun(input); };
+    await start(project); await flush();
+    const first = await accepted();
+    await finalizeNativeRun({ db, runId: first.runId, workspaceFinalizeStatus: "succeeded" });
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(entered).toHaveBeenCalledTimes(1); });
+    const second = await accepted();
+    await finalizeNativeRun({ db, runId: second.runId, workspaceFinalizeStatus: "succeeded" });
+    let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; });
+    await delay(10); expect(stopped).toBe(false); release(); await stopping;
+    expect((await read(first.runId)).status).toBe("succeeded");
+    expect((await read(second.runId)).status).toBe("running");
+    await start(); await flush(); expect((await read(second.runId)).status).toBe("succeeded"); await quiet();
+  });
+  it("rechecks committed authority after candidate discovery", async () => {
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    const project: typeof projectCommittedRun = async input => {
+      await db.update(nativeRunFinalizations).set({ phase: "retryable_failure" }).where(eq(nativeRunFinalizations.runId, binding.runId));
+      return projectCommittedRun(input);
+    };
+    await start(project); await flush();
+    expect((await read(binding.runId)).status).toBe("running"); expect(emitAgentTaskRun).not.toHaveBeenCalled(); await quiet();
+  });
+  it("waits for real pending-effect settlement before projection", async () => {
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    const committed = await state(binding.runId);
+    const [effect] = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, committed.decisionId!));
+    expect(effect).toBeDefined();
+    await db.update(statusDecisionEffects).set({ deliveryState: "pending", deliveredAt: null }).where(eq(statusDecisionEffects.id, effect.id));
+    await start(); await flush(); await quiet(); expect((await read(binding.runId)).status).toBe("running");
+    await statusCommitter.dispatchPendingNativeStatusEffects({ db, ...binding, decisionId: committed.decisionId!, onlyWhenDue: true });
+    await flush(); expect((await read(binding.runId)).status).toBe("succeeded"); await quiet();
+  });
+  it("wakes from the real lifecycle status writer when an ownership hold is released", async () => {
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    await db.update(heartbeatRuns).set({ nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" })
+      .where(eq(heartbeatRuns.id, binding.runId));
+    await start(); await flush(); await quiet();
+
+
+    const lifecycle = createHeartbeatLifecycle(db, { emitTerminalAgentTaskRun: vi.fn() } as unknown as HeartbeatLifecycleDependencies);
+    await lifecycle.setRunStatus(binding.runId, "running", { errorCode: null, nativePhase: "committed" });
+    await flush(); expect((await read(binding.runId)).status).toBe("succeeded"); await quiet();
+  });
+  it("waits for delegated run projection before committed resource settlement", async () => {
+    const binding = await accepted();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" });
+    const settled = vi.fn(async () => undefined);
+    await reconcileNativeFinalizations(db, [binding.runId], { skipNativeRunProjection: true, skipNativeResultCommit: true,
+      skipNativeResponsePresentation: true, skipNativeReviewResponse: true, onWorkspaceSettled: settled });
+    expect((await read(binding.runId)).status).toBe("running"); expect(settled).not.toHaveBeenCalled();
+    await start(); await flush(); expect((await read(binding.runId)).status).toBe("succeeded"); await quiet();
+    await reconcileNativeFinalizations(db, [binding.runId], { skipNativeRunProjection: true, skipNativeResultCommit: true,
+      skipNativeResponsePresentation: true, skipNativeReviewResponse: true, onWorkspaceSettled: settled });
+    expect(settled).toHaveBeenCalledTimes(1);
+  });
+});
@@ -120,6 +120,7 @@ const {
    registerRetainedNativeSessionCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })),
    registerNativeWorkspaceFinalizationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })),
    registerNativeResultCommitWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
+    registerNativeRunProjectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
    registerNativeStatusReassessmentWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
    registerNativeReviewResponseWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
    registerNativeResponsePresentationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })),
@@ -598,6 +599,13 @@ describe("startServer feedback export wiring", () => {
    }
  });

+  it.each([false, true])("registers native run projection only when maintenance is enabled (%s)", async enabled => {
+    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
+    await startServer(); await settleStartupHousekeeping();
+    expect(heartbeatServiceMock.registerNativeRunProjectionWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
+    if (enabled) expect(heartbeatServiceMock.registerNativeRunProjectionWork.mock.calls[0][1]()).toBe(true);
+  });
+
  it.each([false, true])("registers native result commit only when maintenance is enabled (%s)", async enabled => {
    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
    await startServer(); await settleStartupHousekeeping();
@@ -1066,7 +1074,7 @@ describe("startServer feedback export wiring", () => {
      intervalCallback?.();
      await new Promise<void>((resolve) => setImmediate(resolve));
      expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
-      expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true });
+      expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true });
      expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
      expect(retiredDetector).not.toHaveBeenCalled();
    } finally {
@@ -1222,15 +1230,17 @@ describe("startServer feedback export wiring", () => {
    }
  });

-  it.each(["retained cleanup", "workspace finalization", "native result commit"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => {
+  it.each(["retained cleanup", "workspace finalization", "native run projection", "native result commit"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => {
    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true }));
    let releaseCleanup!: () => void;
    const ready = new Promise<void>(resolve => { releaseCleanup = resolve; });
    const stop = vi.fn();
    const register = owner === "retained cleanup" ? heartbeatServiceMock.registerRetainedNativeSessionCleanupWork
      : owner === "workspace finalization" ? heartbeatServiceMock.registerNativeWorkspaceFinalizationWork
+      : owner === "native run projection" ? heartbeatServiceMock.registerNativeRunProjectionWork
      : heartbeatServiceMock.registerNativeResultCommitWork;
-    if (owner === "native result commit") heartbeatServiceMock.registerNativeResultCommitWork.mockReturnValueOnce({ ready });
+    if (owner === "native run projection") heartbeatServiceMock.registerNativeRunProjectionWork.mockReturnValueOnce({ ready });
+    else if (owner === "native result commit") heartbeatServiceMock.registerNativeResultCommitWork.mockReturnValueOnce({ ready });
    else (owner === "retained cleanup" ? heartbeatServiceMock.registerRetainedNativeSessionCleanupWork
      : heartbeatServiceMock.registerNativeWorkspaceFinalizationWork).mockReturnValueOnce({ ready, stop });
    const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(() => 1 as unknown as ReturnType<typeof setInterval>);
@@ -1247,7 +1257,7 @@ describe("startServer feedback export wiring", () => {
      const server = await starting;
      stopping = server.shutdown().then(() => { stopped = true; });
      await settleStartupHousekeeping();
-      if (owner !== "native result commit") expect(stop).toHaveBeenCalledOnce();
+      if (owner === "retained cleanup" || owner === "workspace finalization") expect(stop).toHaveBeenCalledOnce();
      expect(stopped).toBe(false);
    } finally {
      releaseCleanup();
@@ -1842,6 +1842,13 @@ async function startServerWithDatabaseTeardown(
      server.on("close", worker.stop);


+      trackHeartbeatSchedulerWork(worker.ready);
+    }
+    if (config.heartbeatSchedulerEnabled) {
+      const worker = heartbeat.registerNativeRunProjectionWork(deliveryWork,
+        () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby());
+
+
      trackHeartbeatSchedulerWork(worker.ready);
    }
    if (config.heartbeatSchedulerEnabled) {
@@ -2001,7 +2008,7 @@ async function startServerWithDatabaseTeardown(
          // Periodically reap orphaned runs (5-min staleness threshold) and make sure
          // persisted queued work is still being driven forward.
          runPeriodicRecovery(() => heartbeat
-            .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true })
+            .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true })
            .then(() => heartbeat.promoteDueScheduledRetries())
            .then(async (promotion) => {
              await heartbeat.resumeQueuedRuns();
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";

 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  nativeRunProjection: "native-run-projection",
  nativeResultCommit: "native-result-commit",
  nativeWorkspaceFinalization: "native-workspace-finalization",
  retainedNativeSessionCleanup: "retained-native-session-cleanup",
@@ -1,3 +1,4 @@
+import { nativeRunProjectionWork } from "./native-runtime/native-run-projection-work.js";
 import { nativeResultCommitWork } from "./native-runtime/native-result-commit-work.js";
 import { nativeStatusReassessmentWork } from "./native-runtime/native-status-reassessment-work.js";
 import { registerNativeReviewResponseWork } from "./native-runtime/native-review-response-work.js";
@@ -4095,6 +4096,14 @@ export function heartbeatService(
    triggerIssueMonitor,

    reportRunActivity: clearDetachedRunWarning,
+    registerNativeRunProjectionWork: (
+      coordinator: Parameters<ReturnType<typeof nativeRunProjectionWork>["register"]>[0], canRun: () => boolean,
+    ) => nativeRunProjectionWork(db).register(coordinator,
+      () => canRun() && readTaskDrain(new Date()) === null,
+      async () => {
+        const state = await getSchedulingSuppression(true);
+        return !state.suppressed || state.reason === "task_drain";
+      }),
    registerNativeResultCommitWork: (
      coordinator: Parameters<ReturnType<typeof nativeResultCommitWork>["register"]>[0], canRun: () => boolean,
    ) => nativeResultCommitWork(db).register(coordinator,
@@ -1967,7 +1967,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD

  let settledProcessesDiscovered = false;

-  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean }) {
+  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean }) {
    if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db);
    const staleThresholdMs = opts?.staleThresholdMs ?? 0;
    const now = new Date();
@@ -1994,6 +1994,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
      skipNativeStatusReassessment: opts?.skipNativeStatusReassessment,
      skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization,
      skipNativeResultCommit: opts?.skipNativeResultCommit,
+      skipNativeRunProjection: opts?.skipNativeRunProjection,
      environmentRuntime,
      pluginWorkerManager: dependencies.pluginWorkerManager,
      onWorkspaceSettled: settleRecoveredNativeWorkspace,
@@ -1,3 +1,4 @@
+import { notifyNativeRunProjection } from "../native-run-projection-notifications.js";
 import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js";
 import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js";
@@ -1018,13 +1019,14 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
    return recordHeartbeatRunRuntimeProgress(currentRun, update, issueId);
  }

-  async function writeRunStatus<T>(status: string, previous: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "status" | "nativeIssueId" | "issueId" | "contextSnapshot"> | null,
+  async function writeRunStatus<T>(status: string, previous: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "status" | "runtimeMode" | "nativeIssueId" | "issueId" | "contextSnapshot"> | null,
    write: (tx: Db) => Promise<T>) {
    const companyId = previous?.companyId;
    // Register on the outer transaction before the terminal write. An unknown
    // commit cannot strand newly-actionable accounting after the live event.
    return db.transaction(async transaction => {
      const tx = transaction as unknown as Db;
+      if (previous?.runtimeMode === "native") await notifyNativeRunProjection(tx, { companyId: previous.companyId, runId: previous.id });
      if (previous && previous.status !== status) await notifyUnsafeWorkspaceExportRunChange(tx, previous);
      if (previous && (previous.status === "scheduled_retry" || status === "scheduled_retry"))
        await notifyScheduledRetryCancellation(tx, { companyId: previous.companyId, runId: previous.id });
@@ -1,3 +1,4 @@
+import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
 import { notifyNativeResultCommit } from "./native-result-commit-notifications.js";
 import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
@@ -628,6 +629,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
          await notifyDirectoryCleanup(tx);
          await notifyInstructionCollection(tx);
          await notifyCancelledWorkspacePreservation(tx);
+          await notifyNativeRunProjection(tx);
          await notifyNativeResultCommit(tx);
          await notifyNativeWorkspaceFinalization(tx);
        }
@@ -1,3 +1,4 @@
+import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
 import { notifyNativeResultCommit } from "./native-result-commit-notifications.js";
 import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
@@ -11504,6 +11505,7 @@ export function issueService(db: Db) {
        if (patch.status !== undefined || patch.statusVersion !== undefined || patch.executionState !== undefined
          || patch.executionRunId !== undefined || patch.checkoutRunId !== undefined
          || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || blockedByIssueIds !== undefined) {
+          await notifyNativeRunProjection(tx, { companyId: receiptExisting.companyId, issueId: id });
          await notifyNativeResultCommit(tx, { companyId: receiptExisting.companyId, issueId: id });
          await notifyNativeWorkspaceFinalization(tx, { companyId: receiptExisting.companyId, issueId: id });
          await notifyNativeStatusReassessment(tx, { companyId: receiptExisting.companyId, issueId: id });
@@ -1,3 +1,4 @@
+import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
 import { notifyNativeResultCommit } from "./native-result-commit-notifications.js";
 import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
 import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js";
@@ -10,6 +11,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i

 export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) {
  if (scope.issueId || scope.runId) {
+    await notifyNativeRunProjection(transaction, scope);
    await notifyNativeResultCommit(transaction, scope);
    await notifyNativeWorkspaceFinalization(transaction, scope);
    await notifyRetainedNativeSessionCleanup(transaction, scope);
@@ -0,0 +1,8 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type NativeRunProjectionScope = { companyId: string; issueId?: string; runId?: string };
+
+export function notifyNativeRunProjection(tx: object, scope?: NativeRunProjectionScope) {
+  return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeRunProjection,
+    scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId }) : "*");
+}
@@ -1,3 +1,4 @@
+import { pendingNativeRunProjectionCondition } from "./native-run-projection.js";
 import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js";
 import { resolveNativeReconciliationStatus } from "./native-reconciliation-status.js";
 export { resolveNativeReconciliationStatus, type NativeReconciliationFacts } from "./native-reconciliation-status.js";
@@ -389,6 +390,7 @@ type NativeReconciliationOptions = {
  skipNativeStatusReassessment?: boolean;
  skipNativeWorkspaceFinalization?: boolean;
  skipNativeResultCommit?: boolean;
+  skipNativeRunProjection?: boolean;
  pluginWorkerManager?: WorkspaceWorkers;
  environmentRuntime?: EnvironmentRuntimeService;
  onWorkspaceSettled?: (input: {
@@ -507,6 +509,7 @@ async function reconcileNativeFinalizationsPass(
      issueStatusVersion: issues.statusVersion,
      issueDecisionId: issues.lastStatusDecisionId,
      coordinatorPhase: nativeRunFinalizations.phase,
+      runProjectionPending: options.skipNativeRunProjection ? pendingNativeRunProjectionCondition(db) : sql<boolean>`false`,
      resultId: nativeRunFinalizations.resultId,
      assessmentId: nativeRunFinalizations.assessmentId,
      decisionId: nativeRunFinalizations.decisionId,
@@ -618,6 +621,9 @@ async function reconcileNativeFinalizationsPass(
        }
      }
    }
+
+
+    if (options.skipNativeRunProjection && row.coordinatorPhase === "committed" && row.runProjectionPending) continue;
    const barrier = await db.select({ status: workspaceOperations.status })
      .from(workspaceOperations)
      .where(and(
@@ -681,6 +687,7 @@ async function reconcileNativeFinalizationsPass(
          projectRunStatus: true,
          skipNativeResponsePresentation: options.skipNativeResponsePresentation,
          skipNativeReviewResponse: options.skipNativeReviewResponse,
+        skipCommittedRunProjection: options.skipNativeRunProjection,
          preserveProviderAttempt: Boolean(
            readNativeWorkspaceSyncReference(
              row.nativeWorkspaceSync,
@@ -742,6 +749,7 @@ async function reconcileNativeFinalizationsPass(
        projectRunStatus: true,
        skipNativeResponsePresentation: options.skipNativeResponsePresentation,
        skipNativeReviewResponse: options.skipNativeReviewResponse,
+        skipCommittedRunProjection: options.skipNativeRunProjection,
        preserveProviderAttempt: Boolean(
          readNativeWorkspaceSyncReference(
            row.nativeWorkspaceSync,
@@ -447,7 +447,7 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a
        selection: any,
      ) => {
        const query = select(selection);
-        return selection?.resultJson === nativeRunResults.resultJson
+        return selection?.terminalState
          ? wrap(query)
          : query;
      }) as typeof db.select);
@@ -1,3 +1,5 @@
+import { notifyNativeRunProjection } from "../native-run-projection-notifications.js";
+import { projectCommittedRun, recoveredExecutionFailureMetadata } from "./native-run-projection.js";
 import { notifyNativeResultCommit } from "../native-result-commit-notifications.js";
 import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js";
 import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
@@ -93,7 +95,6 @@ import {
 } from "../durable-chat-wakeup.js";
 import {
  isNativeRunnerOwnershipHeld,
-  nativeRunnerOwnershipNotHeldCondition,
 } from "./native-runner-ownership.js";

 function record(value: unknown): Record<string, unknown> {
@@ -137,12 +138,7 @@ function isCommittedAuditOnlyRouting(value: unknown) {
  );
 }

-/** Server-owned terminal conversion used by live finalization and read models. */
-export function projectNativeTerminalRunStatus(
-  terminalState: "succeeded" | "failed" | "cancelled" | "active",
-) {
-  return terminalState === "active" ? ("running" as const) : terminalState;
-}
+export { projectNativeTerminalRunStatus } from "./native-run-projection.js";

 /** Fact-based arbitration at the production finalization authority seam. */
 export function resolveNativeFinalizerStatus(
@@ -614,128 +610,6 @@ async function resolveCommittedFinalizationRecovery(db: Db, run: Pick<typeof hea
  });
 }

-function recoveredExecutionFailureMetadata() {
-  // Read the row being updated so a concurrent cleanup diagnostic is retained.
-  return sql`case when ${heartbeatRuns.error} is not null or ${heartbeatRuns.errorCode} is not null
-    then jsonb_build_object('recoveredExecutionFailure', jsonb_build_object(
-      'schema', 'paperclip.recovered_execution_failure.v1',
-      'errorCode', ${heartbeatRuns.errorCode}, 'error', ${heartbeatRuns.error},
-      'observedAt', ${heartbeatRuns.updatedAt}
-    )) else '{}'::jsonb end`;
-}
-
-async function projectCommittedRun(input: {
-  db: Db;
-  run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "status" | "finishedAt" | "runtimeMode" | "nativePhase" | "nativeIssueId" | "contextSnapshot">;
-  coordinator: typeof nativeRunFinalizations.$inferSelect;
-}) {
-  if (!input.coordinator.resultId) return;
-  const resultRow = await input.db
-    .select({ resultJson: nativeRunResults.resultJson })
-    .from(nativeRunResults)
-    .where(
-      and(
-        eq(nativeRunResults.id, input.coordinator.resultId),
-        eq(nativeRunResults.runId, input.run.id),
-        eq(nativeRunResults.companyId, input.run.companyId),
-      ),
-    )
-    .limit(1)
-    .then((rows) => rows[0] ?? null);
-  const terminalState = record(
-    record(resultRow?.resultJson).terminal,
-  ).runTerminalState;
-  if (!["succeeded", "failed", "cancelled"].includes(String(terminalState))) {
-    throw new Error("native_finalization_invalid");
-  }
-  const projectedStatus = projectNativeTerminalRunStatus(
-    terminalState as "succeeded" | "failed" | "cancelled",
-  );
-  const needsRepair = and(
-    eq(heartbeatRuns.id, input.run.id),
-    eq(heartbeatRuns.runtimeMode, "native"),
-    or(
-      inArray(heartbeatRuns.status, ["queued", "running", "failed"]),
-      and(
-        eq(heartbeatRuns.status, "succeeded"),
-        or(
-          isNotNull(heartbeatRuns.error),
-          isNotNull(heartbeatRuns.errorCode),
-          isNull(heartbeatRuns.finishedAt),
-          sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
-          sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
-          sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
-          sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
-          sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
-        ),
-      ),
-    ),
-    // Reconciliation revisits committed results periodically. Only repair
-    // a changed projection; rewriting an unchanged failed run would mint a
-    // fresh status delivery (and failure toast) on every sweep. Check the
-    // current row so concurrent replays cannot both queue the same repair.
-    or(
-      sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`,
-      isNull(heartbeatRuns.finishedAt),
-      sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
-      sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
-      sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
-      sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
-      sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
-      ...(terminalState === "succeeded"
-        ? [isNotNull(heartbeatRuns.error), isNotNull(heartbeatRuns.errorCode)]
-        : []),
-    ),
-    nativeRunnerOwnershipNotHeldCondition(),
-  );
-  const [candidate] = await input.db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(needsRepair).limit(1);
-  if (!candidate) return;
-  const now = new Date();
-  const [updatedRun] = await input.db.transaction(async transaction => {
-    const tx = transaction as unknown as Db;
-    await notifyRetainedNativeSessionCleanup(tx, { companyId: input.run.companyId, runId: input.run.id });
-    await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id });
-    await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id });
-    await notifyUnsafeWorkspaceExportRunChange(tx, input.run);
-    await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
-    await notifyExecutionRunWork(tx);
-    await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
-    return tx
-      .update(heartbeatRuns)
-      .set({
-        executionStatusDeliveryId: randomUUID(),
-        status: projectedStatus,
-        finishedAt: input.run.finishedAt ?? now,
-        nativePhase: "committed",
-        nativePhaseUpdatedAt: now,
-        ...(terminalState === "succeeded"
-          ? {
-              error: null,
-              errorCode: null,
-              // Capture the row being updated, not the earlier admission read:
-              // cleanup may have recorded a new diagnostic in the meantime.
-              // Other result metadata and all physical-owner evidence stay put.
-              resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
-                || ${recoveredExecutionFailureMetadata()}
-                || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)`,
-            }
-          : { resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)` }),
-        updatedAt: now,
-      })
-      .where(needsRepair)
-      .returning();
-  });
-  // Metadata repairs can preserve the terminal status. Only a genuine status
-  // transition should emit another terminal event. A failed run that leaves a
-  // pending retry still needs its first failure report.
-  if (updatedRun && updatedRun.status !== input.run.status) {
-    await emitAgentTaskRun(input.db, updatedRun);
-    void reportRunFailure(input.db, updatedRun);
-  } else if (updatedRun && isNewTerminalRunFailure(input.run, updatedRun)) {
-    void reportRunFailure(input.db, updatedRun);
-  }
-}
-
 async function materializeNativeReviewResponse(db: Db, runId: string, dbOrTx: Db = db) {
  const [run] = await dbOrTx
    .select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
@@ -1225,6 +1099,7 @@ export async function finalizeNativeRun(input: {
  projectRunStatus?: boolean;
  skipNativeResponsePresentation?: boolean;
  skipNativeReviewResponse?: boolean;
+  skipCommittedRunProjection?: boolean;
  /** Workspace-only replay must not consume the provider recovery budget. */
  preserveProviderAttempt?: boolean;
  failpoint?: NativeStatusCommitFailpoint;
@@ -1275,7 +1150,7 @@ export async function finalizeNativeRun(input: {
        }),
      );
    }
-    if (input.projectRunStatus)
+    if (input.projectRunStatus && !input.skipCommittedRunProjection)
      await projectCommittedRun({ db: input.db, run: admittedRun, coordinator });
    if (input.projectRunStatus && !presentationAlreadyMaterialized)
      await materializeCommittedReviewResponse(input.db, input.runId);
@@ -1760,8 +1635,17 @@ export async function finalizeNativeRun(input: {
        })
        .where(eq(heartbeatRuns.id, run.id))
        .returning();
-      const [updatedRun] = await input.db.transaction(async transaction => {
+      const projection = await input.db.transaction(async transaction => {
        const tx = transaction as unknown as Db;
+
+
+
+        const [before] = await tx.select({ status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode,
+          nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt })
+          .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId)))
+          .for("update").limit(1);
+        if (!before) return null;
+        await notifyNativeRunProjection(tx, { companyId: run.companyId, runId: run.id });
        await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id });
        if (input.projectRunStatus) await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
        if (input.projectRunStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
@@ -1769,15 +1653,20 @@ export async function finalizeNativeRun(input: {
        if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
        await notifyExecutionRunWork(tx);
        await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId });
-        return projectRun(tx);
+        const [after] = await projectRun(tx);
+        return after ? { before, after } : null;
      });
+      const updatedRun = projection?.after;
      if (
        input.projectRunStatus &&
        !alreadyEmittedByCommittedDecision &&
-        updatedRun
+        projection && projection.after.status !== projection.before.status
      ) {
-        await emitAgentTaskRun(input.db, updatedRun);
-        void reportRunFailure(input.db, updatedRun);
+        await emitAgentTaskRun(input.db, projection.after);
+        void reportRunFailure(input.db, projection.after);
+      } else if (input.projectRunStatus && !alreadyEmittedByCommittedDecision && projection
+        && isNewTerminalRunFailure(projection.before, projection.after)) {
+        void reportRunFailure(input.db, projection.after);
      }
      if (input.projectRunStatus)
        await materializeCommittedReviewResponse(input.db, input.runId);
@@ -0,0 +1,75 @@
+import { and, asc, eq, gt, not, sql } from "drizzle-orm";
+import { heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import type { NativeRunProjectionScope } from "../native-run-projection-notifications.js";
+import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
+import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js";
+import { pendingNativeRunProjectionCondition, projectCommittedRun } from "./native-run-projection.js";
+
+type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
+
+
+
+export function nativeRunProjectionWork(db: Db, project = projectCommittedRun) {
+  function candidates(scope?: NativeRunProjectionScope, after?: string) {
+    return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: nativeRunFinalizations.issueId })
+      .from(nativeRunFinalizations)
+      .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
+        eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId)))
+      .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId)))
+      .where(and(eq(nativeRunFinalizations.phase, "committed"),
+        pendingNativeRunProjectionCondition(db),
+        not(pendingNativeStatusEffectsCondition(db)),
+        scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
+        scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined,
+        scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
+        after ? gt(heartbeatRuns.id, after) : undefined,
+      )).orderBy(asc(heartbeatRuns.id)).limit(100);
+  }
+  async function load(scope?: NativeRunProjectionScope, signal?: AbortSignal, canRun = () => true): Promise<Entry[]> {
+    const entries: Entry[] = [];
+    let after: string | undefined;
+    while (canRun() && !signal?.aborted) {
+      const rows = await candidates(scope, after);
+      entries.push(...rows.map(row => ({ ...row, dueAt: 0 })));
+      if (rows.length < 100) return entries;
+      after = rows[rows.length - 1].id;
+    }
+    throw new Error("Native run projection discovery stopped");
+  }
+  async function check(entry: Entry, signal?: AbortSignal): Promise<Entry | null> {
+    if (signal?.aborted) throw new Error("Native run projection stopped");
+    const scope = { companyId: entry.companyId, issueId: entry.issueId, runId: entry.id };
+    const [pending] = await candidates(scope);
+    if (!pending) return null;
+    const [run] = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
+      status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt, nativePhase: heartbeatRuns.nativePhase,
+      runtimeMode: heartbeatRuns.runtimeMode, nativeIssueId: heartbeatRuns.nativeIssueId,
+      contextSnapshot: sql<Record<string, unknown>>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->>'issueId')`,
+    }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, entry.id), eq(heartbeatRuns.companyId, entry.companyId)));
+    const [coordinator] = await db.select({ resultId: nativeRunFinalizations.resultId, issueId: nativeRunFinalizations.issueId })
+      .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.runId, entry.id),
+        eq(nativeRunFinalizations.companyId, entry.companyId), eq(nativeRunFinalizations.issueId, entry.issueId),
+        eq(nativeRunFinalizations.phase, "committed")));
+    if (!run || !coordinator) return null;
+    if (signal?.aborted) throw new Error("Native run projection stopped");
+    await project({ db, run, coordinator, onlyWhenEffectsSettled: true });
+    return (await candidates(scope)).length ? { ...entry, dueAt: Date.now() + 30_000 } : null;
+  }
+  function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean,
+    canRecover: () => Promise<boolean> = async () => true) {
+    return registerScopedDeadlineWork(coordinator, {
+      queue: DELIVERY_QUEUES.nativeRunProjection, batchSize: 25,
+      decodeScope: (hint: string): NativeRunProjectionScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (entry: Entry, scope?: NativeRunProjectionScope) => !scope || entry.companyId === scope.companyId
+        && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId),
+      load: async (scope, signal) => await canRecover() ? load(scope, signal, canRun) : [],
+      check: async (entry, signal) => {
+        if (!canRun()) throw new Error("Native run projection paused");
+        return await canRecover() ? check(entry, signal) : null;
+      },
+    }, canRun);
+  }
+  return { load, check, register };
+}
@@ -0,0 +1,172 @@
+import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js";
+import { notifyNativeRunProjection } from "../native-run-projection-notifications.js";
+import { randomUUID } from "node:crypto";
+import { and, eq, exists, inArray, isNotNull, isNull, not, or, sql, type SQL } from "drizzle-orm";
+import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
+import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js";
+import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
+import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js";
+import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
+import { notifyExecutionRunWork } from "../execution-control-notifications.js";
+import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
+import { emitAgentTaskRun } from "../agent-task-run-telemetry.js";
+import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js";
+import { nativeRunnerOwnershipNotHeldCondition } from "./native-runner-ownership.js";
+
+
+export function projectNativeTerminalRunStatus(
+  terminalState: "succeeded" | "failed" | "cancelled" | "active",
+) {
+  return terminalState === "active" ? ("running" as const) : terminalState;
+}
+
+
+
+export function nativeRunProjectionNeedsRepair(projectedStatus: "succeeded" | "failed" | "cancelled" | SQL) {
+  return and(
+    eq(heartbeatRuns.runtimeMode, "native"),
+    or(
+      inArray(heartbeatRuns.status, ["queued", "running", "failed"]),
+      and(
+        eq(heartbeatRuns.status, "succeeded"),
+        or(
+          isNotNull(heartbeatRuns.error),
+          isNotNull(heartbeatRuns.errorCode),
+          isNull(heartbeatRuns.finishedAt),
+          sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
+          sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
+          sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
+          sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
+          sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
+        ),
+      ),
+    ),
+
+
+
+
+    or(
+      sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`,
+      isNull(heartbeatRuns.finishedAt),
+      sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
+      sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
+      sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
+      sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
+      sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
+      and(sql`${projectedStatus} = 'succeeded'`, or(isNotNull(heartbeatRuns.error), isNotNull(heartbeatRuns.errorCode))),
+    ),
+    nativeRunnerOwnershipNotHeldCondition(),
+  );
+}
+
+
+
+export function pendingNativeRunProjectionCondition(db: Db) {
+  const status = sql<string>`${nativeRunResults.resultJson}->'terminal'->>'runTerminalState'`;
+  return exists(db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and(
+    eq(nativeRunResults.id, nativeRunFinalizations.resultId),
+    eq(nativeRunResults.runId, nativeRunFinalizations.runId),
+    eq(nativeRunResults.companyId, nativeRunFinalizations.companyId),
+    eq(nativeRunResults.issueId, nativeRunFinalizations.issueId),
+    eq(nativeRunResults.schemaStatus, "accepted"),
+    inArray(status, ["succeeded", "failed", "cancelled"]),
+    nativeRunProjectionNeedsRepair(status),
+  )));
+}
+
+export function recoveredExecutionFailureMetadata() {
+
+  return sql`case when ${heartbeatRuns.error} is not null or ${heartbeatRuns.errorCode} is not null
+    then jsonb_build_object('recoveredExecutionFailure', jsonb_build_object(
+      'schema', 'paperclip.recovered_execution_failure.v1',
+      'errorCode', ${heartbeatRuns.errorCode}, 'error', ${heartbeatRuns.error},
+      'observedAt', ${heartbeatRuns.updatedAt}
+    )) else '{}'::jsonb end`;
+}
+
+export async function projectCommittedRun(input: {
+  db: Db;
+  onlyWhenEffectsSettled?: boolean;
+  run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "status" | "finishedAt" | "runtimeMode" | "nativePhase" | "nativeIssueId" | "contextSnapshot">;
+  coordinator: Pick<typeof nativeRunFinalizations.$inferSelect, "resultId" | "issueId">;
+}) {
+  if (!input.coordinator.resultId) return;
+  const resultRow = await input.db
+    .select({ terminalState: sql<string>`${nativeRunResults.resultJson}->'terminal'->>'runTerminalState'` })
+    .from(nativeRunResults)
+    .where(
+      and(
+        eq(nativeRunResults.id, input.coordinator.resultId),
+        eq(nativeRunResults.runId, input.run.id),
+        eq(nativeRunResults.companyId, input.run.companyId),
+        eq(nativeRunResults.issueId, input.coordinator.issueId),
+        eq(nativeRunResults.schemaStatus, "accepted"),
+      ),
+    )
+    .limit(1)
+    .then((rows) => rows[0] ?? null);
+  const terminalState = resultRow?.terminalState;
+  if (!["succeeded", "failed", "cancelled"].includes(String(terminalState))) {
+    throw new Error("native_finalization_invalid");
+  }
+  const projectedStatus = terminalState as "succeeded" | "failed" | "cancelled";
+  const needsRepair = and(eq(heartbeatRuns.id, input.run.id), eq(heartbeatRuns.companyId, input.run.companyId),
+    eq(heartbeatRuns.nativeIssueId, input.coordinator.issueId), nativeRunProjectionNeedsRepair(projectedStatus),
+    exists(input.db.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and(
+      eq(nativeRunFinalizations.runId, input.run.id), eq(nativeRunFinalizations.companyId, input.run.companyId),
+      eq(nativeRunFinalizations.issueId, input.coordinator.issueId), eq(nativeRunFinalizations.resultId, input.coordinator.resultId),
+      eq(nativeRunFinalizations.phase, "committed"),
+      input.onlyWhenEffectsSettled ? not(pendingNativeStatusEffectsCondition(input.db)) : undefined,
+    ))));
+  const now = new Date();
+  const projected = await input.db.transaction(async transaction => {
+    const tx = transaction as unknown as Db;
+    const [before] = await tx.select({ status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode,
+      nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt })
+      .from(heartbeatRuns).where(needsRepair).for("update").limit(1);
+    if (!before) return null;
+    await notifyNativeRunProjection(tx, { companyId: input.run.companyId, runId: input.run.id });
+    await notifyRetainedNativeSessionCleanup(tx, { companyId: input.run.companyId, runId: input.run.id });
+    await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id });
+    await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id });
+    await notifyUnsafeWorkspaceExportRunChange(tx, input.run);
+    await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
+    await notifyExecutionRunWork(tx);
+    await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
+    const [after] = await tx
+      .update(heartbeatRuns)
+      .set({
+        executionStatusDeliveryId: randomUUID(),
+        status: projectedStatus,
+        finishedAt: before.finishedAt ?? now,
+        nativePhase: "committed",
+        nativePhaseUpdatedAt: now,
+        ...(terminalState === "succeeded"
+          ? {
+              error: null,
+              errorCode: null,
+
+
+
+              resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
+                || ${recoveredExecutionFailureMetadata()}
+                || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)`,
+            }
+          : { resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)` }),
+        updatedAt: now,
+      })
+      .where(needsRepair)
+      .returning();
+    return after ? { before, after } : null;
+  });
+
+
+
+  if (projected && projected.after.status !== projected.before.status) {
+    await emitAgentTaskRun(input.db, projected.after);
+    void reportRunFailure(input.db, projected.after);
+  } else if (projected && isNewTerminalRunFailure(projected.before, projected.after)) {
+    void reportRunFailure(input.db, projected.after);
+  }
+}