feat: add native response recovery worker and deadline fences

Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-10-31 06:04:11 +01:00 · 2026-10-25 23:37:37 -05:00
1 parent 235a5d9007
commit 85c94ef85d
@@ -2308,3 +2308,22 @@ Slice 48 remains local and incomplete: owner error separation and the 33-case
 baseline pass; candidate discovery, producer coverage, lifecycle wiring, and
 periodic removal remain. Older unresolved stack gates and every other original
 recurring-server duty still require completion. The active goal is not complete.
+
+
+## Slice 48 local worker progress
+
+Accepted-response recovery now has a local scoped worker, locked retry/effect
+fences, milestone-topic notifications, and live authority registration invalidation.
+The final selected real-PostgreSQL/lifecycle run passes 45 cases, and server types
+pass. An elapsed-deadline race was reproduced before repair. A generated concurrent
+partial index plus expression statistics avoids selected-history scans in the
+10,000-row fixture; fresh migration and replay pass. See the
+[response audit](2026-10-25-native-response-presentation-sleep-audit.md) for exact
+scope and preserved failures.
+
+The worker is not wired into startup and periodic response recovery remains.
+Producer coverage, remaining variants/failure tests, large dormant-backlog cost,
+full local/remote acceptance, and PR publication are still required. Master was
+refreshed to `7703220d38`; the complete stack has not yet been reconciled with it.
+No merge, complete migration, whole-instance sleep, or full-goal completion is
+claimed.
@@ -1,6 +1,6 @@
 # Native accepted-response presentation recovery

-Status: source and baseline audit; no worker migration yet.
+Status: worker and owner guards implemented locally; producer audit and startup/periodic integration incomplete.
 Branch: `codex/sleep-48-native-response-presentation`.
 Parent: [#17690](https://github.com/paperclipai/paperclip/pull/17690),
 `56c3601ea605b333141e86e0a810ed790fb6ce48`.
@@ -133,3 +133,100 @@ source changes can authorize descendants. Audit its coverage for this duty befor
 adding caller wrappers. It already reaches status-decision settlement and run
 projection, but raw coordinator-store/session release and accepted-turn journal
 writers need individual inspection. Reuse is not proof that all producers exist.
+
+
+## Local worker and locked scheduling fences
+
+The local worker now discovers exact run/coordinator/result/decision bindings,
+filters selected answers and structurally irrelevant results, and dispatches
+through `recoverCommittedNativeChatResponse`. It does not own publication policy.
+`onlyWhenDue` rechecks the saved retry and outstanding status effects while the
+existing coordinator lock is held. Existing explicit callers retain their old
+behavior. A future retry is retained even if it becomes due between the locked
+check and the follow-up read; its next attempt is immediate rather than discarded.
+
+Operational errors remain retryable through the shared coordinator. A structurally
+eligible but unauthorized row becomes dormant until an eligibility mutation or
+live authority registration invalidates its scope. Selected comments, including
+deleted selections, remain excluded. No provider rerun or new work table is added.
+
+The existing milestone notification boundary now includes the response topic.
+Hints explicitly select company and issue IDs: structural TypeScript callers may
+pass larger objects, so serializing the whole argument is not safe. The first
+integration run exposed that mistake as circular-object serialization during
+status materialization; it was corrected before the passing runs below.
+
+Two narrow lifecycle capabilities support the new consumer:
+
+- `registerScopedDeadlineWork.invalidate` marks reconstruction dirty before waking.
+  A plain wake is insufficient after a prior dormant result removed its entry.
+  Existing consumers keep the same registration and scheduling behavior.
+- The committed-response authority owner exposes registration notifications.
+  Registering/replacing the actual chat service reactivates dormant responses.
+  The worker's `stop` removes this subscription; the delivery coordinator still
+  owns worker cancellation and awaiting in-flight work. Repeated unsubscription
+  cannot delete a later listener set. The authority remains scoped to the exact Db.
+
+No startup registration or periodic skip is wired yet. The old recovery path is
+still present. Full producer coverage, internal/continuation variants, recovery
+failure/drain tests, and repeated large-scope mutation cost remain acceptance work.
+Current discovery pages by run ID but reconstructs all matching pages in one
+load; its resource bound under a large unresolved/dormant backlog needs proof
+before activation. Do not describe the worker as a completed migration yet.
+
+## Query-plan and migration evidence
+
+The actual discovery query was captured from the driver. On a disposable database
+with 10,000 additional selected runs, results, assessments, decisions, and
+coordinators, it initially scanned completed history (22.656 ms). A partial
+unpresented-run index alone still chose the wrong join order (16.532 ms).
+Expression statistics for the selected-comment predicate allowed PostgreSQL to
+start from the one unpresented run (0.091 ms).
+
+Generated migration `0400_futuristic_earthquake.sql` creates the partial index
+concurrently, adds idempotent expression statistics, and analyzes the table.
+The existing top-level migration runner owns concurrent-index replay/repair.
+Snapshot retention contains exactly 0396–0400. Fresh migration, journal replay,
+direct SQL replay, ready/valid index flags, and statistics presence passed.
+
+Rechecking the actual migration's objects (after removing experimental objects)
+used the driver's real global, company, and issue-scoped queries. Global discovery
+found one candidate in 0.078 ms; history-only company/issue scopes returned none
+in 0.024/0.019 ms without sequential scans. These are query-plan fixtures, not
+latency or fleet-sleep claims. Large unselected ineligible histories and repeated
+configuration signals still need separate qualification.
+
+Evidence: `/tmp/sleep48-query.json`, `/tmp/sleep48-query-plan.sql`,
+`/tmp/sleep48-query-plan.log`, `/tmp/sleep48-query-statistics.log`,
+`/tmp/sleep48-production-plan.log`, `/tmp/sleep48-migration-verify.log`.
+
+## Current verification and preserved failures
+
+The expanded final selection passed 45 cases in two files: 33 existing accepted-
+response cases, ten new worker/owner cases, and two authority lifecycle cases.
+The 1,551 unselected chat cases are not claimed passed. Server typecheck passed.
+The worker cases establish startup recovery, selected-history quiet over former
+poll periods, real chat-service authority restoration, outer commit/rollback,
+first notification during drain, SQL-free drain waiting, resume, saved future
+retry, pending/failed status-effect ordering, and an elapsed-deadline race.
+
+The elapsed-deadline test failed before its fix: the worker returned null even
+though the locked owner had deferred the response. It now retains the due entry
+and the next attempt publishes exactly once. Log:
+`/tmp/sleep48-elapsed-deadline-before.log`.
+
+Earlier worker runs exposed the hint serialization defect (nine fixture failures),
+then two test-observation mistakes. A timer created before fake-time installation
+was not advanced by the fake clock; the test now installs time before registration.
+A null next deadline can occur while a sweep is in flight; commit assertions now
+wait for the persisted publication, zero tracked activity, and no next deadline
+jointly. The original logs remain `/tmp/sleep48-worker-tests.log`,
+`/tmp/sleep48-worker-tests-v2.log`, `/tmp/sleep48-drain-isolated.log`, and
+`/tmp/sleep48-worker-tests-v3.log`. The corrected initial selection passed eleven
+cases in `/tmp/sleep48-worker-tests-v4.log`; the final expanded result is
+`/tmp/sleep48-acceptance-v2.log` with server types in
+`/tmp/sleep48-acceptance-typecheck-v2.log`.
+
+No slice 48 PR, full workspace acceptance, remote CI/review, or provider sleep
+qualification is claimed by these local results. The earlier stack gates and
+remaining recurring-server duties are unchanged.
@@ -525,6 +525,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
  "0397_lumpy_talos.sql",
  "0398_shocking_molecule_man.sql",
  "0399_jazzy_ben_grimm.sql",
+  "0400_futuristic_earthquake.sql",
  "0380_funny_adam_warlock.sql",
 ]);

@@ -0,0 +1,3 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_native_unpresented_response_idx" ON "heartbeat_runs" USING btree ("company_id","native_issue_id","id") WHERE "heartbeat_runs"."runtime_mode" = 'native' AND "heartbeat_runs"."status" in ('succeeded', 'failed', 'timed_out') AND "heartbeat_runs"."finished_at" is not null AND jsonb_typeof("heartbeat_runs"."result_json" #> '{presentationDecision,commentId}') is distinct from 'string';--> statement-breakpoint
+CREATE STATISTICS IF NOT EXISTS "heartbeat_runs_native_response_selection_stats" ON (jsonb_typeof("result_json" #> '{presentationDecision,commentId}')) FROM "heartbeat_runs";--> statement-breakpoint
+ANALYZE "heartbeat_runs";
@@ -1,6 +1,6 @@
 {
-  "id": "e9c56475-b2c8-4cb4-b7dd-89bfa645a93a",
-  "prevId": "90ac76c4-b6d8-4c38-b4b8-d51699ca2ec1",
+  "id": "d51b06de-ba83-4221-8ea1-863fd947f021",
+  "prevId": "5b5659ef-e4b6-4584-9b4f-ed075047ef2f",
  "version": "7",
  "dialect": "postgresql",
  "tables": {
@@ -2637,6 +2637,28 @@
        }
      },
      "indexes": {
+        "agent_instruction_copies_collection_idx": {
+          "name": "agent_instruction_copies_collection_idx",
+          "columns": [
+            {
+              "expression": "run_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "NOT (coalesce(\"agent_instruction_working_copies\".\"receipt\", '{}'::jsonb) ? 'retainedByRunId') AND (\n      (\"agent_instruction_working_copies\".\"attempts\" < 3 AND (\"agent_instruction_working_copies\".\"state\" IN ('prepared', 'pending_collection', 'unchanged_turn', 'warm_saved')\n        OR (\"agent_instruction_working_copies\".\"state\" = 'preparing' AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1')))\n      OR (\"agent_instruction_working_copies\".\"state\" = 'unavailable' AND \"agent_instruction_working_copies\".\"process_stopped_at\" IS NULL\n        AND \"agent_instruction_working_copies\".\"location\" LIKE 'remote:%' AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1'))",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
        "agent_instruction_copies_cleanup_idx": {
          "name": "agent_instruction_copies_cleanup_idx",
          "columns": [
@@ -33649,6 +33671,56 @@
        }
      },
      "indexes": {
+        "heartbeat_runs_native_unpresented_response_idx": {
+          "name": "heartbeat_runs_native_unpresented_response_idx",
+          "columns": [
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "native_issue_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" in ('succeeded', 'failed', 'timed_out') AND \"heartbeat_runs\".\"finished_at\" is not null AND jsonb_typeof(\"heartbeat_runs\".\"result_json\" #> '{presentationDecision,commentId}') is distinct from 'string'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
+        "heartbeat_runs_cancelled_workspace_preservation_idx": {
+          "name": "heartbeat_runs_cancelled_workspace_preservation_idx",
+          "columns": [
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" = 'cancelled'\n        AND \"heartbeat_runs\".\"runner_profile_json\"->'nativeWorkspaceSync'->>'state' = 'prepared'\n        AND \"heartbeat_runs\".\"result_json\"->'nativeCancellation'->>'dispatchState' = 'acknowledged'\n        AND (\"heartbeat_runs\".\"result_json\"->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'\n          AND \"heartbeat_runs\".\"result_json\"->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
        "heartbeat_runs_retained_recovery_adoption_idx": {
          "name": "heartbeat_runs_retained_recovery_adoption_idx",
          "columns": [
@@ -42830,6 +42902,62 @@
        }
      },
      "indexes": {
+        "native_run_finalizations_pending_effect_owner_idx": {
+          "name": "native_run_finalizations_pending_effect_owner_idx",
+          "columns": [
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "issue_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "decision_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"native_run_finalizations\".\"decision_id\" is not null AND \"native_run_finalizations\".\"result_id\" is not null AND \"native_run_finalizations\".\"phase\" <> 'terminal_failure'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
+        "native_run_finalizations_unsafe_export_recovery_idx": {
+          "name": "native_run_finalizations_unsafe_export_recovery_idx",
+          "columns": [
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "issue_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "run_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"native_run_finalizations\".\"phase\" = 'terminal_failure' AND \"native_run_finalizations\".\"failure_code\" = 'native_workspace_sync_out_unsafe_archive'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
        "native_run_finalizations_control_deadline_idx": {
          "name": "native_run_finalizations_control_deadline_idx",
          "columns": [
@@ -52016,6 +52144,34 @@
        }
      },
      "indexes": {
+        "status_decision_effects_pending_delivery_idx": {
+          "name": "status_decision_effects_pending_delivery_idx",
+          "columns": [
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "issue_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "decision_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"status_decision_effects\".\"delivery_state\" in ('pending', 'failed')",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
        "status_decision_effects_decision_ordinal_uq": {
          "name": "status_decision_effects_decision_ordinal_uq",
          "columns": [
@@ -2787,6 +2787,13 @@
      "when": 1792986633369,
      "tag": "0399_jazzy_ben_grimm",
      "breakpoints": true
+    },
+    {
+      "idx": 400,
+      "version": "7",
+      "when": 1792989084087,
+      "tag": "0400_futuristic_earthquake",
+      "breakpoints": true
    }
  ]
 }
@@ -111,6 +111,9 @@ export const heartbeatRuns = pgTable(
    updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
  },
  (table) => ({
+    nativeUnpresentedResponseIdx: index("heartbeat_runs_native_unpresented_response_idx")
+      .on(table.companyId, table.nativeIssueId, table.id)
+      .where(sql`${table.runtimeMode} = 'native' AND ${table.status} in ('succeeded', 'failed', 'timed_out') AND ${table.finishedAt} is not null AND jsonb_typeof(${table.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`),
    cancelledWorkspacePreservationIdx: index("heartbeat_runs_cancelled_workspace_preservation_idx").on(table.id, table.companyId)
      .where(sql`${table.runtimeMode} = 'native' AND ${table.status} = 'cancelled'
        AND ${table.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared'
@@ -1,3 +1,5 @@
+import { nativeResponsePresentationWork } from "../services/native-runtime/native-response-presentation-work.js";
+import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifications.js";
 import { registerGitHubReviewProjectionWork } from "../services/chat-channels/github/review-projection-work.js";
 import { retainGitHubReviewReceipt } from "../services/chat-channels/github/task-preparation.js";
 import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js";
@@ -132,6 +134,7 @@ import {
  issues,
  nativeRunResults,
  nativeRunFinalizations,
+  statusDecisionEffects,
  environmentLeases,
  environments,
  principalPermissionGrants,
@@ -72595,6 +72598,191 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
    },
  );

+  it("recovers accepted-response work on startup and leaves selected history quiet", async () => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const work = nativeResponsePresentationWork(db);
+    const worker = work.register(coordinator, () => true);
+    try {
+      await worker.ready;
+      await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
+      const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId));
+      expect(comments).toHaveLength(1);
+      expect(comments[0].body).toBe(context.result.summary);
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+      expect(await work.load({ companyId: context.fixture.companyId })).toEqual([]);
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+      const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
+      try {
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
+      } finally { reads.mockRestore(); writes.mockRestore(); }
+      expect(errors).toEqual([]);
+    } finally { worker.stop(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); }
+  });
+
+  it("reactivates dormant accepted-response work when the live chat authority registers", async () => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    await context.service.shutdown();
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const worker = nativeResponsePresentationWork(db).register(coordinator, () => true);
+    let replacement: ReturnType<typeof createService> | undefined;
+    try {
+      await worker.ready;
+      await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+      replacement = createService(new FakeChatSdkRuntime(), fakeTelegramFetch() as typeof fetch);
+      await expect.poll(async () => (await readChatPublicationsByEndpointId(context.endpoint.id)).length, { timeout: 5000 }).toBe(1);
+      await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
+      expect(errors).toEqual([]);
+    } finally { worker.stop(); await coordinator.stop(); await replacement?.service.shutdown(); }
+  });
+
+  it.each(["commit", "rollback"])("keeps accepted-response work behind the outer %s boundary", async outcome => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, context.runId));
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const worker = nativeResponsePresentationWork(db).register(coordinator, () => true);
+    try {
+      await worker.ready;
+      await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
+      const transaction = db.transaction(async tx => {
+        await tx.transaction(async nested => {
+          await notifyChatRunMilestones(nested, { companyId: context.fixture.companyId, issueId: context.issue.id });
+          await nested.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, context.runId));
+        });
+        expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+        expect(idleWorkSnapshot().active).toBeGreaterThan(0);
+        if (outcome === "rollback") throw new Error("response fixture rollback");
+      });
+      if (outcome === "rollback") await expect(transaction).rejects.toThrow("response fixture rollback");
+      else await transaction;
+      await expect.poll(async () => ({
+        publications: (await readChatPublicationsByEndpointId(context.endpoint.id)).length,
+        active: idleWorkSnapshot().active,
+        next: coordinator.nextWakeAt(),
+      }), { timeout: 5000 }).toEqual({ publications: outcome === "commit" ? 1 : 0, active: 0, next: null });
+      expect(errors).toEqual([]);
+    } finally { worker.stop(); await coordinator.stop(); await context.service.shutdown(); }
+  });
+
+  it("retains the first accepted-response wake through idle drain without passive SQL", async () => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, context.runId));
+    let canRun = true;
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, canReconcile: () => true, onError: error => errors.push(error) });
+    const worker = nativeResponsePresentationWork(db).register(coordinator, () => canRun);
+    try {
+      await worker.ready;
+      await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
+      canRun = false;
+      await db.transaction(async tx => {
+        await notifyChatRunMilestones(tx, { companyId: context.fixture.companyId, issueId: context.issue.id });
+        await tx.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, context.runId));
+      });
+      await expect.poll(() => idleWorkSnapshot().active, { timeout: 5000 }).toBe(0);
+      expect(coordinator.nextWakeAt()).not.toBeNull();
+      const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
+      try {
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
+      } finally { reads.mockRestore(); writes.mockRestore(); }
+      canRun = true; worker.wake();
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+        expect(coordinator.nextWakeAt()).toBeNull();
+      }, { timeout: 5000 });
+      expect(errors).toEqual([]);
+    } finally { worker.stop(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); }
+  });
+
+  it("reconstructs a future accepted-response deadline without early publication", async () => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    const at = new Date(Date.now() + 60_000);
+    await db.update(nativeRunFinalizations).set({ nextAttemptAt: at }).where(eq(nativeRunFinalizations.runId, context.runId));
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const worker = nativeResponsePresentationWork(db).register(coordinator, () => true);
+    try {
+      await worker.ready;
+      expect(coordinator.nextWakeAt()).toBe(at.getTime());
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+      await vi.advanceTimersByTimeAsync(Math.max(0, at.getTime() - Date.now()));
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+        expect(coordinator.nextWakeAt()).toBeNull();
+      }, { timeout: 5000 });
+      expect(errors).toEqual([]);
+    } finally { worker.stop(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); }
+  });
+
+  it("retains an accepted-response deadline that elapses during the locked recovery check", async () => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    const work = nativeResponsePresentationWork(db);
+    const [entry] = await work.load({ companyId: context.fixture.companyId });
+    expect(entry?.id).toBe(context.runId);
+    vi.useFakeTimers({ toFake: ["Date"] });
+    const due = new Date(Date.now() + 1000);
+    const transaction = db.transaction.bind(db);
+    const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => {
+      await transaction(async tx => {
+        await tx.update(nativeRunFinalizations).set({ nextAttemptAt: due }).where(eq(nativeRunFinalizations.runId, context.runId));
+      });
+      const result = await transaction(callback, options);
+      vi.setSystemTime(due.getTime() + 1);
+      return result;
+    }) as Db["transaction"]);
+    try {
+      const deferred = await work.check(entry);
+      intercept.mockRestore();
+      expect(deferred).toEqual({ ...entry, dueAt: due.getTime() });
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+      await expect(work.check(deferred!)).resolves.toBeNull();
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+    } finally { intercept.mockRestore(); vi.useRealTimers(); await context.service.shutdown(); }
+  });
+
+  it("defers scheduled accepted-response recovery until the persisted retry is due", async () => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    const input = { companyId: context.fixture.companyId, issueId: context.issue.id, runId: context.runId, onlyWhenDue: true };
+    try {
+      await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(Date.now() + 60_000) })
+        .where(eq(nativeRunFinalizations.runId, context.runId));
+      await expect(recoverCommittedNativeChatResponse(db, input)).resolves.toBe(false);
+      expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId))).toHaveLength(0);
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+      await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0) })
+        .where(eq(nativeRunFinalizations.runId, context.runId));
+      await expect(recoverCommittedNativeChatResponse(db, input)).resolves.toBe(true);
+      expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId))).toHaveLength(1);
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+    } finally { await context.service.shutdown(); }
+  });
+
+  it.each(["pending", "failed"])("defers scheduled accepted-response recovery behind %s status effects", async deliveryState => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    const input = { companyId: context.fixture.companyId, issueId: context.issue.id, runId: context.runId, onlyWhenDue: true };
+    try {
+      const [effect] = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, context.coordinator.decisionId!));
+      expect(effect).toBeDefined();
+      await db.update(statusDecisionEffects).set({ deliveryState }).where(eq(statusDecisionEffects.id, effect.id));
+      await expect(recoverCommittedNativeChatResponse(db, input)).resolves.toBe(false);
+      expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId))).toHaveLength(0);
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+      await db.update(statusDecisionEffects).set({ deliveryState: "delivered" }).where(eq(statusDecisionEffects.id, effect.id));
+      await expect(recoverCommittedNativeChatResponse(db, input)).resolves.toBe(true);
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
+    } finally { await context.service.shutdown(); }
+  });
+
  it("leaves a contended accepted response retryable without reclassifying or partially publishing it", async () => {
    const context = await committedChatResponseRecoveryFixture("telegram");
    let release!: () => void;
@@ -0,0 +1,41 @@
+import { describe, expect, it, vi } from "vitest";
+import type { Db } from "@paperclipai/db";
+import { authorizeCommittedChatResponse, onCommittedChatResponseAuthorityRegistered, registerCommittedChatResponseAuthority } from "../services/durable-chat-wakeup.js";
+
+describe("committed response authority lifecycle", () => {
+  it("reactivates only its database and unsubscribes without removing a replacement listener", () => {
+    const db = {} as Db, otherDb = {} as Db;
+    const first = vi.fn(), second = vi.fn(), other = vi.fn();
+    const disposeFirst = onCommittedChatResponseAuthorityRegistered(db, first);
+    const disposeOther = onCommittedChatResponseAuthorityRegistered(otherDb, other);
+    const removeAuthority = registerCommittedChatResponseAuthority(db, async () => {});
+    expect(first).toHaveBeenCalledTimes(1);
+    expect(other).not.toHaveBeenCalled();
+    disposeFirst();
+    const disposeSecond = onCommittedChatResponseAuthorityRegistered(db, second);
+    disposeFirst();
+    const removeReplacement = registerCommittedChatResponseAuthority(db, async () => {});
+    expect(first).toHaveBeenCalledTimes(1);
+    expect(second).toHaveBeenCalledTimes(1);
+    removeAuthority();
+    expect(second).toHaveBeenCalledTimes(1);
+    disposeSecond(); disposeOther(); removeReplacement();
+    const removeLast = registerCommittedChatResponseAuthority(db, async () => {});
+    expect(second).toHaveBeenCalledTimes(1);
+    removeLast();
+  });
+
+  it("retains the replacement authority when the old service shuts down", async () => {
+    const db = {} as Db;
+    const first = vi.fn(async () => {}), second = vi.fn(async () => {});
+    const removeFirst = registerCommittedChatResponseAuthority(db, first);
+    const removeSecond = registerCommittedChatResponseAuthority(db, second);
+    try {
+      removeFirst();
+      const input = { companyId: "company", issueId: "issue", agentId: "agent", runId: "run", resultId: "result" };
+      await authorizeCommittedChatResponse(db, db, input);
+      expect(first).not.toHaveBeenCalled();
+      expect(second).toHaveBeenCalledExactlyOnceWith(db, input);
+    } finally { removeSecond(); }
+  });
+});
@@ -8,6 +8,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type ChatRunMilestoneScope = { companyId: string; issueId?: string };

 export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeResponsePresentation, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewChecks, JSON.stringify({ companyId: scope.companyId }));
  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";

 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  nativeResponsePresentation: "native-response-presentation",
  nativePendingEffects: "native-pending-effects",
  unsafeWorkspaceExportRecovery: "unsafe-workspace-export-recovery",
  automaticCompletionReviews: "automatic-completion-reviews",
@@ -21,6 +21,21 @@ const committedChatResponseAuthorities = new WeakMap<
  CommittedChatResponseAuthority
 >();

+const committedChatResponseAuthorityListeners = new WeakMap<Db, Set<() => void>>();
+
+export function onCommittedChatResponseAuthorityRegistered(db: Db, listener: () => void): () => void {
+  let listeners = committedChatResponseAuthorityListeners.get(db);
+  if (!listeners) {
+    listeners = new Set();
+    committedChatResponseAuthorityListeners.set(db, listeners);
+  }
+  listeners.add(listener);
+  return () => {
+    listeners.delete(listener);
+    if (listeners.size === 0 && committedChatResponseAuthorityListeners.get(db) === listeners) committedChatResponseAuthorityListeners.delete(db);
+  };
+}
+
 export class CommittedChatResponseAuthorizationError extends HttpError {
  constructor() {
    super(
@@ -41,6 +56,7 @@ export function registerCommittedChatResponseAuthority(
  authority: CommittedChatResponseAuthority,
 ): () => void {
  committedChatResponseAuthorities.set(db, authority);
+  for (const listener of committedChatResponseAuthorityListeners.get(db) ?? []) listener();
  return () => {
    if (committedChatResponseAuthorities.get(db) === authority) {
      committedChatResponseAuthorities.delete(db);
@@ -0,0 +1,91 @@
+import { and, asc, eq, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm";
+import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, type Db } from "@paperclipai/db";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import type { ChatRunMilestoneScope } from "../chat-run-milestone-notifications.js";
+import { onCommittedChatResponseAuthorityRegistered } from "../durable-chat-wakeup.js";
+import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
+import { recoverCommittedNativeChatResponse } from "./native-run-finalizer.js";
+
+type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
+
+export function nativeResponsePresentationWork(db: Db) {
+  function candidates(scope?: ChatRunMilestoneScope, after?: string, runId?: string) {
+    return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
+      issueId: nativeRunFinalizations.issueId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
+    }).from(nativeRunFinalizations)
+      .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
+        eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId)))
+      .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId),
+        eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), eq(nativeRunResults.issueId, nativeRunFinalizations.issueId),
+        eq(nativeRunResults.runId, heartbeatRuns.id), eq(nativeRunResults.completionContractId, heartbeatRuns.completionContractId)))
+      .innerJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId),
+        eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId),
+        eq(statusDecisions.runId, heartbeatRuns.id), eq(statusDecisions.assessmentId, nativeRunFinalizations.assessmentId)))
+      .where(and(
+        eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out"]),
+        isNotNull(heartbeatRuns.finishedAt), eq(nativeRunFinalizations.phase, "committed"), isNull(nativeRunFinalizations.leaseOwner),
+        eq(nativeRunResults.schemaStatus, "accepted"),
+        sql`jsonb_typeof(${heartbeatRuns.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`,
+        sql`coalesce(${statusDecisions.decisionJson}->'externalChatReviewPresentation', 'null'::jsonb) in ('null'::jsonb, 'false'::jsonb, '0'::jsonb, '""'::jsonb)`,
+        sql`${nativeRunResults.resultJson} #>> '{result,schema}' = 'paperclip.run_result.v1'`,
+        sql`${nativeRunResults.resultJson} #>> '{terminal,runTerminalState}' = 'succeeded'`,
+        sql`${nativeRunResults.resultJson} #>> '{terminal,turnTerminalState}' = 'completed'`,
+        or(
+          and(eq(heartbeatRuns.status, "succeeded"),
+            sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'done'`,
+            sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'done'`),
+          and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`,
+            sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'yielded'`,
+            sql`${nativeRunResults.resultJson} #>> '{result,continuation,kind}' in ('response_wake', 'same_agent', 'monitor')`),
+        ),
+        notExists(db.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and(
+          eq(statusDecisionEffects.companyId, nativeRunFinalizations.companyId),
+          eq(statusDecisionEffects.issueId, nativeRunFinalizations.issueId),
+          eq(statusDecisionEffects.decisionId, nativeRunFinalizations.decisionId),
+          inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]),
+        ))),
+        scope ? and(eq(heartbeatRuns.companyId, scope.companyId), scope.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined) : undefined,
+        after ? gt(heartbeatRuns.id, after) : undefined, runId ? eq(heartbeatRuns.id, runId) : undefined,
+      )).orderBy(asc(heartbeatRuns.id)).limit(100);
+  }
+  async function load(scope?: ChatRunMilestoneScope, signal?: AbortSignal, canRun = () => true): Promise<Entry[]> {
+    const entries: Entry[] = [];
+    let after: string | undefined;
+    while (canRun() && !signal?.aborted) {
+      const rows = await candidates(scope, after);
+      for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
+      if (rows.length < 100) return entries;
+      after = rows[rows.length - 1].id;
+    }
+    throw new Error("Native response presentation discovery stopped");
+  }
+  async function check(entry: Entry, signal?: AbortSignal): Promise<Entry | null> {
+    if (signal?.aborted) throw new Error("Native response presentation recovery stopped");
+    const scope = { companyId: entry.companyId, issueId: entry.issueId };
+    const [current] = await candidates(scope, undefined, entry.id);
+    if (!current) return null;
+    if (current.nextAttemptAt && current.nextAttemptAt.getTime() > Date.now()) return { ...entry, dueAt: current.nextAttemptAt.getTime() };
+    if (signal?.aborted) throw new Error("Native response presentation recovery stopped");
+    const recoveryStartedAt = Date.now();
+    await recoverCommittedNativeChatResponse(db, { ...scope, runId: entry.id, onlyWhenDue: true });
+    const [remaining] = await candidates(scope, undefined, entry.id);
+    return remaining?.nextAttemptAt && remaining.nextAttemptAt.getTime() > recoveryStartedAt
+      ? { ...entry, dueAt: remaining.nextAttemptAt.getTime() } : null;
+  }
+  function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
+    const worker = registerScopedDeadlineWork(coordinator, {
+      queue: DELIVERY_QUEUES.nativeResponsePresentation,
+      decodeScope: (hint: string): ChatRunMilestoneScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (entry: Entry, scope?: ChatRunMilestoneScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId),
+      load: (scope, signal) => load(scope, signal, canRun),
+      check: (entry, signal) => {
+        if (!canRun()) throw new Error("Native response presentation recovery paused");
+        return check(entry, signal);
+      },
+    }, canRun);
+    const unsubscribe = onCommittedChatResponseAuthorityRegistered(db, () => worker.invalidate());
+    return { ...worker, stop: unsubscribe };
+  }
+  return { load, check, register };
+}
@@ -37,6 +37,7 @@ import {
  nativeRunFinalizations,
  nativeRunResults,
  statusDecisions,
+  statusDecisionEffects,
  workAssessments,
  workspaceOperations,
 } from "@paperclipai/db";
@@ -857,7 +858,7 @@ export async function repairCommittedNativeChatResponse(

 export async function recoverCommittedNativeChatResponse(
  db: Db,
-  input: { companyId: string; issueId: string; runId: string },
+  input: { companyId: string; issueId: string; runId: string; onlyWhenDue?: boolean },
 ): Promise<boolean> {
  let agentId: string | null = null;
  let materialized = false;
@@ -920,6 +921,16 @@ export async function recoverCommittedNativeChatResponse(
        !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(
+          eq(statusDecisionEffects.companyId, input.companyId),
+          eq(statusDecisionEffects.issueId, input.issueId),
+          eq(statusDecisionEffects.decisionId, coordinator.decisionId),
+          inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]),
+        )).limit(1);
+        if (pendingEffect) return false;
+      }
      const [accepted] = await tx
        .select()
        .from(nativeRunResults)
@@ -90,5 +90,8 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
    hasPending: async () => canRun() && dirty.size > 0,
    nextRunAt: next,
  });
-  return worker;
+  return { ...worker, invalidate(hint = "*") {
+    dirty.add(hint);
+    worker.wake();
+  } };
 }