fix(server): resolve native finalization notices from committed wakes (#17758)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native finalization can commit before its recovery notice is resolved. > - The periodic orphan pass revisits committed runs to resolve those notices. > - Retained completed history should not require repeated discovery. > - This PR moves that one duty to transaction wakes and concrete failure retries. > - Existing notice rows reconstruct work after restart. The action service still owns resolution. > - Other completion and resource duties keep their current recovery paths. ## Linked Issues or Issue Description Refs #17752 (stack parent: committed run projection), #17748 (accepted-result commit), and #17734 (workspace finalization). These are distinct recovery obligations. This slice incorporates parent #17752 at `a49a2dac13b0e1c19da5d3d6b435c781703b1c0b`, now merged as `50fe32684d2d4e1db129e3c1bb455571344a0c09`. Related #17724 retires obsolete exact-session notices for terminal runs with no result and preserves their replay hold. This slice resolves notices after a result committed; it does not change that no-replay policy. **Current behavior** Committed native finalization recovery repeatedly inspects active watchdog notices. A notice can also be updated in place to describe a different run, so resolving a previously selected ID alone is unsafe. **Proposed behavior** Wake a scoped worker from relevant committed writes. Reconstruct active finalization notices at startup. Resolve only the current notice after the native owner rechecks its binding under lifecycle locks. Stop scheduled work when no matching obligation remains. **Reason and benefit** Notice recovery becomes quiet after settlement and reacts to real changes without a polling delay. Reused notices remain attached to their current incident. This does not claim that every remaining server duty or the host can sleep yet. ## What Changed - Track the initial recovery-notice sweep through shutdown without blocking HTTP readiness on task locks. - Require issue/run scope for ordinary finalization fanout. Routine company-only hints leave dormant notices quiet; dedicated owner and explicit settings signals retain their scopes. - Extract the native finalization notice predicate and resolution owner. Keep the existing recovery-action service responsible for writing the resolution. - Bind candidates by company, issue and run. Require a committed result and an active or escalated finalization notice. Exclude held ownership, pending status effects and unfinished run projection. - Recheck coordinator, issue, run and matching actions under locks. A newer incident cannot be resolved from a stale candidate snapshot. - Use the shared scoped worker, transaction signals, retry backoff and lifecycle handling. Add no table, per-feature timer or provider branch. - Signal action insertion, update, supersession and resolution; finalization and effect changes; run projection and ownership release; issue changes; settings activation; and the recovery route that reopens an action. - Preserve no-op resolution quiet. Serialize only the three supported notification IDs, even when a caller supplies a larger internal object. - Delegate only notice resolution from the periodic path. Keep manual/default finalization and other completion/resource work. - Add real PostgreSQL producer, concurrency and lifecycle coverage and a detailed audit. ## Verification - Current local source `730e00a7b007a36d188ca0a328ef050d3a25d1ee` incorporates the parent with a normal merge. All 1,209 parent-only, eight owned-only and 12 automatic shared paths are preserved. The two manual paths retain both chat import sets and scope the new notice fanout. The entire package tree matches parent profile 80 byte-for-byte. - Controlled real-database regression: original notice-only unscoped fanout produces three candidate SELECTs for routine company hints. The same zero-query assertion passes after restoring scoped fanout. The real notice writer then creates and resolves an action for a committed run and returns to quiet. Both phases restore production bytes exactly; the 22 cases outside this focused pattern are not counted as passing there. - The complete main command passes **1,873 cases in 21 files**, without skips, in 309.69 seconds. It includes all 23 notice-worker cases, full presentation/ownership read coverage, 88 startup cases, and the projection/result/physical-work consumers. Process recovery passes 410 in 163.54 seconds and external-chat response passes 273 in 53.48 seconds. Total complete-file acceptance: **2,556 cases in 23 complete files**, with no skips. The additional focused chat selection passes 18 cases in 13.44 seconds; 1,796 unselected chat cases do not count. Total selected passes: **2,574**. Normal merge `a0f2b3f4348b078219a53d8cc135c84202ba505e` adopts actual parent ancestry with this complete tested child tree unchanged. - The chat selection covers the real reply/notice row-lock regression, Slack and Telegram response recovery, current-authority denial cases, exact canonical response binding and internal GitHub attachment navigation. Other cases in the giant chat file remain unselected; this is not full chat-suite acceptance. - Production server typecheck, build, module boundaries, migration order and source identity checks pass. Twenty dedicated PostgreSQL 17 databases use the unchanged migration 413 schema. Embedded-only invocations use the existing fresh-per-call recovery helper. All temporary imports restore exactly and all 24 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 existing opt-in closed Pi snapshot case is skipped because its distribution root is unset. This does not qualify a live provider. - Failure history and scope are retained in `doc/plans/2026-10-26-native-recovery-notice-sleep-audit.md`. Current-head acceptance: all 57 checks are clean, including protected CI and security. Greptile gives 5/5 on `2a8777bbed8b1e10810e3d6d99aae3f25b4854ef`, with no actionable findings. Prior scores do not substitute. No Ponytail verdict is present under the authorized missing-review policy. No full local repository suite, deployed sleep, expanded Cloud wake coverage or measured savings is claimed. ## Risks - Resolution takes coordinator, issue, run and action locks in the native recovery order. Contention must preserve the current incident; the real concurrent-upsert and competing-resolver cases exercise this boundary. - Process-local signals cover instrumented application writers. They do not add support for independent SQL writers or replicas without supported ingress. Startup reconstructs the existing durable rows. - Completed finalization is not proof that external conversation settlement, Muse reconciliation or physical cleanup finished. Those obligations retain their owners and recovery paths. - Discovery uses narrow keyset pages. No fleet benchmark, host suspension or cost reduction is claimed. Cloud wake coverage remains unchanged. - Parent #17752 has merged. This PR targets master at `50fe32684d2d4e1db129e3c1bb455571344a0c09`. Fresh 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-10-30 05:16:50 +01:00 · 2026-10-29 05:35:01 -05:00
1 parent 50fe32684d
commit 8ac5a128e6
@@ -3270,3 +3270,20 @@ obligations. Workspace typechecking passes; the full local suite remains limited
 by embedded PostgreSQL startup capacity. Final local
 and remote acceptance is pending. Master refreshed to `ad861a7f94`; stack integration,
 prior failed gates, suppression audits and all remaining duties stay in scope.
+
+## Slice 54 accepted; slice 55 recovery-notice resolution
+
+Slice 54 [#17752](https://github.com/paperclipai/paperclip/pull/17752) is verified at
+`f98e6b5b57558c78ba111170ab3d8d19ff667827`: all 53 checks/status entries passed,
+Greptile 5/5, zero unresolved threads, 400 passing local cases, workspace types
+and build. The PR remains draft/open and unmerged. Intermediate failures and the
+full-local-suite limitation remain recorded in its audit and PR body.
+
+Slice 55, `codex/sleep-55-native-recovery-notices`, moves only committed native
+finalization recovery-notice resolution to scoped transaction wakes. The existing
+action rows remain truth; the owner rechecks reused action identities under
+lifecycle locks. [Its audit](2026-10-26-native-recovery-notice-sleep-audit.md)
+records producer coverage, 102 passing worker/startup cases and 340 passing
+affected cases (442 total), plus workspace typechecking. Build and remote
+acceptance remain pending. All other inventory, prior gate failures,
+suppression audits and master propagation remain active obligations.
@@ -0,0 +1,275 @@
+# Committed native recovery notices: sleep audit
+
+Date: 2026-10-26. Slice 55 starts at
+`f98e6b5b57558c78ba111170ab3d8d19ff667827`,
+[parent #17752](https://github.com/paperclipai/paperclip/pull/17752).
+Branch: `codex/sleep-55-native-recovery-notices`. The full scope remains in the
+[recurring-server migration ledger](2026-10-24-recurring-server-sleep-stack.md).
+
+## One obligation and its owner
+
+A native finalization may commit while its recovery notice remains active. The
+old periodic finalizer replay called `resolveCommittedFinalizationRecovery` to
+resolve that notice. This slice extracts that duty into `native-recovery-notices`
+and schedules it from committed changes. The recovery-action service still owns
+the actual resolution write and its existing notifications. Other watchdogs,
+external conversation settlement, Muse reconciliation and physical resource
+settlement are separate obligations.
+
+The existing `issue_recovery_actions` rows are durable truth. No new table,
+per-feature timer or provider branch is introduced. Direct/manual finalizer calls
+retain notice resolution by default. The periodic path passes a narrow
+`skipNativeRecoveryNotices` flag; all its other completion/resource behavior
+remains enabled. A real reconciler test verifies the resource-settlement callback
+still runs when notice resolution is delegated.
+
+## Pending predicate and quiet states
+
+A candidate binds company, issue and run across the native coordinator, heartbeat
+run and issue. The coordinator must be committed and have a result. Discovery
+excludes held runner ownership, pending status effects and unfinished run
+projection. A correlated notice lookup requires:
+
+- The same company, source issue and `evidence.runId`.
+- Kind `active_run_watchdog` and status `active` or `escalated`.
+- `wakePolicy.kind = resume_native_run`, or the existing special cause
+  `native_workspace_finalization_owner_unverified`.
+
+An exhausted owner-verification notice can still need resolution even after its
+retry policy is cleared. Unrelated notices, uncommitted runs, resolved history,
+and otherwise idle configuration do not arm a recurring scan. Discovery reads
+IDs in keyset pages of 100 and the shared worker executes company-fair batches of
+25. Real resolution failures retain shared per-entry backoff, starting at 30
+seconds and bounded at five minutes. A successful attempt with remaining work
+retains a scoped 30-second retry. Startup reconstructs durable unfinished notices.
+
+The owner first checks for a matching notice to avoid taking lifecycle locks on
+the normal already-settled path. That read is an optimization, not authority.
+Actual resolution locks coordinator, issue, run, then matching actions, following
+native recovery's existing order. It rechecks the current committed coordinator,
+run binding and physical ownership. The background caller also checks effect and
+projection settlement under those locks. The existing action service performs
+the resolution inside that transaction.
+
+## Why the action must be read again
+
+`upsertSourceScoped` can reuse an active action ID and change its run, cause,
+fingerprint or policy. Resolving a captured ID alone can erase a newer incident.
+The owner selects matching actions only after acquiring lifecycle locks and locks
+them before calling the resolution service. The regression holds the run row lock,
+starts resolution, changes the notice through the real upsert service, and then
+releases the lock. The new notice stays active. A separate four-writer regression
+observes exactly one recovery-action UPDATE.
+
+## Producer and boundary audit
+
+| Producer | Wake contract |
+| --- | --- |
+| Recovery-action insert, supersession, in-place update and budget exhaustion | Scoped notice hint before the existing transaction write. This includes changes to kind or wake policy alone, not just run/cause changes. |
+| Recovery-action resolution | Signal only after a matching pending row is found and before the write. An empty resolution call must not wake this worker. |
+| Native decision/finalization and pending-effect settlement | Existing finalization notification fanout now includes notices. The committed transaction remains authoritative. |
+| Run projection and heartbeat lifecycle status writers | Scoped signal before the write; finishing projection or releasing an ownership hold can make a notice actionable. |
+| Relevant issue changes | Existing issue transaction sends an issue-scoped hint. |
+| Worktree execution activation | Global reconstruction hint through the existing settings transaction. |
+| Recovery route reopening a resolved action | Explicit issue-scoped hint in the reopening transaction. |
+
+`native-workspace-export-retry` already uses the finalization fanout when it changes
+the action to `resume_native_run`; its coordinator becomes `result_accepted`, so
+notice resolution waits for the later commit. Native workspace owner recovery
+likewise queues result finalization; the subsequent commit supplies the activation.
+
+Other direct action writers were reviewed. Legacy execution admission and
+historical backfill create different causes without a native-resume policy.
+Backfill, system recovery, workspace owner progress and queue continuation writes
+preserve the top-level run/cause/kind/status eligibility while adding evidence.
+The wake-queue adapter adds failure-block evidence. Explicit continuation creates
+resolved authorization receipts or resolves previous actions. Safe replacement,
+workspace reconstruction/export recovery and execution-recovery resolution remove
+active obligations. Their existing queued candidate is rechecked and disarmed;
+they cannot create new eligible notice work. Recovery-service monitor/deadline and
+board-disposition writes do not introduce the matching native policy. New eligible
+notice creation remains with the instrumented action service.
+
+All producer hints use the existing database intent and outer transaction protocol.
+Nested savepoint release is not commit. Unknown outcomes stay with that shared
+protocol. Process-local signals cover instrumented writers in this process;
+independent SQL writers and replicas without supported ingress are not newly
+supported. No database LISTEN connection is added.
+
+The notification boundary explicitly copies only company, issue and run IDs. A
+structurally compatible caller can carry additional internal fields; serializing
+its entire object is not safe. A real finalization regression exposed this during
+implementation when a caller's database object caused circular JSON serialization.
+The corrected hint uses the same narrow payload contract as adjacent workers.
+
+## Lifecycle and verification
+
+Public heartbeat registration preserves task drain, restore and worktree execution
+suppression. Settings activation reconstructs work. The coordinator retains a
+first write during drain and a failed retry through drain/restart, stops future
+subscription work on shutdown and joins active resolution. A new startup discovers
+notices written while stopped.
+
+At this checkpoint, **102 tests pass**: 22 new worker/owner cases and 80 startup
+cases. Coverage includes startup, zero idle SQL, resolved history, real first and
+late producers, outer commit/rollback, finalizer commit, policy/kind-only changes,
+supersession, escalated owner notices, concurrent resolution, action reuse while
+waiting for a real row lock, no-op resolution, pending effects, projection,
+ownership release, suppression/activation, drain/retry/restart, shutdown and
+retained periodic resource settlement.
+
+Preserved failures:
+
+- Before producer wiring, six tests produced three failures: late notice, outer
+  commit and later finalizer commit did not wake the worker. The other three passed.
+- The first wiring run failed all six notice tests while all 80 startup cases
+  passed. A stronger fixture assertion exposed `side_effect_planning_failed` with
+  circular JSON serialization. The hint now copies its three supported IDs. All
+  initial and expanded cases pass with the real finalizer.
+
+Local logs: `/tmp/sleep55-initial-red.log`, `/tmp/sleep55-producers-green.log`
+(the failed intermediate run), `/tmp/sleep55-producer-diagnostic.log`,
+`/tmp/sleep55-lifecycle.log` (96 passes), and `/tmp/sleep55-final-lifecycle.log`
+(102 passes). Six complete affected suites also pass: 340 cases across native finalization
+recovery, finalizer telemetry, recovery actions, the status arbiter corpus,
+result-commit work and run projection. Total: **442 passing local cases**.
+Workspace typechecking passes. Evidence is in `/tmp/sleep55-affected.log` and
+`/tmp/sleep55-types.log`. Build, publication and exact-head CI/review remain
+pending at this checkpoint. The full local suite is not claimed: this host's
+embedded PostgreSQL startup previously exhausted semaphore capacity. No further
+shared OS cleanup was performed; fresh-host CI is required for full acceptance.
+
+## Remaining work
+
+Master was refreshed; `origin/master` was subsequently observed at `125ddcabe5`.
+Full-stack integration of later master changes and fresh gates remains required.
+The older failed gates and eight-worker restore/worktree suppression audit remain
+open. All other duties in the ledger remain in scope. This establishes worker
+behavior, not deployed database suspension, host sleep or fleet savings. The stack
+is unmerged, and this request does not authorize merging it.
+
+## Reply and notice contention repair
+
+The committed-response owner previously locked issue, run, then coordinator.
+Notice resolution, status arbitration, and physical workspace recovery lock
+coordinator, issue, then run. Concurrent reply and notice repair could therefore
+deadlock even though both were authorized to settle the same committed result.
+The reply owner now acquires and validates its exact committed, unleased
+coordinator first, then retains its existing issue/run and presentation checks.
+No publication authority, completion policy, retry cadence, or transaction scope
+is broadened. The change belongs with this notice worker because introducing
+that concurrent owner exposed the conflicting order.
+
+The regression uses a real committed Telegram reply, real notice creation, and
+the real resolution/publication owners. It holds the notice transaction at its
+first coordinator lock, observes the reply's actual PostgreSQL blocking state,
+then continues notice resolution with a bounded local lock timeout. Before the
+fix PostgreSQL rejected reply recovery with `40P01` (deadlock detected). After
+the fix both operations commit, the notice resolves, and exactly one accepted
+comment/publication exists; replay publishes nothing.
+
+The focused regression passed, followed by 73 selected committed/accepted-response
+chat cases (1,570 unrelated chat cases filtered out) and all 141 cases in the
+notice, finalizer telemetry, startup, and response-authority suites. Logs:
+`/tmp/sleep55-lock-order-{red,green,chat,consumers}.log`. This is 214 distinct
+passing cases across five files, not a full chat-file or full repository test
+run. Integrated typecheck, build, and fresh remote checks remain required.
+
+## Projection fixture follow-through after the lock-order repair
+
+The coordinator-first response-recovery guard exposed stale projection fixtures.
+Thirteen cases inserted a native run without a committed finalization record.
+Recovery correctly stopped before loading that run, so the expected measured
+projection was absent. The failure reproduced locally against a new disposable
+PostgreSQL database: 13 failures and four passes, matching the CI failures on
+later stack branches.
+
+The fixture now uses the existing accepted-result and finalization setup. It
+retains a future retry while comparing narrow and full-row reads, so both paths
+observe the same state without publishing between comparisons. The selected
+answer guard, typed context flags, non-object result evidence, payload byte
+limits, unchanged-row assertions, company/issue scope checks, and separate real
+publication comparison remain. All 17 cases now pass; server typechecking also
+passes. The existing external-test-database convention permits a real database
+replay without requiring another embedded PostgreSQL instance. No production
+lock, recovery eligibility, retry policy, or projection was weakened.
+
+## Startup readiness follow-up (2026-10-27)
+
+A held recovery-notice sweep reproduced HTTP startup waiting for task locks.
+Registration now tracks its readiness promise through the existing heartbeat
+shutdown owner instead of awaiting it on the readiness path. The coordinator
+continues to own worker cancellation. The regression proves HTTP readiness and
+subsequent registration proceed while shutdown waits for the held work. All 86
+startup cases pass on the master-integrated branch; no timer or persistence is
+added. Full integrated typechecking and head-specific CI remain separate gates.
+
+## Current parent integration and routine-hint regression (2026-10-29)
+
+This slice incorporates parent #17752 at
+`a49a2dac13b0e1c19da5d3d6b435c781703b1c0b` through a normal merge. A
+batched tree comparison preserves 1,209 parent-only entries, eight owned-only
+entries, 11,524 identical entries and 12 automatic shared merges. Two manual
+paths remain explicit: ordinary notice fanout is inside the existing issue/run
+guard; the chat test's leading import conflict retains all four independent
+imports and removes only the three marker lines. Every other automatically
+merged byte in that test remains intact before comment removal. The complete
+package tree matches parent profile 80, including all historical identities.
+
+Thirteen added comment objects in five files were removed with identical nontrivia
+TypeScript token streams. No directive was removed. Notice settlement retains
+coordinator, issue, run and action lock ordering; exact current incident matching;
+existing action-resolution side effects; and the separate status/effect/resource
+owners. A discovered action ID alone is not authority to resolve a later incident.
+
+The new real-database regression submits three routine company-only finalization
+hints after startup becomes quiet. A controlled baseline moves only notice fanout
+outside the guard and keeps every other parent guard. It issues three candidate
+SELECTs and fails the zero-query assertion. Restoring production bytes passes the
+same assertion. The real notice writer then creates and resolves an action for a
+committed run, returns outcome `restored`, and settles back to quiet. Red fails one
+selected case in 7.46 seconds; green passes that same case. Twenty-two cases outside
+the focused pattern are not counted as passing there. Both phases use fresh
+dedicated PostgreSQL 17 databases and restore the production helper exactly.
+
+The complete main command passes 1,873 cases in 21 files in 309.69 seconds, with
+no skips. Process recovery and external-chat response run in separate database
+phases. The giant chat integration file is intentionally a focused selection,
+including the real reply/notice row-lock regression and related provider and
+current-authority cases; unselected chat cases do not constitute acceptance.
+All test assertions and timeouts are unchanged.
+
+### Complete current-source acceptance
+
+Source `730e00a7b007a36d188ca0a328ef050d3a25d1ee` passes **2,556 cases
+in 23 complete files**, with no skips in those files. The main command passes
+1,873 cases in 21 files in 309.69 seconds. Process recovery passes 410 in 163.54
+seconds; external-chat response passes 273 in 53.48 seconds. A separate focused
+chat selection passes 18 cases in 13.44 seconds, including the actual reply/notice
+row-lock race, Slack/Telegram recovery, current-authority refusals, canonical
+response binding and internal GitHub navigation. The other 1,796 cases in that
+chat file remain unselected. Total selected passing cases: 2,574. This is not full
+chat-suite or repository acceptance.
+
+The complete files include all 23 notice-worker cases, 17 presentation/ownership
+read cases and 88 startup cases. Production server typechecking, build, module
+boundaries, migration order and the source identity check pass. Twenty dedicated
+PostgreSQL 17 databases use the unchanged migration 413 schema. Embedded-only
+calls use existing fresh-per-call recovery fixtures. Every temporary import is
+restored byte-for-byte and all 24 selected file hashes match after completion.
+No assertion or timeout is relaxed. The focused scope green takes 6.69 seconds;
+the intended failing unscoped baseline is not counted as acceptance.
+
+Parent #17752 merged as `50fe32684d2d4e1db129e3c1bb455571344a0c09`, after
+57 clean current-head checks and Greptile 5/5. Its complete reviewed tree equals
+the actual squash. Normal merge `a0f2b3f4348b078219a53d8cc135c84202ba505e`
+adopts that ancestry with this child's entire tested tree unchanged. This slice
+can now target master. Fresh exact-head CI and review remain separate gates.
+
+The unchanged package tree retains parent profile 80 acceptance: 130 passing
+installation/compatibility/transport cases, 22 installed ACPX/Grok contract cases
+and 26 native-provider Rust cases. One existing opt-in closed Pi snapshot case
+was skipped because its distribution root was unset. This does not qualify a live
+Pi installation. Earlier failures in this audit retain their revision and
+subsequent disposition. No deployed sleep, expanded Cloud alarm coverage or
+measured savings is claimed.
@@ -1,3 +1,6 @@
+import { resolveCommittedNativeRecoveryNotices, pendingNativeRecoveryNotices } from "../services/native-runtime/native-recovery-notices.js";
+import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
+import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js";
 import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js";
 import { nativeResponsePresentationWork } from "../services/native-runtime/native-response-presentation-work.js";
 import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifications.js";
@@ -74469,6 +74472,54 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
    };
  }

+  it("repairs a committed reply and its recovery notice without reversing row locks", async () => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    const scope = { companyId: context.fixture.companyId, issueId: context.issue.id, runId: context.runId };
+    let response: Promise<boolean> | undefined;
+    try {
+      const [run] = await readHeartbeatRunsById(context.runId);
+      await projectCommittedRun({ db, run, coordinator: context.coordinator });
+      const notice = await issueRecoveryActionService(db).upsertSourceScoped({
+        companyId: scope.companyId, sourceIssueId: scope.issueId, kind: "active_run_watchdog",
+        ownerType: "system", cause: "native_workspace_sync_out_failed", fingerprint: scope.runId,
+        evidence: { runId: scope.runId }, nextAction: "Retry finalization",
+        wakePolicy: { kind: "resume_native_run", runId: scope.runId },
+      });
+      expect(await pendingNativeRecoveryNotices(db, scope)).toHaveLength(1);
+      await db.transaction(async tx => {
+
+
+        await tx.select().from(nativeRunFinalizations)
+          .where(eq(nativeRunFinalizations.runId, scope.runId)).for("update");
+        const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`) as unknown as Array<{ pid: number }>;
+        response = recoverCommittedNativeChatResponse(db, { ...scope, onlyWhenDue: true });
+
+        void response.catch(() => {});
+        await vi.waitFor(async () => {
+          const [state] = await db.execute(sql`select exists (
+            select 1 from pg_stat_activity where datname = current_database()
+              and ${backend.pid} = any(pg_blocking_pids(pid))
+          ) as blocked`) as unknown as Array<{ blocked: boolean }>;
+          expect(state.blocked).toBe(true);
+        });
+        await tx.execute(sql`set local lock_timeout = '1s'`);
+        await resolveCommittedNativeRecoveryNotices(tx as unknown as Db, scope, true);
+      });
+      await expect(response).resolves.toBe(true);
+      expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, notice.id)))[0])
+        .toMatchObject({ status: "resolved", outcome: "restored" });
+      const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, scope.runId));
+      expect(comments).toHaveLength(1);
+      expect(comments[0].body).toBe(context.result.summary);
+      expect(await readChatPublicationsByCommentId(comments[0].id)).toHaveLength(1);
+      await expect(recoverCommittedNativeChatResponse(db, scope)).resolves.toBe(false);
+      expect(await pendingNativeRecoveryNotices(db, scope)).toEqual([]);
+    } finally {
+      await response?.catch(() => {});
+      await context.service.shutdown();
+    }
+  });
+
  it.each(["slack", "telegram"] as const)(
    "recovers a committed %s answer once without rerunning or changing review/quarantine evidence",
    async (provider) => {
@@ -0,0 +1,345 @@
+import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js";
+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 { nativeRecoveryNoticeWork } from "../services/native-runtime/native-recovery-notice-work.js";
+import { resolveCommittedNativeRecoveryNotices } from "../services/native-runtime/native-recovery-notices.js";
+import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js";
+import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
+import { issueRecoveryActions } from "@paperclipai/db";
+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 recovery notice lifecycle", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: Db, root: string;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let worker: ReturnType<ReturnType<typeof nativeRecoveryNoticeWork>["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-recovery-notice-"));
+    process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = root;
+    const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RECOVERY_NOTICE_TEST_DATABASE_URL");
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-recovery-notice-");
+    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 = resolveCommittedNativeRecoveryNotices) {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted,
+      onError: error => errors.push(error) });
+    worker = nativeRecoveryNoticeWork(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;
+  }
+  async function committed() {
+    const binding = await accepted();
+    const result = await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    expect(result.phase, JSON.stringify(await state(binding.runId))).toBe("committed");
+    return binding;
+  }
+  async function notice(binding: Awaited<ReturnType<typeof seed>>, writer: Db = db,
+    patch: Partial<Parameters<ReturnType<typeof issueRecoveryActionService>["upsertSourceScoped"]>[0]> = {}) {
+    return issueRecoveryActionService(writer).upsertSourceScoped({ companyId, sourceIssueId: binding.issueId,
+      kind: "active_run_watchdog", ownerType: "system", cause: "native_workspace_sync_out_failed",
+      fingerprint: binding.runId, evidence: { runId: binding.runId }, nextAction: "Retry finalization",
+      wakePolicy: { kind: "resume_native_run", runId: binding.runId }, ...patch });
+  }
+  async function action(id: string) {
+    return db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)).then(rows => rows[0]);
+  }
+  it("reconstructs pending notices and leaves resolved history quiet across restart", async () => {
+    const binding = await committed(); const pending = await notice(binding);
+    await start(); await flush(); expect(await action(pending.id)).toMatchObject({ status: "resolved", outcome: "restored" });
+    await quiet(); const settled = await action(pending.id);
+    await coordinator!.stop(); await start(); await flush(); await quiet(); expect(await action(pending.id)).toEqual(settled);
+  });
+  it("ignores routine company-only finalization hints and still resolves a real notice", 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 committed();
+    const pending = await notice(binding); await flush();
+    expect(await action(pending.id)).toMatchObject({ status: "resolved", outcome: "restored" });
+    await quiet();
+  });
+
+  it("wakes from a real notice created after startup is quiet", async () => {
+    const binding = await committed(); await start(); await flush(); await quiet();
+    const pending = await notice(binding); await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it.each(["commit", "rollback"])("honors the notice writer's outer %s", async outcome => {
+    const binding = await committed(); await start(); await flush();
+    let actionId = "";
+    const transaction = db.transaction(async tx => {
+      actionId = (await notice(binding, tx as unknown as Db)).id;
+      await vi.advanceTimersByTimeAsync(5000);
+      expect(await action(actionId)).toBeUndefined();
+      if (outcome === "rollback") throw new Error("rollback notice");
+    });
+    if (outcome === "rollback") await expect(transaction).rejects.toThrow("rollback notice"); else await transaction;
+    await flush();
+    if (outcome === "rollback") expect(await action(actionId)).toBeUndefined();
+    else expect((await action(actionId)).status).toBe("resolved");
+    await quiet();
+  });
+  it("wakes when the real finalizer commits after a notice was already present", async () => {
+    const binding = await accepted(); const pending = await notice(binding);
+    await start(); await flush(); await quiet();
+    await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true,
+      skipNativeRecoveryNotices: true });
+    await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it("preserves a notice rebound to a new run after discovery", async () => {
+    const binding = await committed(); const pending = await notice(binding);
+    const differentRun = randomUUID();
+    const resolve: typeof resolveCommittedNativeRecoveryNotices = async (database, scope, settled) => {
+      const updated = await notice({ ...binding, runId: differentRun }); expect(updated.id).toBe(pending.id);
+      await resolveCommittedNativeRecoveryNotices(database, scope, settled);
+    };
+    await start(resolve); await flush();
+    expect(await action(pending.id)).toMatchObject({ status: "active", evidence: { runId: differentRun } }); await quiet();
+  });
+  it("does not wake for an already-resolved notice", async () => {
+    const binding = await committed(); const pending = await notice(binding);
+    await start(); await flush();
+    const queries: string[] = []; const previous = db.$client.options.debug;
+    db.$client.options.debug = (_id, query) => { queries.push(query); };
+    try {
+      await issueRecoveryActionService(db).resolveActiveForIssue({ companyId, sourceIssueId: binding.issueId,
+        actionId: pending.id, status: "resolved", outcome: "restored" });
+      await flush();
+      expect(queries.filter(query => query.includes('from "native_run_finalizations"'))).toEqual([]);
+    } finally { db.$client.options.debug = previous; }
+    await quiet();
+  });
+  it.each(["restore", "worktree"])("preserves %s suppression and settings activation", async mode => {
+    const binding = await committed(); const pending = await notice(binding);
+    const runtimeEnv = mode === "restore" ? { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true" }
+      : { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "notice-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 }).registerNativeRecoveryNoticeWork(coordinator, () => admitted);
+    await worker.ready; await flush(); await quiet(); expect((await action(pending.id)).status).toBe("active");
+    if (mode === "worktree") {
+      try { await settings.updateExperimental({ enableWorktreeRunExecution: true }); await flush();
+        expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+      } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); }
+    }
+  });
+  it("retains the first notice committed during drain", async () => {
+    const binding = await committed(); await start(); await flush(); admitted = false;
+    const pending = await notice(binding); await vi.advanceTimersByTimeAsync(120_000); await quiet();
+    expect((await action(pending.id)).status).toBe("active"); admitted = true;
+    await vi.advanceTimersByTimeAsync(30_000); await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it("retains a real failure retry through drain and restart", async () => {
+    const binding = await committed(); const pending = await notice(binding);
+    const resolve: typeof resolveCommittedNativeRecoveryNotices = async () => { throw new Error("notice unavailable"); };
+    await start(resolve); expect(errors).toHaveLength(1); errors.length = 0;
+    expect(coordinator!.nextWakeAt()).not.toBeNull(); admitted = false;
+    await vi.advanceTimersByTimeAsync(120_000); await quiet(); expect((await action(pending.id)).status).toBe("active");
+    await coordinator!.stop(); admitted = true; await start(); await flush();
+    expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it("joins in-flight resolution on shutdown and reconstructs later notices", async () => {
+    let release!: () => void; const gate = new Promise<void>(resolve => { release = resolve; }); const entered = vi.fn();
+    const resolve: typeof resolveCommittedNativeRecoveryNotices = async (...args) => {
+      entered(); await gate; return resolveCommittedNativeRecoveryNotices(...args);
+    };
+    const first = await committed(); const second = await committed(); await start(resolve); await flush();
+    const pending = await notice(first);
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(entered).toHaveBeenCalledTimes(1); });
+    const later = await notice(second); let stopped = false;
+    const stopping = coordinator!.stop().then(() => { stopped = true; });
+    await delay(20); expect(stopped).toBe(false); release(); await stopping;
+    expect((await action(pending.id)).status).toBe("resolved"); expect((await action(later.id)).status).toBe("active");
+    await start(); await flush(); expect((await action(later.id)).status).toBe("resolved"); await quiet();
+  });
+  it("waits for real pending-effect settlement", async () => {
+    const binding = await committed(); const pending = await notice(binding); const current = await state(binding.runId);
+    const [effect] = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, current.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 action(pending.id)).status).toBe("active");
+    await statusCommitter.dispatchPendingNativeStatusEffects({ db, ...binding, decisionId: current.decisionId!, onlyWhenDue: true });
+    await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it("waits for projection and wakes from the real projection owner", async () => {
+    const binding = await committed(); const pending = await notice(binding);
+    await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, binding.runId));
+    await start(); await flush(); await quiet(); expect((await action(pending.id)).status).toBe("active");
+    await projectCommittedRun({ db, run: await read(binding.runId), coordinator: await state(binding.runId) });
+    await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it("wakes after the real lifecycle writer releases an ownership hold", async () => {
+    const binding = await committed(); const pending = await notice(binding);
+    await db.update(heartbeatRuns).set({ status: "running", nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" })
+      .where(eq(heartbeatRuns.id, binding.runId));
+    await start(); await flush(); await quiet(); expect((await action(pending.id)).status).toBe("active");
+    const lifecycle = createHeartbeatLifecycle(db, { emitTerminalAgentTaskRun: vi.fn() } as unknown as HeartbeatLifecycleDependencies);
+    await lifecycle.setRunStatus(binding.runId, "succeeded", { nativePhase: "committed", errorCode: null });
+    await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it("delegates only notice repair and preserves periodic resource settlement", async () => {
+    const binding = await committed(); const pending = await notice(binding); const settled = vi.fn(async () => undefined);
+    await reconcileNativeFinalizations(db, [binding.runId], { skipNativeRecoveryNotices: true,
+      skipNativeResponsePresentation: true, skipNativeReviewResponse: true, onWorkspaceSettled: settled });
+    expect((await action(pending.id)).status).toBe("active"); expect(settled).toHaveBeenCalledTimes(1);
+    await start(); await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it.each(["wakePolicy", "kind"])("activates an existing notice after only its %s changes", async field => {
+    const binding = await committed();
+    const pending = await notice(binding, db, field === "kind" ? { kind: "missing_disposition" } : { wakePolicy: null });
+    await start(); await flush(); await quiet(); expect((await action(pending.id)).status).toBe("active");
+    const updated = await notice(binding); expect(updated.id).toBe(pending.id);
+    await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it("resolves the escalated owner-verification notice without a retry policy", async () => {
+    const binding = await committed();
+    const pending = await notice(binding, db, { cause: "native_workspace_finalization_owner_unverified", maxAttempts: 1 });
+    await notice(binding, db, { cause: "native_workspace_finalization_owner_unverified", maxAttempts: 1 });
+    expect(await action(pending.id)).toMatchObject({ status: "escalated", wakePolicy: null });
+    await start(); await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it("wakes for a new identity while preserving the superseded notice", async () => {
+    const binding = await committed(); const old = await notice({ ...binding, runId: randomUUID() });
+    await start(); await flush(); await quiet();
+    const pending = await notice(binding, db, { supersedeOnIdentityChange: true });
+    expect(pending.id).not.toBe(old.id); await flush();
+    expect((await action(old.id)).status).toBe("cancelled"); expect((await action(pending.id)).status).toBe("resolved"); await quiet();
+  });
+  it("writes a resolution once across competing live/recovery owners", async () => {
+    const binding = await committed(); const pending = await notice(binding);
+    const queries: string[] = []; const previous = db.$client.options.debug;
+    db.$client.options.debug = (_id, query) => { queries.push(query); };
+    try {
+      await Promise.all(Array.from({ length: 4 }, () => resolveCommittedNativeRecoveryNotices(db, binding, true)));
+      expect(queries.filter(query => query.startsWith('update "issue_recovery_actions"'))).toHaveLength(1);
+    } finally { db.$client.options.debug = previous; }
+    expect((await action(pending.id)).status).toBe("resolved");
+  });
+  it("rechecks the action after waiting for the run lock", async () => {
+    const binding = await committed(); const pending = await notice(binding);
+    let release!: () => void; let locked!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const ready = new Promise<void>(resolve => { locked = resolve; });
+    const holding = db.transaction(async tx => {
+      await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, binding.runId)).for("update");
+      locked(); await gate;
+    });
+    await ready;
+    let waiting = false; const previous = db.$client.options.debug;
+    db.$client.options.debug = (_id, query) => { if (query.includes('from "heartbeat_runs"') && query.includes('for update')) waiting = true; };
+    const resolving = resolveCommittedNativeRecoveryNotices(db, binding, true);
+    const replacementRun = randomUUID();
+    try {
+      await vi.waitFor(() => { expect(waiting).toBe(true); });
+      const updated = await notice({ ...binding, runId: replacementRun }); expect(updated.id).toBe(pending.id);
+    } finally { release(); db.$client.options.debug = previous; await holding; await resolving; }
+    expect(await action(pending.id)).toMatchObject({ status: "active", evidence: { runId: replacementRun } });
+  });
+});
@@ -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() })),
+    registerNativeRecoveryNoticeWork: 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() })),
@@ -599,6 +600,13 @@ describe("startServer feedback export wiring", () => {
    }
  });

+  it.each([false, true])("registers native recovery notices only when maintenance is enabled (%s)", async enabled => {
+    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
+    await startServer(); await settleStartupHousekeeping();
+    expect(heartbeatServiceMock.registerNativeRecoveryNoticeWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
+    if (enabled) expect(heartbeatServiceMock.registerNativeRecoveryNoticeWork.mock.calls[0][1]()).toBe(true);
+  });
+
  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();
@@ -1074,7 +1082,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, skipNativeRunProjection: 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, skipNativeRecoveryNotices: true });
      expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
      expect(retiredDetector).not.toHaveBeenCalled();
    } finally {
@@ -1230,7 +1238,7 @@ describe("startServer feedback export wiring", () => {
    }
  });

-  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 => {
+  it.each(["retained cleanup", "workspace finalization", "native run projection", "native result commit", "native recovery notices"] 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; });
@@ -1238,9 +1246,11 @@ describe("startServer feedback export wiring", () => {
    const register = owner === "retained cleanup" ? heartbeatServiceMock.registerRetainedNativeSessionCleanupWork
      : owner === "workspace finalization" ? heartbeatServiceMock.registerNativeWorkspaceFinalizationWork
      : owner === "native run projection" ? heartbeatServiceMock.registerNativeRunProjectionWork
-      : heartbeatServiceMock.registerNativeResultCommitWork;
+      : owner === "native result commit" ? heartbeatServiceMock.registerNativeResultCommitWork
+      : heartbeatServiceMock.registerNativeRecoveryNoticeWork;
    if (owner === "native run projection") heartbeatServiceMock.registerNativeRunProjectionWork.mockReturnValueOnce({ ready });
    else if (owner === "native result commit") heartbeatServiceMock.registerNativeResultCommitWork.mockReturnValueOnce({ ready });
+    else if (owner === "native recovery notices") heartbeatServiceMock.registerNativeRecoveryNoticeWork.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>);
@@ -1842,6 +1842,12 @@ async function startServerWithDatabaseTeardown(
      server.on("close", worker.stop);


+      trackHeartbeatSchedulerWork(worker.ready);
+    }
+    if (config.heartbeatSchedulerEnabled) {
+      const worker = heartbeat.registerNativeRecoveryNoticeWork(deliveryWork,
+        () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby());
+
      trackHeartbeatSchedulerWork(worker.ready);
    }
    if (config.heartbeatSchedulerEnabled) {
@@ -2008,7 +2014,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, skipNativeRunProjection: 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, skipNativeRecoveryNotices: true })
            .then(() => heartbeat.promoteDueScheduledRetries())
            .then(async (promotion) => {
              await heartbeat.resumeQueuedRuns();
@@ -1,3 +1,4 @@
+import { notifyNativeRecoveryNotices } from "../services/native-recovery-notice-notifications.js";
 import { notifyNativeStatusReassessment } from "../services/native-status-reassessment-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../services/unsafe-workspace-export-notifications.js";
 import { notifyChatInboundWakeNotices } from "../services/chat-inbound-wake-notice-notifications.js";
@@ -9933,6 +9934,7 @@ export function issueRoutes(
                );
              }
              await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: lockedIssue.companyId, issueId: lockedIssue.id });
+              await notifyNativeRecoveryNotices(tx, { companyId: lockedIssue.companyId, issueId: lockedIssue.id });
              const [reopened] = await tx
                .update(issueRecoveryActions)
                .set({ status: "active", outcome: null, resolvedAt: null })
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";

 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  nativeRecoveryNotices: "native-recovery-notices",
  nativeRunProjection: "native-run-projection",
  nativeResultCommit: "native-result-commit",
  nativeWorkspaceFinalization: "native-workspace-finalization",
@@ -1,3 +1,4 @@
+import { nativeRecoveryNoticeWork } from "./native-runtime/native-recovery-notice-work.js";
 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";
@@ -4096,6 +4097,14 @@ export function heartbeatService(
    triggerIssueMonitor,

    reportRunActivity: clearDetachedRunWarning,
+    registerNativeRecoveryNoticeWork: (
+      coordinator: Parameters<ReturnType<typeof nativeRecoveryNoticeWork>["register"]>[0], canRun: () => boolean,
+    ) => nativeRecoveryNoticeWork(db).register(coordinator,
+      () => canRun() && readTaskDrain(new Date()) === null,
+      async () => {
+        const state = await getSchedulingSuppression(true);
+        return !state.suppressed || state.reason === "task_drain";
+      }),
    registerNativeRunProjectionWork: (
      coordinator: Parameters<ReturnType<typeof nativeRunProjectionWork>["register"]>[0], canRun: () => boolean,
    ) => nativeRunProjectionWork(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; skipNativeRunProjection?: 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; skipNativeRecoveryNotices?: boolean }) {
    if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db);
    const staleThresholdMs = opts?.staleThresholdMs ?? 0;
    const now = new Date();
@@ -1995,6 +1995,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
      skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization,
      skipNativeResultCommit: opts?.skipNativeResultCommit,
      skipNativeRunProjection: opts?.skipNativeRunProjection,
+      skipNativeRecoveryNotices: opts?.skipNativeRecoveryNotices,
      environmentRuntime,
      pluginWorkerManager: dependencies.pluginWorkerManager,
      onWorkspaceSettled: settleRecoveredNativeWorkspace,
@@ -1,3 +1,4 @@
+import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js";
 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";
@@ -1026,6 +1027,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
    // 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 notifyNativeRecoveryNotices(tx, { companyId: previous.companyId, runId: previous.id });
      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"))
@@ -1,3 +1,4 @@
+import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
 import { notifyNativeResultCommit } from "./native-result-commit-notifications.js";
 import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
@@ -629,6 +630,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
          await notifyDirectoryCleanup(tx);
          await notifyInstructionCollection(tx);
          await notifyCancelledWorkspacePreservation(tx);
+          await notifyNativeRecoveryNotices(tx);
          await notifyNativeRunProjection(tx);
          await notifyNativeResultCommit(tx);
          await notifyNativeWorkspaceFinalization(tx);
@@ -1,3 +1,4 @@
+import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
 import { writeExecutionControl } from "./execution-control-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -321,6 +322,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
  ): Promise<IssueRecoveryAction> {
    try {
      const created = await db.transaction(async (tx) => {
+        await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
        await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition);
        await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation);
        await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
@@ -411,6 +413,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
          Math.min(nextAttemptCount, effectiveMaxAttempts),
        );
        const [exhausted] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => {
+          await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
          if ((!input.preserveExistingOwner && existing.cause !== input.cause)
            || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) {
            await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
@@ -471,6 +474,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
        return toReadModel(exhausted);
      }
      const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => {
+        await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
        if ((!input.preserveExistingOwner && existing.cause !== input.cause)
          || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) {
          await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
@@ -544,6 +548,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {

    try {
      const [created] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => {
+        await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
        await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
        return tx
          .insert(issueRecoveryActions)
@@ -590,6 +595,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
      const pending = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions)
        .where(and(...predicates)).limit(1);
      if (!pending.length) return [];
+      await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
      await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
      return tx
        .update(issueRecoveryActions)
@@ -1,3 +1,4 @@
+import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
 import { notifyNativeResultCommit } from "./native-result-commit-notifications.js";
 import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
@@ -11505,6 +11506,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 notifyNativeRecoveryNotices(tx, { companyId: receiptExisting.companyId, issueId: id });
          await notifyNativeRunProjection(tx, { companyId: receiptExisting.companyId, issueId: id });
          await notifyNativeResultCommit(tx, { companyId: receiptExisting.companyId, issueId: id });
          await notifyNativeWorkspaceFinalization(tx, { companyId: receiptExisting.companyId, issueId: id });
@@ -1,3 +1,4 @@
+import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
 import { notifyNativeResultCommit } from "./native-result-commit-notifications.js";
 import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
@@ -11,6 +12,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i

 export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) {
  if (scope.issueId || scope.runId) {
+    await notifyNativeRecoveryNotices(transaction, scope);
    await notifyNativeRunProjection(transaction, scope);
    await notifyNativeResultCommit(transaction, scope);
    await notifyNativeWorkspaceFinalization(transaction, scope);
@@ -0,0 +1,8 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type NativeRecoveryNoticeScope = { companyId: string; issueId?: string; runId?: string };
+
+export function notifyNativeRecoveryNotices(tx: object, scope?: NativeRecoveryNoticeScope) {
+  return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeRecoveryNotices,
+    scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId }) : "*");
+}
@@ -391,6 +391,7 @@ type NativeReconciliationOptions = {
  skipNativeWorkspaceFinalization?: boolean;
  skipNativeResultCommit?: boolean;
  skipNativeRunProjection?: boolean;
+  skipNativeRecoveryNotices?: boolean;
  pluginWorkerManager?: WorkspaceWorkers;
  environmentRuntime?: EnvironmentRuntimeService;
  onWorkspaceSettled?: (input: {
@@ -688,6 +689,7 @@ async function reconcileNativeFinalizationsPass(
          skipNativeResponsePresentation: options.skipNativeResponsePresentation,
          skipNativeReviewResponse: options.skipNativeReviewResponse,
        skipCommittedRunProjection: options.skipNativeRunProjection,
+        skipNativeRecoveryNotices: options.skipNativeRecoveryNotices,
          preserveProviderAttempt: Boolean(
            readNativeWorkspaceSyncReference(
              row.nativeWorkspaceSync,
@@ -750,6 +752,7 @@ async function reconcileNativeFinalizationsPass(
        skipNativeResponsePresentation: options.skipNativeResponsePresentation,
        skipNativeReviewResponse: options.skipNativeReviewResponse,
        skipCommittedRunProjection: options.skipNativeRunProjection,
+        skipNativeRecoveryNotices: options.skipNativeRecoveryNotices,
        preserveProviderAttempt: Boolean(
          readNativeWorkspaceSyncReference(
            row.nativeWorkspaceSync,
@@ -0,0 +1,43 @@
+import type { Db } from "@paperclipai/db";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import type { NativeRecoveryNoticeScope } from "../native-recovery-notice-notifications.js";
+import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
+import { pendingNativeRecoveryNotices, resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js";
+
+type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
+
+export function nativeRecoveryNoticeWork(db: Db, resolve = resolveCommittedNativeRecoveryNotices) {
+  async function load(scope?: NativeRecoveryNoticeScope, signal?: AbortSignal, canRun = () => true): Promise<Entry[]> {
+    const entries: Entry[] = [];
+    let after: string | undefined;
+    while (canRun() && !signal?.aborted) {
+      const rows = await pendingNativeRecoveryNotices(db, 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 recovery notice discovery stopped");
+  }
+  async function check(entry: Entry, signal?: AbortSignal): Promise<Entry | null> {
+    if (signal?.aborted) throw new Error("Native recovery notice resolution stopped");
+    const scope = { companyId: entry.companyId, issueId: entry.issueId, runId: entry.id };
+    await resolve(db, scope, true);
+    return (await pendingNativeRecoveryNotices(db, 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.nativeRecoveryNotices, batchSize: 25,
+      decodeScope: (hint: string): NativeRecoveryNoticeScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (entry: Entry, scope?: NativeRecoveryNoticeScope) => !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 recovery notice resolution paused");
+        return await canRecover() ? check(entry, signal) : null;
+      },
+    }, canRun);
+  }
+  return { load, check, register };
+}
@@ -0,0 +1,73 @@
+import { and, asc, eq, gt, inArray, isNotNull, not, or, sql } from "drizzle-orm";
+import { heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db";
+import { issueRecoveryActionService } from "../issue-recovery-actions.js";
+import type { NativeRecoveryNoticeScope } from "../native-recovery-notice-notifications.js";
+import { nativeRunnerOwnershipNotHeldCondition } from "./native-runner-ownership.js";
+import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js";
+import { pendingNativeRunProjectionCondition } from "./native-run-projection.js";
+
+function matchingNotice(runId: string | typeof nativeRunFinalizations.runId) {
+  return and(
+    eq(issueRecoveryActions.kind, "active_run_watchdog"),
+    inArray(issueRecoveryActions.status, ["active", "escalated"]),
+    sql`${issueRecoveryActions.evidence}->>'runId' = ${runId}::text`,
+    or(sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`,
+      eq(issueRecoveryActions.cause, "native_workspace_finalization_owner_unverified")),
+  );
+}
+
+export function pendingNativeRecoveryNotices(db: Db, scope?: NativeRecoveryNoticeScope, after?: string) {
+  return db.select({ id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.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"), isNotNull(nativeRunFinalizations.resultId),
+      eq(heartbeatRuns.runtimeMode, "native"), nativeRunnerOwnershipNotHeldCondition(),
+      not(pendingNativeStatusEffectsCondition(db)), not(pendingNativeRunProjectionCondition(db)),
+      sql`exists (${db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(
+        eq(issueRecoveryActions.companyId, nativeRunFinalizations.companyId),
+        eq(issueRecoveryActions.sourceIssueId, nativeRunFinalizations.issueId), matchingNotice(nativeRunFinalizations.runId)))})`,
+      scope ? eq(nativeRunFinalizations.companyId, scope.companyId) : undefined,
+      scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined,
+      scope?.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined,
+      after ? gt(nativeRunFinalizations.runId, after) : undefined,
+    )).orderBy(asc(nativeRunFinalizations.runId)).limit(100);
+}
+
+
+
+export async function resolveCommittedNativeRecoveryNotices(db: Db,
+  scope: { companyId: string; issueId: string; runId: string }, onlyWhenSettled = false) {
+  const [pending] = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(
+    eq(issueRecoveryActions.companyId, scope.companyId), eq(issueRecoveryActions.sourceIssueId, scope.issueId),
+    matchingNotice(scope.runId))).limit(1);
+  if (!pending) return;
+  return db.transaction(async transaction => {
+    const tx = transaction as unknown as Db;
+
+
+    const [coordinator] = await tx.select({ phase: nativeRunFinalizations.phase, resultId: nativeRunFinalizations.resultId })
+      .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, scope.companyId),
+        eq(nativeRunFinalizations.issueId, scope.issueId), eq(nativeRunFinalizations.runId, scope.runId))).for("update");
+    if (!coordinator || coordinator.phase !== "committed" || !coordinator.resultId) return;
+    const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(
+      eq(issues.companyId, scope.companyId), eq(issues.id, scope.issueId))).for("update");
+    if (!issue) return;
+    const [run] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+      eq(heartbeatRuns.id, scope.runId), eq(heartbeatRuns.companyId, scope.companyId),
+      eq(heartbeatRuns.nativeIssueId, scope.issueId), eq(heartbeatRuns.runtimeMode, "native"),
+      nativeRunnerOwnershipNotHeldCondition())).for("update");
+    if (!run) return;
+    if (onlyWhenSettled && !(await pendingNativeRecoveryNotices(tx, scope)).length) return;
+    const actions = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(
+      eq(issueRecoveryActions.companyId, scope.companyId), eq(issueRecoveryActions.sourceIssueId, scope.issueId),
+      matchingNotice(scope.runId))).for("update");
+    for (const action of actions) await issueRecoveryActionService(tx).resolveActiveForIssue({
+      companyId: scope.companyId, sourceIssueId: scope.issueId, actionId: action.id,
+      status: "resolved", outcome: "restored",
+      resolutionNote: "The accepted native result and workspace finalization committed successfully; this run needs no further finalization retry.",
+    });
+  });
+}
@@ -1,3 +1,5 @@
+import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js";
+import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js";
 import { notifyNativeRunProjection } from "../native-run-projection-notifications.js";
 import { projectCommittedRun, recoveredExecutionFailureMetadata } from "./native-run-projection.js";
 import { notifyNativeResultCommit } from "../native-result-commit-notifications.js";
@@ -37,7 +39,6 @@ import {
  heartbeatRuns,
  heartbeatRunEvents,
  issueComments,
-  issueRecoveryActions,
  issueThreadInteractions,
  issues,
  nativeRunFinalizations,
@@ -594,21 +595,6 @@ export async function recordNativeFinalizationFailure(input: {
  });
 }

-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"]),
-    sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
-    or(
-      sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`,
-      eq(issueRecoveryActions.cause, "native_workspace_finalization_owner_unverified"),
-    ),
-  ));
-  for (const action of actions) await issueRecoveryActionService(db).resolveActiveForIssue({
-    companyId: run.companyId, sourceIssueId: issueId, actionId: action.id,
-    status: "resolved", outcome: "restored", resolutionNote: "The accepted native result and workspace finalization committed successfully; this run needs no further finalization retry.",
-  });
-}

 async function materializeNativeReviewResponse(db: Db, runId: string, dbOrTx: Db = db) {
  const [run] = await dbOrTx
@@ -754,6 +740,27 @@ export async function recoverCommittedNativeChatResponse(
  try {
    materialized = await db.transaction(async (transaction) => {
      const tx = transaction as unknown as Db;
+
+
+      const [coordinator] = await tx
+        .select()
+        .from(nativeRunFinalizations)
+        .where(
+          and(
+            eq(nativeRunFinalizations.runId, input.runId),
+            eq(nativeRunFinalizations.companyId, input.companyId),
+            eq(nativeRunFinalizations.issueId, input.issueId),
+            eq(nativeRunFinalizations.phase, "committed"),
+            isNull(nativeRunFinalizations.leaseOwner),
+          ),
+        )
+        .for("update");
+      if (
+        !coordinator?.resultId ||
+        !coordinator.decisionId ||
+        !coordinator.assessmentId
+      )
+        return false;
      const [issue] = await tx
        .select()
        .from(issues)
@@ -816,25 +823,6 @@ export async function recoverCommittedNativeChatResponse(
        record(run.contextSnapshot).skipIssueComment === true
      )
        return false;
-      const [coordinator] = await tx
-        .select()
-        .from(nativeRunFinalizations)
-        .where(
-          and(
-            eq(nativeRunFinalizations.runId, run.id),
-            eq(nativeRunFinalizations.companyId, input.companyId),
-            eq(nativeRunFinalizations.issueId, input.issueId),
-            eq(nativeRunFinalizations.phase, "committed"),
-            isNull(nativeRunFinalizations.leaseOwner),
-          ),
-        )
-        .for("update");
-      if (
-        !coordinator?.resultId ||
-        !coordinator.decisionId ||
-        !coordinator.assessmentId
-      )
-        return false;
      if (input.onlyWhenDue) {
        if (coordinator.nextAttemptAt && coordinator.nextAttemptAt.getTime() > Date.now()) return false;
        const [pendingEffect] = await tx.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and(
@@ -1100,6 +1088,7 @@ export async function finalizeNativeRun(input: {
  skipNativeResponsePresentation?: boolean;
  skipNativeReviewResponse?: boolean;
  skipCommittedRunProjection?: boolean;
+  skipNativeRecoveryNotices?: boolean;
  /** Workspace-only replay must not consume the provider recovery budget. */
  preserveProviderAttempt?: boolean;
  failpoint?: NativeStatusCommitFailpoint;
@@ -1165,7 +1154,7 @@ export async function finalizeNativeRun(input: {
        logger.warn({ err, runId: admittedRun.id }, "Slack conversation settlement deferred to reconciliation");
      });
    }
-    await resolveCommittedFinalizationRecovery(input.db, admittedRun, coordinator.issueId);
+    if (!input.skipNativeRecoveryNotices) await resolveCommittedNativeRecoveryNotices(input.db, { companyId: admittedRun.companyId, runId: admittedRun.id, issueId: coordinator.issueId });
    if (input.projectRunStatus) {
      const [projectedRun] = await input.db.select({
        id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
@@ -1645,6 +1634,7 @@ export async function finalizeNativeRun(input: {
          .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId)))
          .for("update").limit(1);
        if (!before) return null;
+        await notifyNativeRecoveryNotices(tx, { companyId: run.companyId, runId: run.id });
        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 });
@@ -1681,7 +1671,7 @@ export async function finalizeNativeRun(input: {
          logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation");
        });
      }
-      if (finalizationPhase === "committed") await resolveCommittedFinalizationRecovery(input.db, run, coordinator.issueId);
+      if (finalizationPhase === "committed" && !input.skipNativeRecoveryNotices) await resolveCommittedNativeRecoveryNotices(input.db, { companyId: run.companyId, runId: run.id, issueId: coordinator.issueId });
      if (input.projectRunStatus && updatedRun) await reconcileMuseNativeTerminalRun(input.db, updatedRun);
      return {
        ...coordinator,
@@ -1,3 +1,4 @@
+import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js";
 import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js";
 import { notifyNativeRunProjection } from "../native-run-projection-notifications.js";
 import { randomUUID } from "node:crypto";
@@ -126,6 +127,7 @@ export async function projectCommittedRun(input: {
      nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt })
      .from(heartbeatRuns).where(needsRepair).for("update").limit(1);
    if (!before) return null;
+    await notifyNativeRecoveryNotices(tx, { companyId: input.run.companyId, runId: input.run.id });
    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 });
@@ -6,15 +6,18 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro
 import { NativeRunCoordinatorStore } from "./native-runtime/native-run-coordinator-store.js";
 import { getConversationOwnershipBlocker } from "./conversation-continuation.js";
 import { isExternalChatPresentationContext } from "./heartbeat-run-summary.js";
-import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js";
+import { finalizeNativeRun, recoverCommittedNativeChatResponse, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js";

-const support = await getEmbeddedPostgresTestSupport();
+import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js";
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PRESENTATION_READ_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
 describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read projections", () => {
-  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
  let db: ReturnType<typeof createDb>;
  beforeAll(async () => {
-    database = await startEmbeddedPostgresTestDatabase("presentation-ownership-read-");
-    db = createDb(database.connectionString);
+    if (!external) database = await startEmbeddedPostgresTestDatabase("presentation-ownership-read-");
+    db = createDb(external ?? database!.connectionString);
  }, 60_000);
  afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });

@@ -113,8 +116,8 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read
  async function compareRepair(f: Awaited<ReturnType<typeof fixture>>) {
    const baseline = captureReads(true), projected = captureReads();
    const input = { companyId: f.companyId, issueId: f.issueId, runId: f.run.id };
-    expect(await repairCommittedNativeChatResponse(projected.recorded, input))
-      .toBe(await repairCommittedNativeChatResponse(baseline.recorded, input));
+    expect(await recoverCommittedNativeChatResponse(projected.recorded, { ...input, onlyWhenDue: true }))
+      .toBe(await recoverCommittedNativeChatResponse(baseline.recorded, { ...input, onlyWhenDue: true }));
    expect(projected.rows.flat()).toHaveLength(1);
    expect(Buffer.byteLength(JSON.stringify(projected.rows))).toBeLessThan(4096);
    expect(Buffer.byteLength(JSON.stringify(baseline.rows))).toBeGreaterThan(128 * 1024);
@@ -123,7 +126,7 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read
  }

  it("preserves selected answers without downloading prompts, profiles, or excerpts", async () => {
-    const f = await fixture({ runtimeMode: "native", status: "succeeded", resultJson: {
+    const f = await repairFixture({ resultJson: {
      presentationDecision: { commentId: randomUUID() }, unknownFutureReceipt: { preserved: true } } });
    const row = await compareRepair(f);
    expect(row.resultJson).toEqual(f.run.resultJson);
@@ -167,6 +170,18 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read
    return { ...f, before, reply };
  }

+  async function repairFixture(patch: Partial<typeof heartbeatRuns.$inferInsert>) {
+
+
+    const f = await unpublishedAcceptedReply();
+    const [run] = await db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, f.run.id)).returning();
+
+
+    await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(Date.now() + 60_000) })
+      .where(eq(nativeRunFinalizations.runId, run.id));
+    return { ...f, run };
+  }
+
  it("publishes the same accepted reply and decision from equivalent full-row and projected starting states", async () => {
    const outcomes: unknown[] = [];
    for (const fullRows of [true, false]) {
@@ -200,7 +215,7 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read
    { source: "chat:fixture", skipIssueComment: true, paperclipWake: { externalInteractionContinuation: "true" } },
    { source: "internal", paperclipWake: { externalInteractionContinuation: true } },
  ].map(context => ({ context })))("preserves typed publication flags for $context", async ({ context }) => {
-    const f = await fixture({ runtimeMode: "native", status: "succeeded", contextSnapshot: context as Record<string, unknown> | null });
+    const f = await repairFixture({ contextSnapshot: context as Record<string, unknown> | null });
    const row = await compareRepair(f);
    expect(isExternalChatPresentationContext(row.contextSnapshot)).toBe(isExternalChatPresentationContext(context));
    const original = context && typeof context === "object" && !Array.isArray(context) ? context : {};
@@ -208,7 +223,7 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read
  });

  it.each([null, ["stdout", { receipt: true }], "scalar", 42].map(result => ({ result })))("preserves non-object result evidence $result", async ({ result }) => {
-    const f = await fixture({ runtimeMode: "native", status: "succeeded", resultJson: result as Record<string, unknown> | null });
+    const f = await repairFixture({ resultJson: result as Record<string, unknown> | null });
    expect((await compareRepair(f)).resultJson).toEqual(result);
  });
 });