From 898144f4b3025ecca5ce7907069e19660298bf7b Mon Sep 17 00:00:00 2001
From: Dotta <34892728+cryppadotta@users.noreply.github.com>
Date: Wed, 21 Oct 2026 09:54:28 -0500
Subject: [PATCH] fix: stop idle execution-control recovery polling (#17094)

Replace idle execution-control polling with workers driven by committed writes and persisted deadlines. Preserve startup recovery, bounded retries, and shutdown safety. Stabilize the Slack setup browser regression by waiting for the restored setup screen before simulating its webhook.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
---
 doc/idle-execution-control.md                 |  71 +++++
 ...-instance-sleep-and-scheduler-inventory.md |  14 +-
 .../__tests__/execution-control-work.test.ts  | 289 ++++++++++++++++++
 .../heartbeat-process-recovery.test.ts        |  33 +-
 .../__tests__/issue-recovery-actions.test.ts  |   1 +
 .../server-startup-feedback-export.test.ts    |  18 +-
 server/src/index.ts                           |  39 +--
 .../services/delivery-work-notifications.ts   |   6 +
 .../execution-control-notifications.ts        |  20 ++
 .../execution-control-reconciliation.ts       |   2 +
 server/src/services/execution-control-work.ts |  98 ++++++
 .../services/execution-recovery-resolution.ts | 188 ++++++------
 .../services/explicit-native-continuation.ts  |  24 +-
 server/src/services/heartbeat.ts              |   6 +-
 .../src/services/heartbeat/run-lifecycle.ts   |   5 +-
 .../services/heartbeat/runtime-execution.ts   |  10 +-
 server/src/services/issue-recovery-actions.ts |  24 +-
 .../src/services/legacy-execution-recovery.ts |   2 +
 server/src/services/local-ai-login.ts         |   2 +
 .../native-failed-workspace-recovery.ts       |   2 +
 .../native-runtime/native-restart-recovery.ts |   2 +
 .../native-runtime/native-run-finalizer.ts    |  16 +-
 .../native-runtime/native-safe-replacement.ts |  29 +-
 .../native-session-executor.test.ts           |  25 +-
 .../native-runtime/native-session-executor.ts |  12 +-
 tests/e2e/chat-adapters-ui-providers.spec.ts  |   1 +
 26 files changed, 737 insertions(+), 202 deletions(-)
 create mode 100644 doc/idle-execution-control.md
 create mode 100644 server/src/__tests__/execution-control-work.test.ts
 create mode 100644 server/src/services/execution-control-notifications.ts
 create mode 100644 server/src/services/execution-control-work.ts

diff --git a/doc/idle-execution-control.md b/doc/idle-execution-control.md
new file mode 100644
index 00000000000..a18600124ec
--- /dev/null
+++ b/doc/idle-execution-control.md
@@ -0,0 +1,71 @@
+# Execution-control workers and idle instances
+
+Execution-control maintenance uses the shared delivery coordinator instead of an
+unconditional 15-second interval. This changes when recovery is inspected; it
+does not change the authority, evidence, locking, idempotency, or cleanup rules
+inside the six owning services. No table or migration is added.
+
+## Duties and durable state
+
+| Duty | Trigger and durable source | Remaining scheduled work | Quiet condition |
+| --- | --- | --- | --- |
+| Abandoned finalization | Committed control-deadline writes on `heartbeat_runs` and `native_run_finalizations` | Earliest non-null control deadline; a due row still present after recovery retries after 15 seconds | No control deadlines |
+| Safe native replacement | Committed run transitions and reconciliation decisions; existing failed run/coordinator rows | Retry while the existing replacement predicate selects a candidate, including one awaiting provider termination or sandbox release | No replacement candidate, or a recorded successor/denial |
+| Reconciled continuation | Committed recovery-action marking/resolution; `evidence.continuationDelivery` | Retry pending delivery after 15 seconds | No resolved action awaiting continuation delivery |
+| Execution status | Committed run status writes; `execution_status_delivery_id` | Retry while a delivery marker remains | All markers acknowledged |
+| Automatic disposition | Committed run/recovery-action changes; existing eligible recovery-action queries | Retry while a selected action still needs disposition or an obsolete conversation hold remains | No eligible action/hold; resolved history alone is quiet |
+| Local login cleanup | Committed local login creation; existing `adapter_auth_sessions.expires_at` | Earliest local login expiry; retry overdue failed cleanup | No local login expiry remains |
+
+Each worker runs once on startup to restore durable obligations. Passes are
+coalesced at the former 15-second resolution: a tenant's repeated writes cannot
+force continuous global scans. This is a minimum spacing between requested
+passes, not a recurring poll; an empty worker never arms that spacing by itself.
+A newly discovered deadline within the spacing window runs at its end, retaining
+the former maximum 15-second scheduling resolution. Future deadlines outside
+that window retain their exact due time. Workers remain
+independent: a slow replacement proof cannot stall status delivery. Batches and
+failed obligations retain the existing 15-second retry budget; empty workers do
+not repeatedly inspect the database. Login authorization still checks expiry on
+use even before physical cleanup completes.
+
+## Producer contract
+
+`notifyExecutionRunWork(transaction)` signals the run transition's status,
+finalization, replacement, and disposition consumers before the write. Existing
+transaction owners call it directly. Standalone deadline and recovery-action
+writes use `writeExecutionControl` to give the write a transaction boundary.
+Nested transactions retain the existing outer-commit notification semantics.
+
+Producer coverage includes heartbeat status transitions, legacy terminalization,
+native finalization and restart/failure settlement, replacement successors and
+denials, recovery-action upserts/resolution, operator reconciliation, control
+deadline creation/cancellation, and local login creation. Workers acknowledge
+existing records rather than creating a second queue. The replacement and
+disposition pending checks reuse the executing services' eligibility queries.
+
+This is the existing single live server / shared `createDb` owner signal model.
+Independent SQL writers or separate replicas must integrate notifications; these
+workers do not continuously discover unannounced external writes. Startup still
+recovers committed rows after process loss. Uncertain commit handling remains in
+`packages/db/src/work-signals.ts`.
+
+## Lifecycle and verification
+
+A worker consumes a requested generation only after its sweep and deadline
+inspection succeed. Writes during a sweep, startup during drain, and a failure
+before drain remain pending for resume. Passive waiting during drain performs no
+worker SQL. Shutdown uses the coordinator to unsubscribe and await in-flight work.
+
+`server/src/__tests__/execution-control-work.test.ts` covers empty and retained
+history with the real sweep implementations, a real producer inside an outer
+commit/rollback, saved deadlines, deadline replacement/cancellation, durable
+status retries, recovery continuations, independent duties, and drain/resume.
+Database tests use embedded PostgreSQL by default. An isolated external test
+instance can be supplied through `PAPERCLIP_EXECUTION_SLEEP_TEST_DATABASE_URL`;
+the database must have a dedicated name ending in `_test` and is cleared by the
+suite. Never use a tenant database.
+
+Worker quiet is not host sleep authorization. The existing idle-safety blockers
+remain, and Cloud's external wake contract does not gain coverage for execution
+or login deadlines through this change. Provider suspension and wake must be
+qualified separately before relaxing those blockers.
diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md
index 35db2511e4a..e77a8719c71 100644
--- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md
+++ b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md
@@ -536,13 +536,13 @@ Longer polling intervals alone do not complete an item.
   retain subscription scope, expiry, and replay safety. Source:
   [events.ts](../../server/src/services/public-mcp/events.ts).
 
-- [ ] **JOB-06 — Execution-control reconciliation.** Current: independent
-  **15s** recovery/delivery lanes, including when the main heartbeat is disabled.
-  Change: activate lanes from durable pending work and deadlines. **Complete
-  when:** empty lanes are quiet and abandoned finalizations, replacements,
-  continuations, status/disposition delivery, and login expiry still recover.
-  Sources: [index.ts](../../server/src/index.ts),
-  [execution-control-deadline.ts](../../server/src/services/execution-control-deadline.ts).
+- [x] **JOB-06 — Execution-control reconciliation.** Implemented in
+  [PR #17094](https://github.com/paperclipai/paperclip/pull/17094): six independent
+  workers wake from committed writes, saved control/login deadlines, or concrete
+  unfinished obligations. Empty queues and completed history no longer cause
+  recurring SQL. Existing 15s retries remain for pending work; Cloud sleep
+  eligibility and external alarm coverage are unchanged. See the
+  [worker and producer contract](../idle-execution-control.md).
 
 - [ ] **JOB-07 — Plugin job scheduling.** Current: **30s** SQL even without jobs.
   Change: register nearest `nextRunAt` and signal registration/update/removal.
diff --git a/server/src/__tests__/execution-control-work.test.ts b/server/src/__tests__/execution-control-work.test.ts
new file mode 100644
index 00000000000..9d107b75dd7
--- /dev/null
+++ b/server/src/__tests__/execution-control-work.test.ts
@@ -0,0 +1,289 @@
+import { writeExecutionControl } from "../services/execution-control-notifications.js";
+import { reconcileSafeNativeReplacements } from "../services/native-runtime/native-safe-replacement.js";
+import { localAiLoginService } from "../services/local-ai-login.js";
+import { randomUUID } from "node:crypto";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { eq, sql } from "drizzle-orm";
+import { agents, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations, applyPendingMigrations } from "@paperclipai/db";
+import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js";
+import { executionControlSchedule, registerExecutionControlWork } from "../services/execution-control-work.js";
+import { deliverExecutionStatuses } from "../services/execution-status-delivery.js";
+import { terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
+import { reconcileAbandonedExecutionControl } from "../services/execution-control-reconciliation.js";
+import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
+import { markExecutionReconciliation, deliverReconciledExecutions, settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+
+const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
+afterEach(async () => { for (const worker of coordinators.splice(0)) await worker.stop(); vi.useRealTimers(); });
+function coordinator(owner: object, canRun = () => true, onError: (error: unknown) => void = vi.fn()) {
+  const c = createDeliveryWorkCoordinator({ owner, canRun, canReconcile: () => true, onError });
+  coordinators.push(c); return c;
+}
+function fakeOwner() {
+  const execute = vi.fn(async () => [{ xid: "42" }]);
+  const owner = installDatabaseWorkSignals({ execute, async transaction(fn: (tx: any) => Promise<any>): Promise<any> {
+    return fn({ execute, transaction: owner.transaction });
+  } });
+  return owner;
+}
+
+describe("execution-control lifecycle", () => {
+  it.each(["startup", "commit", "failure"])("retains %s work through drain without querying while paused", async kind => {
+    vi.useFakeTimers();
+    const owner = fakeOwner(); let draining = kind === "startup";
+    const errors = vi.fn(); const c = coordinator(owner, () => !draining, errors);
+    const run = vi.fn(async () => {}), next = vi.fn(async () => null);
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, run, next }]).ready;
+    if (kind !== "startup") {
+      if (kind === "failure") run.mockRejectedValueOnce(new Error("delivery unavailable"));
+      else draining = true;
+      await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus));
+      await vi.advanceTimersByTimeAsync(15_000);
+      draining = true;
+    }
+    const reads = next.mock.calls.length, runs = run.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(next).toHaveBeenCalledTimes(reads); expect(run).toHaveBeenCalledTimes(runs);
+    draining = false;
+    await vi.advanceTimersByTimeAsync(15_000);
+    expect(run).toHaveBeenCalledTimes(runs + 1);
+    expect(c.nextWakeAt()).toBeNull();
+  });
+  it("keeps a change arriving during a sweep and does not overlap it", async () => {
+    vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner);
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const run = vi.fn(async () => {});
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, run, next: async () => null }]).ready;
+    run.mockImplementationOnce(() => gate);
+    await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus));
+    await vi.advanceTimersByTimeAsync(15_000);
+    await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus));
+    await vi.advanceTimersByTimeAsync(1);
+    expect(run).toHaveBeenCalledTimes(2);
+    release(); await vi.advanceTimersByTimeAsync(15_000);
+    expect(run).toHaveBeenCalledTimes(3); expect(c.nextWakeAt()).toBeNull();
+  });
+  it("coalesces tenant write bursts without repeatedly scanning global queues", async () => {
+    vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner);
+    const run = vi.fn(async () => {}), next = vi.fn(async () => null);
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, run, next }]).ready;
+    for (let i = 0; i < 100; i++) {
+      await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus, `company-${i % 2}`));
+      await vi.advanceTimersByTimeAsync(10);
+    }
+    expect(run).toHaveBeenCalledTimes(1); expect(next).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(14_000);
+    expect(run).toHaveBeenCalledTimes(2); expect(next).toHaveBeenCalledTimes(2);
+    expect(c.nextWakeAt()).toBeNull();
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(run).toHaveBeenCalledTimes(2);
+  });
+  it.each([false, true])("does not start sweeps or inspect deadlines once server shutdown begins (notification=%s)", async notification => {
+    vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner); let stopping = false;
+    const run = vi.fn(async () => {}), next = vi.fn(async () => Date.now() + 1000);
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus, run, next }], () => !stopping).ready;
+    stopping = true;
+    if (notification) await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionStatus));
+    await vi.advanceTimersByTimeAsync(30_000);
+    expect(run).toHaveBeenCalledTimes(1); expect(next).toHaveBeenCalledTimes(1);
+    await c.stop(); expect(c.nextWakeAt()).toBeNull();
+  });
+  it("keeps other obligations responsive while one is waiting, and awaits shutdown", async () => {
+    vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner);
+    let release!: () => void; const gate = new Promise<void>(resolve => { release = resolve; });
+    const status = vi.fn(async () => {});
+    const started = registerExecutionControlWork(c, [
+      { queue: DELIVERY_QUEUES.executionReplacement, run: () => gate, next: async () => null },
+      { queue: DELIVERY_QUEUES.executionStatus, run: status, next: async () => null },
+    ]);
+    await vi.advanceTimersByTimeAsync(1); expect(status).toHaveBeenCalledTimes(1);
+    let stopped = false; const stopping = c.stop().then(() => { stopped = true; });
+    await vi.advanceTimersByTimeAsync(1); expect(stopped).toBe(false);
+    release(); await started.ready; await stopping; expect(stopped).toBe(true);
+  });
+});
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXECUTION_SLEEP_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+describe.skipIf(!support.supported)("execution-control durable work", () => {
+  let db: ReturnType<typeof createDb>, observed: typeof db;
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let calls: number, companyId: string, agentId: string, issueId: string;
+  const errors: unknown[] = [];
+  beforeAll(async () => {
+    if (external) await applyPendingMigrations(external);
+    else temporary = await startEmbeddedPostgresTestDatabase("execution-sleep-");
+    db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 });
+  }, 90000);
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+  beforeEach(async () => {
+    await db.execute(sql`TRUNCATE TABLE companies CASCADE`);
+    companyId = randomUUID(); agentId = randomUUID(); issueId = randomUUID(); calls = 0; errors.length = 0;
+    await db.insert(companies).values({ id: companyId, name: "Idle execution", issuePrefix: "EXE" });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "codex_local" });
+    await db.insert(issues).values({ id: issueId, companyId, title: "Execution", status: "in_progress", assigneeAgentId: agentId });
+    observed = new Proxy(db, { get(target, key) {
+      if (!["select", "update", "insert", "delete", "execute", "transaction"].includes(String(key))) return Reflect.get(target, key);
+      return (...args: unknown[]) => { calls++; return Reflect.apply(Reflect.get(target, key), target, args); };
+    } });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
+  });
+  async function settle() {
+    await vi.advanceTimersByTimeAsync(15_000);
+    await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 });
+    await vi.advanceTimersByTimeAsync(1);
+    await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 });
+    expect(errors).toEqual([]);
+  }
+  async function quiet(c: ReturnType<typeof coordinator>) {
+    await settle(); expect(c.nextWakeAt()).toBeNull(); const before = calls;
+    await vi.advanceTimersByTimeAsync(90_000); expect(calls).toBe(before);
+  }
+  async function runRow(status = "running") {
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status, runtimeMode: "legacy", contextSnapshot: { issueId } }).returning();
+    return run!;
+  }
+  it("stops all six pending/deadline scans with empty queues and completed history", async () => {
+    await runRow("succeeded");
+    const schedule = executionControlSchedule(observed);
+    const c = coordinator(db, undefined, error => { errors.push(error); });
+    const topics = [DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionContinuation,
+      DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.localLoginCleanup];
+    const runs = [() => reconcileAbandonedExecutionControl(observed), () => reconcileSafeNativeReplacements(observed),
+      () => deliverReconciledExecutions(observed, vi.fn(async () => null)), () => deliverExecutionStatuses(observed),
+      () => settleUnrecoverableExecutions(observed), () => localAiLoginService(observed).reapExpired()];
+    await registerExecutionControlWork(c, Object.values(schedule).map((next, i) => ({ queue: topics[i]!, run: runs[i]!, next }))).ready;
+    await quiet(c);
+  });
+  it("bounds real producer bursts with another company's completed history present", async () => {
+    const otherCompany = randomUUID(), otherAgent = randomUUID();
+    await db.insert(companies).values({ id: otherCompany, name: "Other tenant", issuePrefix: "OTH" });
+    await db.insert(agents).values({ id: otherAgent, companyId: otherCompany, name: "Other worker", adapterType: "codex_local" });
+    await db.insert(heartbeatRuns).values({ companyId: otherCompany, agentId: otherAgent, status: "succeeded" });
+    const run = await runRow(); const publish = vi.fn(); const c = coordinator(db, undefined, error => { errors.push(error); });
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus,
+      run: () => deliverExecutionStatuses(observed, { publish }), next: executionControlSchedule(observed).status }]).ready;
+    const before = calls;
+    for (let i = 0; i < 20; i++) {
+      await terminalizeLegacyExecution({ db, run, status: "succeeded", fromStatuses: ["running", "succeeded"] });
+      await vi.advanceTimersByTimeAsync(10);
+    }
+    expect(calls).toBe(before); expect(publish).not.toHaveBeenCalled();
+    await settle(); expect(publish).toHaveBeenCalledTimes(1);
+    expect(publish.mock.calls[0]?.[0]).toMatchObject({ companyId, payload: { runId: run.id } });
+    await quiet(c);
+  });
+  it("wakes status delivery through the real producer only after outer commit, then becomes quiet", async () => {
+    const run = await runRow(); const publish = vi.fn(); const c = coordinator(db, undefined, error => { errors.push(error); });
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus,
+      run: () => deliverExecutionStatuses(observed, { publish }), next: executionControlSchedule(observed).status }]).ready;
+    await quiet(c);
+    await db.transaction(async tx => {
+      await terminalizeLegacyExecution({ db: tx as unknown as typeof db, run, status: "succeeded", patch: { finishedAt: new Date() } });
+      await vi.advanceTimersByTimeAsync(1); expect(publish).not.toHaveBeenCalled();
+    });
+    await settle(); expect(publish).toHaveBeenCalledTimes(1);
+    await quiet(c);
+    const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
+    expect(saved?.status).toBe("succeeded"); expect(saved?.executionStatusDeliveryId).toBeNull();
+  });
+  it("rolls back a nested producer without delivering or leaving periodic work", async () => {
+    const run = await runRow(), publish = vi.fn(); const c = coordinator(db, undefined, error => { errors.push(error); });
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionStatus,
+      run: () => deliverExecutionStatuses(observed, { publish }), next: executionControlSchedule(observed).status }]).ready;
+    await expect(db.transaction(async tx => {
+      await terminalizeLegacyExecution({ db: tx as unknown as typeof db, run, status: "succeeded" });
+      throw new Error("rollback");
+    })).rejects.toThrow("rollback");
+    await quiet(c); expect(publish).not.toHaveBeenCalled();
+  });
+  it("restores an exact control deadline on restart without querying before it", async () => {
+    const run = await runRow("succeeded"), due = new Date(Date.now() + 60_000);
+    await db.update(heartbeatRuns).set({ executionControlDeadlineAt: due }).where(eq(heartbeatRuns.id, run.id));
+    const c = coordinator(db, undefined, error => { errors.push(error); });
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionFinalization,
+      run: () => reconcileAbandonedExecutionControl(observed), next: executionControlSchedule(observed).finalization }]).ready;
+    expect(c.nextWakeAt()).toBe(due.getTime()); const before = calls;
+    await vi.advanceTimersByTimeAsync(55_000); expect(calls).toBe(before);
+    await vi.advanceTimersByTimeAsync(5000); await settle(); await quiet(c);
+  });
+  it("retains replacement retries for an unresolved failed execution, excludes completed/denied history", async () => {
+    const run = await runRow("failed");
+    await db.update(heartbeatRuns).set({ nativeIssueId: issueId }).where(eq(heartbeatRuns.id, run.id));
+    await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId, issueId, phase: "terminal_failure", failureCode: "provider_transport_failed" });
+    const schedule = executionControlSchedule(observed);
+    expect(await schedule.replacement()).toBeGreaterThan(Date.now());
+    await db.update(nativeRunFinalizations).set({ failureDetail: { replacementDenied: "operator_reconciled" } }).where(eq(nativeRunFinalizations.runId, run.id));
+    expect(await schedule.replacement()).toBeNull();
+  });
+  it("a recovery-action producer wakes automatic disposition and leaves resolved history quiet", async () => {
+    const run = await runRow("failed"); const c = coordinator(db, undefined, error => { errors.push(error); });
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionDisposition,
+      run: () => settleUnrecoverableExecutions(observed), next: executionControlSchedule(observed).disposition }]).ready;
+    await quiet(c);
+    const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: issueId,
+      kind: "active_run_watchdog", cause: "execution_finalization_deadline_exceeded", fingerprint: run.id,
+      returnOwnerAgentId: agentId, evidence: { runId: run.id }, nextAction: "Recover execution" });
+    await settle();
+    expect(await issueRecoveryActionService(db).getActiveForIssue(companyId, issueId)).toBeNull();
+    expect(action.id).toBeTruthy(); await quiet(c);
+  });
+  it("retries a failed broadcast only while its durable marker remains, including after restart", async () => {
+    const run = await runRow();
+    const publish = vi.fn(() => { throw new Error("receiver unavailable"); });
+    let c = coordinator(db, undefined, error => { errors.push(error); });
+    const duty = { queue: DELIVERY_QUEUES.executionStatus,
+      run: () => deliverExecutionStatuses(observed, { publish }), next: executionControlSchedule(observed).status };
+    await registerExecutionControlWork(c, [duty]).ready;
+    await terminalizeLegacyExecution({ db, run, status: "succeeded" });
+    await settle(); expect(publish).toHaveBeenCalledTimes(1);
+    const before = calls;
+    await vi.advanceTimersByTimeAsync(10_000); expect(calls).toBe(before);
+    await c.stop();
+    publish.mockImplementation(() => undefined);
+    c = coordinator(db, undefined, error => { errors.push(error); });
+    await registerExecutionControlWork(c, [duty]).ready;
+    expect(publish).toHaveBeenCalledTimes(2); await quiet(c);
+  });
+  it("reschedules an earlier deadline and disarms a cancelled deadline", async () => {
+    const run = await runRow("succeeded"); const c = coordinator(db, undefined, error => { errors.push(error); });
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionFinalization,
+      run: () => reconcileAbandonedExecutionControl(observed), next: executionControlSchedule(observed).finalization }]).ready;
+    const due = new Date(Date.now() + 120_000), earlier = new Date(Date.now() + 60_000);
+    for (const at of [due, earlier]) {
+      await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns)
+        .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id)));
+      await settle(); expect(c.nextWakeAt()).toBe(at.getTime());
+    }
+    await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns)
+      .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id)));
+    await quiet(c);
+  });
+  it("delivers a reconciled continuation when marking and resolving commit together", async () => {
+    const run = await runRow("failed"), successor = await runRow("queued");
+    const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: issueId,
+      kind: "active_run_watchdog", cause: "execution_finalization_deadline_exceeded", fingerprint: run.id,
+      returnOwnerAgentId: agentId, evidence: { runId: run.id }, nextAction: "Recover execution" });
+    await db.update(heartbeatRuns).set({ contextSnapshot: { issueId, recoveryActionId: action.id, previousRunId: run.id } })
+      .where(eq(heartbeatRuns.id, successor.id));
+    const wake = vi.fn(async () => successor), c = coordinator(db, undefined, error => { errors.push(error); });
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionContinuation,
+      run: () => deliverReconciledExecutions(observed, wake), next: executionControlSchedule(observed).continuation }]).ready;
+    await db.transaction(async tx => {
+      await markExecutionReconciliation(tx as unknown as typeof db, action,
+        { runId: run.id, providerStopped: true, actionOutcome: "not_performed", outcomeEvidence: "Verified the provider stopped without performing an external action." }, "operator");
+      await issueRecoveryActionService(tx).resolveActiveForIssue({ companyId, sourceIssueId: issueId, actionId: action.id, status: "resolved", outcome: "restored" });
+      await vi.advanceTimersByTimeAsync(1); expect(wake).not.toHaveBeenCalled();
+    });
+    await settle(); expect(wake).toHaveBeenCalledTimes(1); await quiet(c);
+    const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successor.id));
+    expect(saved?.retryOfRunId).toBe(run.id);
+  });
+
+});
diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts
index b4faf3b74d1..d0d98e6a52c 100644
--- a/server/src/__tests__/heartbeat-process-recovery.test.ts
+++ b/server/src/__tests__/heartbeat-process-recovery.test.ts
@@ -8297,19 +8297,28 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
       const lease = originalWatch(...args);
       return { ...lease, stop() { stopped(); lease.stop(); } };
     });
-    const originalUpdate = db.update.bind(db);
+    const originalTransaction = db.transaction.bind(db);
     const cleanupFailure = vi.fn(() => { throw new Error("fixture_deadline_cleanup_failure"); });
-    const update = vi.spyOn(db, "update").mockImplementation(((table: typeof heartbeatRuns) => {
-      const builder = originalUpdate(table);
-      const originalSet = builder.set.bind(builder);
-      builder.set = ((values: Record<string, unknown>) => {
-        if (table === heartbeatRuns && Object.keys(values).length === 1 && values.executionControlDeadlineAt === null) {
-          return { where: async () => cleanupFailure() };
+    const transaction = vi.spyOn(db, "transaction").mockImplementation((operation, config) =>
+      originalTransaction(async tx => {
+        const originalUpdate = tx.update.bind(tx);
+        const update = vi.spyOn(tx, "update").mockImplementation(((table: typeof heartbeatRuns) => {
+          const builder = originalUpdate(table);
+          const originalSet = builder.set.bind(builder);
+          builder.set = ((values: Record<string, unknown>) => {
+            if (table === heartbeatRuns && Object.keys(values).length === 1 && values.executionControlDeadlineAt === null) {
+              return { where: async () => cleanupFailure() };
+            }
+            return originalSet(values);
+          }) as typeof builder.set;
+          return builder;
+        }) as typeof tx.update);
+        try {
+          return await operation(tx);
+        } finally {
+          update.mockRestore();
         }
-        return originalSet(values);
-      }) as typeof builder.set;
-      return builder;
-    }) as typeof db.update);
+      }, config));
     const heartbeat = heartbeatService(db);
     try {
       await heartbeat.resumeQueuedRuns();
@@ -8319,7 +8328,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
       expect(adapterExecutionControls.has(runId)).toBe(false);
       expect((await heartbeat.getRun(runId))?.status).not.toBe("running");
     } finally {
-      update.mockRestore();
+      transaction.mockRestore();
       watcher.mockRestore();
     }
   });
diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts
index 0246d509ff3..e1151c9c899 100644
--- a/server/src/__tests__/issue-recovery-actions.test.ts
+++ b/server/src/__tests__/issue-recovery-actions.test.ts
@@ -106,6 +106,7 @@ describe("issueRecoveryActionService", () => {
     });
 
     const fakeDb = {
+      async transaction<T>(fn: (tx: unknown) => Promise<T>): Promise<T> { return fn(fakeDb); },
       select: vi.fn(() => makeSelectQuery(selectResults.shift() ?? [])),
       update: vi.fn(() => ({
         set: vi.fn(() => ({
diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts
index 96d94ef248a..db22ce7b2c2 100644
--- a/server/src/__tests__/server-startup-feedback-export.test.ts
+++ b/server/src/__tests__/server-startup-feedback-export.test.ts
@@ -790,16 +790,9 @@ describe("startServer feedback export wiring", () => {
     }
   });
 
-  it("reconciles native replacements at startup and on the execution-control interval", async () => {
+  it("registers native replacement recovery without an execution-control interval", async () => {
     loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true }));
-    let executionControlTick: (() => void) | undefined;
-    const setIntervalSpy = vi.spyOn(globalThis, "setInterval").mockImplementation(((
-      callback: () => void,
-      interval: number,
-    ) => {
-      if (interval === EXECUTION_RECONCILIATION_INTERVAL_MS) executionControlTick = callback;
-      return 1 as unknown as ReturnType<typeof setInterval>;
-    }) as typeof setInterval);
+    const setIntervalSpy = vi.spyOn(globalThis, "setInterval");
     try {
       await startServer();
       await new Promise<void>((resolve) => setImmediate(resolve));
@@ -808,11 +801,8 @@ describe("startServer feedback export wiring", () => {
         expect.any(Date),
         { verifyStoppedSession: expect.any(Function) },
       );
-
-      expect(executionControlTick).toBeDefined();
-      executionControlTick?.();
-      await new Promise<void>((resolve) => setImmediate(resolve));
-      expect(reconcileSafeNativeReplacements).toHaveBeenCalledTimes(2);
+      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.executionReplacement)).toBe(true);
+      expect(setIntervalSpy.mock.calls.some(([, interval]) => interval === EXECUTION_RECONCILIATION_INTERVAL_MS)).toBe(false);
     } finally {
       setIntervalSpy.mockRestore();
     }
diff --git a/server/src/index.ts b/server/src/index.ts
index 1645b0d7bcb..a7b0d17fabe 100644
--- a/server/src/index.ts
+++ b/server/src/index.ts
@@ -39,7 +39,7 @@ import { deliverExecutionStatuses } from "./services/execution-status-delivery.j
 import { deliverReconciledExecutions, settleUnrecoverableExecutions } from "./services/execution-recovery-resolution.js";
 import { reconcileSafeNativeReplacements } from "./services/native-runtime/native-safe-replacement.js";
 import { reconcileAbandonedExecutionControl } from "./services/execution-control-reconciliation.js";
-import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./services/execution-control-deadline.js";
+import { executionControlSchedule, registerExecutionControlWork } from "./services/execution-control-work.js";
 import { connectionIntentDeliveryService } from "./services/connection-intent-delivery.js";
 import { existsSync, readFileSync, rmSync } from "node:fs";
 import type { Server, RequestListener } from "node:http";
@@ -1205,30 +1205,6 @@ async function startServerWithDatabaseTeardown(
       await Promise.allSettled([...heartbeatSchedulerInFlight]);
     }
   };
-  const executionControlSweepsInFlight = new Set<string>();
-  const executionControlSweeps = [
-    ["finalization", () => reconcileAbandonedExecutionControl(db)],
-    ["replacement", () => heartbeat ? reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }) : undefined],
-    ["reconciliation_delivery", () => heartbeat ? deliverReconciledExecutions(db, heartbeat.wakeup) : undefined],
-    ["status_delivery", () => deliverExecutionStatuses(db)],
-    ["automatic_disposition", () => settleUnrecoverableExecutions(db)],
-    ["local_ai_login_cleanup", () => localAiLoginService(db).reapExpired()],
-  ] as const;
-  const sweepExecutionControl = () => {
-    if (isIdleTaskDrainActive() || heartbeatSchedulerStopped || isWarmStandby()) return;
-    // Independent durable queues must not block one another. Each queue remains
-    // single-flight; a later sweep observes committed transitions from its peers.
-    for (const [queue, work] of executionControlSweeps) {
-      if (executionControlSweepsInFlight.has(queue)) continue;
-      executionControlSweepsInFlight.add(queue);
-      trackHeartbeatSchedulerWork(Promise.resolve().then(async () => { await work(); })
-        .catch(err => logger.error({ err, queue }, "execution control reconciliation failed"))
-        .finally(() => { executionControlSweepsInFlight.delete(queue); }));
-    }
-  };
-  const executionControlInterval = setInterval(sweepExecutionControl, EXECUTION_RECONCILIATION_INTERVAL_MS);
-  executionControlInterval.unref?.();
-  sweepExecutionControl();
   const startHeartbeatSchedulerInterval = (callback: () => void) => {
     heartbeatSchedulerInterval = setInterval(() => {
       if (!isWarmStandby() && !isIdleTaskDrainActive()) callback();
@@ -1346,6 +1322,18 @@ async function startServerWithDatabaseTeardown(
     resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction),
   });
   const deliveryWork = app.locals.deliveryWork as ReturnType<typeof createDeliveryWorkCoordinator>;
+  const executionSchedule = executionControlSchedule(db);
+  registerExecutionControlWork(deliveryWork, [
+    { queue: DELIVERY_QUEUES.executionFinalization, run: () => reconcileAbandonedExecutionControl(db), next: executionSchedule.finalization },
+    { queue: DELIVERY_QUEUES.executionStatus, run: () => deliverExecutionStatuses(db), next: executionSchedule.status },
+    { queue: DELIVERY_QUEUES.executionDisposition, run: () => settleUnrecoverableExecutions(db), next: executionSchedule.disposition },
+    { queue: DELIVERY_QUEUES.localLoginCleanup, run: () => localAiLoginService(db).reapExpired(), next: executionSchedule.localLogin },
+    ...(heartbeat ? [
+      { queue: DELIVERY_QUEUES.executionReplacement, run: () => reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }), next: executionSchedule.replacement },
+      { queue: DELIVERY_QUEUES.executionContinuation, run: () => deliverReconciledExecutions(db, heartbeat.wakeup), next: executionSchedule.continuation },
+    ] : []),
+  ], () => !heartbeatSchedulerStopped);
+
   const budgetEnforcementWorker = registerBudgetEnforcementWork(db, deliveryWork, {
     cancelWorkForScope: accountingHeartbeat.cancelBudgetScopeWork,
   });
@@ -2092,7 +2080,6 @@ async function startServerWithDatabaseTeardown(
     unsubscribeChatCompletions();
     await deliveryWork.stop();
     unregisterTerminalWorkspaceProbe?.();
-    clearInterval(executionControlInterval);
     if (heartbeatSchedulerInterval) {
       clearInterval(heartbeatSchedulerInterval);
       heartbeatSchedulerInterval = null;
diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts
index b1c7c62062a..26f737b39cd 100644
--- a/server/src/services/delivery-work-notifications.ts
+++ b/server/src/services/delivery-work-notifications.ts
@@ -2,6 +2,12 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  executionFinalization: "execution-finalization",
+  executionReplacement: "execution-replacement",
+  executionContinuation: "execution-continuation",
+  executionStatus: "execution-status",
+  executionDisposition: "execution-disposition",
+  localLoginCleanup: "local-login-cleanup",
   budgetEnforcement: "budget-enforcement",
   accounting: "cost-accounting",
   agentLifecycle: "agent-lifecycle",
diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts
new file mode 100644
index 00000000000..fb59deafbe8
--- /dev/null
+++ b/server/src/services/execution-control-notifications.ts
@@ -0,0 +1,20 @@
+import type { Db } from "@paperclipai/db";
+import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js";
+
+export async function notifyExecutionRunWork(transaction: object) {
+  for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
+    DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition]) {
+    await notifyDeliveryWork(transaction, queue);
+  }
+}
+
+export function writeExecutionControl<T>(
+  db: Db | Parameters<Parameters<Db["transaction"]>[0]>[0],
+  queue: DeliveryQueue | readonly DeliveryQueue[],
+  write: (tx: Db) => PromiseLike<T>,
+): Promise<T> {
+  return db.transaction(async transaction => {
+    for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
+    return write(transaction as unknown as Db);
+  });
+}
diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts
index 72266731dff..5b3a7aa3735 100644
--- a/server/src/services/execution-control-reconciliation.ts
+++ b/server/src/services/execution-control-reconciliation.ts
@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "./execution-control-notifications.js";
 import { randomUUID } from "node:crypto";
 import { logger } from "../middleware/logger.js";
 import { and, eq, isNotNull, lte, sql } from "drizzle-orm";
@@ -181,6 +182,7 @@ export async function reconcileAbandonedExecutionControl(
                 updatedAt: now,
               })
               .where(eq(nativeRunFinalizations.runId, run.id));
+          await notifyExecutionRunWork(tx);
           const [updatedRun] = await tx
             .update(heartbeatRuns)
             .set({
diff --git a/server/src/services/execution-control-work.ts b/server/src/services/execution-control-work.ts
new file mode 100644
index 00000000000..1c7a1a6d376
--- /dev/null
+++ b/server/src/services/execution-control-work.ts
@@ -0,0 +1,98 @@
+import { and, asc, eq, isNotNull, sql } from "drizzle-orm";
+import { adapterAuthSessions, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db";
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { type DeliveryQueue } from "./delivery-work-notifications.js";
+import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./execution-control-deadline.js";
+import { obsoleteExecutionConversationHold, unresolvedExecutionDispositions } from "./execution-recovery-resolution.js";
+import { nativeReplacementPending } from "./native-runtime/native-safe-replacement.js";
+
+export function registerExecutionControlWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  duties: ReadonlyArray<{
+    queue: DeliveryQueue;
+    run: () => Promise<unknown>;
+    next: () => Promise<number | null>;
+  }>,
+  canRun: () => boolean = () => true,
+) {
+  const workers = duties.map(duty => {
+    let requested = 1;
+    let reconciled = 0;
+    let next: number | null = null;
+    let earliestRunAt = 0;
+    return coordinator.register(duty.queue, {
+      retryMs: EXECUTION_RECONCILIATION_INTERVAL_MS,
+      pausePassiveMaintenance: true,
+      onSettled: () => { requested += 1; },
+      run: async () => {
+        if (!canRun()) { requested += 1; next = null; return; }
+        const now = Date.now();
+        if (now < earliestRunAt) {
+          if (reconciled !== requested || (next !== null && next <= now)) next = earliestRunAt;
+          return;
+        }
+        const generation = ++requested;
+        earliestRunAt = now + EXECUTION_RECONCILIATION_INTERVAL_MS;
+        await duty.run();
+        const deadline = await duty.next();
+        next = deadline === null ? null : Math.max(deadline, earliestRunAt);
+        reconciled = generation;
+      },
+      hasPending: async () => reconciled !== requested,
+      nextRunAt: () => next,
+    });
+  });
+  return { ready: Promise.all(workers.map(worker => worker.ready)) };
+}
+
+export function executionControlSchedule(db: Db) {
+  const retry = () => Date.now() + EXECUTION_RECONCILIATION_INTERVAL_MS;
+  const deadline = (dates: Array<Date | null | undefined>) => {
+    const at = dates.flatMap(date => date ? [date.getTime()] : []);
+    if (!at.length) return null;
+    const earliest = Math.min(...at);
+    return earliest > Date.now() ? earliest : retry();
+  };
+  return {
+    async finalization() {
+      const [native, legacy] = await Promise.all([
+        db.select({ at: nativeRunFinalizations.controlDeadlineAt }).from(nativeRunFinalizations)
+          .where(isNotNull(nativeRunFinalizations.controlDeadlineAt)).orderBy(asc(nativeRunFinalizations.controlDeadlineAt)).limit(1),
+        db.select({ at: heartbeatRuns.executionControlDeadlineAt }).from(heartbeatRuns)
+          .where(isNotNull(heartbeatRuns.executionControlDeadlineAt)).orderBy(asc(heartbeatRuns.executionControlDeadlineAt)).limit(1),
+      ]);
+      return deadline([native[0]?.at, legacy[0]?.at]);
+    },
+    async replacement() {
+      const rows = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
+        .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
+        .where(nativeReplacementPending()).limit(1);
+      return rows.length ? retry() : null;
+    },
+    async continuation() {
+      const rows = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(
+        eq(issueRecoveryActions.status, "resolved"),
+        sql`${issueRecoveryActions.evidence}->>'continuationDelivery' = 'pending'`,
+      )).limit(1);
+      return rows.length ? retry() : null;
+    },
+    async status() {
+      const rows = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
+        .where(isNotNull(heartbeatRuns.executionStatusDeliveryId)).limit(1);
+      return rows.length ? retry() : null;
+    },
+    async disposition() {
+      const [conversation, execution] = await Promise.all([
+        db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(obsoleteExecutionConversationHold()).limit(1),
+        unresolvedExecutionDispositions(db, 1),
+      ]);
+      return conversation.length || execution.length ? retry() : null;
+    },
+    async localLogin() {
+      const rows = await db.select({ at: adapterAuthSessions.expiresAt }).from(adapterAuthSessions).where(and(
+        eq(adapterAuthSessions.connectionMethod, "local_subscription"), isNotNull(adapterAuthSessions.expiresAt),
+      )).orderBy(asc(adapterAuthSessions.expiresAt)).limit(1);
+      return deadline([rows[0]?.at]);
+    },
+  };
+}
diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts
index 3323bceb93e..e56141565b9 100644
--- a/server/src/services/execution-recovery-resolution.ts
+++ b/server/src/services/execution-recovery-resolution.ts
@@ -1,3 +1,5 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+import { notifyExecutionRunWork } from "./execution-control-notifications.js";
 import { hasWorkspaceRestoreFailure } from "@paperclipai/shared";
 import { randomUUID } from "node:crypto";
 import { canonicalExecutionUuid } from "./execution-identity.js";
@@ -201,38 +203,43 @@ export async function markExecutionReconciliation(
       throw conflict("The authorized chat retry owner is no longer valid.");
     }
   }
-  await db
-    .update(nativeRunFinalizations)
-    .set({
-      failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "operator_reconciled" })}::jsonb`,
-    })
-    .where(
-      and(
-        eq(nativeRunFinalizations.companyId, action.companyId),
-        eq(nativeRunFinalizations.runId, decision.runId),
-      ),
-    );
-  await db
-    .update(issueRecoveryActions)
-    .set({
-      evidence: {
-        ...action.evidence,
-        automaticRecovery: undefined,
-        executionReconciliation: {
-          ...decision,
-          actorId,
-          recordedAt: new Date().toISOString(),
+  await db.transaction(async tx => {
+    await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation);
+    await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement);
+    await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition);
+    await tx
+      .update(nativeRunFinalizations)
+      .set({
+        failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "operator_reconciled" })}::jsonb`,
+      })
+      .where(
+        and(
+          eq(nativeRunFinalizations.companyId, action.companyId),
+          eq(nativeRunFinalizations.runId, decision.runId),
+        ),
+      );
+    await tx
+      .update(issueRecoveryActions)
+      .set({
+        evidence: {
+          ...action.evidence,
+          automaticRecovery: undefined,
+          executionReconciliation: {
+            ...decision,
+            actorId,
+            recordedAt: new Date().toISOString(),
+          },
+          continuationDelivery: options?.workspaceRepairOnly ? "not_requested" : deliveryOwner ? "delegated" : "pending",
+          ...(deliveryOwner ? { continuationDeliveryOwner: deliveryOwner } : {}),
         },
-        continuationDelivery: options?.workspaceRepairOnly ? "not_requested" : deliveryOwner ? "delegated" : "pending",
-        ...(deliveryOwner ? { continuationDeliveryOwner: deliveryOwner } : {}),
-      },
-    })
-    .where(
-      and(
-        eq(issueRecoveryActions.companyId, action.companyId),
-        eq(issueRecoveryActions.id, action.id),
-      ),
-    );
+      })
+      .where(
+        and(
+          eq(issueRecoveryActions.companyId, action.companyId),
+          eq(issueRecoveryActions.id, action.id),
+        ),
+      );
+  });
 }
 
 export async function deliverReconciledExecutions(
@@ -337,60 +344,8 @@ export async function deliverReconciledExecutions(
   }
 }
 
-/**
- * Failed execution is a system responsibility, not a user questionnaire. After
- * automatic recovery is ruled out, preserve evidence and stop without replay.
- * This is NOT evidence that an external action succeeded or never happened.
- * The resolved record retains a dispatch hold until actual evidence clears it.
- */
-export async function settleUnrecoverableExecutions(
-  db: Db,
-  now = new Date(),
-  options: { failpoint?: (phase: "persisted") => void } = {},
-) {
-  // Fold obsolete conversation holds without waking historical work on upgrade.
-  // Keep their evidence and record the policy change in the task's activity log.
-  const obsoleteConversationHold = and(
-    conversationRecoveryActionPredicate(),
-    sql`not (${issueRecoveryActions.evidence} ? 'systemRecovery')`,
-    or(
-      inArray(issueRecoveryActions.status, ["active", "escalated"]),
-      sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`,
-    ),
-  );
-  await db.transaction(async tx => {
-    const foldable = await tx.select().from(issueRecoveryActions).where(obsoleteConversationHold)
-      .limit(25).for("update", { skipLocked: true });
-    for (const candidate of foldable) {
-      if (await getConversationOwnershipBlocker(tx as unknown as Db, candidate.companyId, candidate.sourceIssueId)) continue;
-      const [action] = await tx.update(issueRecoveryActions).set({
-        status: "resolved",
-        outcome: "cancelled",
-        resolvedAt: now,
-        updatedAt: now,
-        nextAction: "Automatic attempts stopped. Send a new message to continue the conversation.",
-        resolutionNote: "Conversation continuation does not replay prior tool calls.",
-        wakePolicy: null,
-        monitorPolicy: null,
-        evidence: sql`case when ${issueRecoveryActions.evidence} ? 'automaticRecovery'
-          then jsonb_set(${issueRecoveryActions.evidence}, '{automaticRecovery,replay}', '"conversation_continuation"'::jsonb)
-          else ${issueRecoveryActions.evidence} end`,
-      }).where(and(obsoleteConversationHold, eq(issueRecoveryActions.id, candidate.id))).returning();
-      if (!action) continue;
-      await persistActivity(tx as unknown as Db, {
-        companyId: action.companyId,
-        actorType: "system",
-        actorId: "execution-recovery",
-        action: "issue.execution_recovery_settled",
-        entityType: "issue",
-        entityId: action.sourceIssueId,
-        details: { recoveryActionId: action.id, outcome: "cancelled", continuation: "conversation" },
-      });
-    }
-  });
-  // Filter eligibility before applying the batch limit. A queue of sessions
-  // awaiting replacement must not starve settled incidents behind it.
-  const candidates = await db
+export function unresolvedExecutionDispositions(db: Db, limit: number) {
+  return db
     .select({ action: issueRecoveryActions })
     .from(issueRecoveryActions)
     .innerJoin(
@@ -437,7 +392,67 @@ export async function settleUnrecoverableExecutions(
         or coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') <> '')`,
       ),
     )
-    .limit(25);
+    .limit(limit);
+}
+
+export function obsoleteExecutionConversationHold() {
+  return and(
+    conversationRecoveryActionPredicate(),
+    sql`not (${issueRecoveryActions.evidence} ? 'systemRecovery')`,
+    or(
+      inArray(issueRecoveryActions.status, ["active", "escalated"]),
+      sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`,
+    ),
+  );
+}
+
+/**
+ * Failed execution is a system responsibility, not a user questionnaire. After
+ * automatic recovery is ruled out, preserve evidence and stop without replay.
+ * This is NOT evidence that an external action succeeded or never happened.
+ * The resolved record retains a dispatch hold until actual evidence clears it.
+ */
+export async function settleUnrecoverableExecutions(
+  db: Db,
+  now = new Date(),
+  options: { failpoint?: (phase: "persisted") => void } = {},
+) {
+  // Fold obsolete conversation holds without waking historical work on upgrade.
+  // Keep their evidence and record the policy change in the task's activity log.
+  const obsoleteConversationHold = obsoleteExecutionConversationHold();
+  await db.transaction(async tx => {
+    const foldable = await tx.select().from(issueRecoveryActions).where(obsoleteConversationHold)
+      .limit(25).for("update", { skipLocked: true });
+    for (const candidate of foldable) {
+      if (await getConversationOwnershipBlocker(tx as unknown as Db, candidate.companyId, candidate.sourceIssueId)) continue;
+      const [action] = await tx.update(issueRecoveryActions).set({
+        status: "resolved",
+        outcome: "cancelled",
+        resolvedAt: now,
+        updatedAt: now,
+        nextAction: "Automatic attempts stopped. Send a new message to continue the conversation.",
+        resolutionNote: "Conversation continuation does not replay prior tool calls.",
+        wakePolicy: null,
+        monitorPolicy: null,
+        evidence: sql`case when ${issueRecoveryActions.evidence} ? 'automaticRecovery'
+          then jsonb_set(${issueRecoveryActions.evidence}, '{automaticRecovery,replay}', '"conversation_continuation"'::jsonb)
+          else ${issueRecoveryActions.evidence} end`,
+      }).where(and(obsoleteConversationHold, eq(issueRecoveryActions.id, candidate.id))).returning();
+      if (!action) continue;
+      await persistActivity(tx as unknown as Db, {
+        companyId: action.companyId,
+        actorType: "system",
+        actorId: "execution-recovery",
+        action: "issue.execution_recovery_settled",
+        entityType: "issue",
+        entityId: action.sourceIssueId,
+        details: { recoveryActionId: action.id, outcome: "cancelled", continuation: "conversation" },
+      });
+    }
+  });
+  // Filter eligibility before applying the batch limit. A queue of sessions
+  // awaiting replacement must not starve settled incidents behind it.
+  const candidates = await unresolvedExecutionDispositions(db, 25);
   for (const { action: candidate } of candidates) {
     const runId = candidate.evidence.runId;
     if (typeof runId !== "string") continue;
@@ -586,6 +601,7 @@ export async function settleUnrecoverableExecutions(
             replay: "not_authorized",
           },
         });
+        await notifyExecutionRunWork(tx);
         await tx
           .update(heartbeatRuns)
           .set({ executionStatusDeliveryId: randomUUID() })
diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts
index c449167dd00..a33d421a24f 100644
--- a/server/src/services/explicit-native-continuation.ts
+++ b/server/src/services/explicit-native-continuation.ts
@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "./execution-control-notifications.js";
 import { canSelectFreshWorkspace, type ContinuationReadiness } from "./fresh-workspace-continuation.js";
 import { readTaskRetryIntent } from "./task-retry.js";
 import { readSystemRecovery } from "./system-recovery-state.js";
@@ -356,16 +357,19 @@ export async function admitExplicitNativeContinuation(input: {
     ...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}),
     ...(queuedRequest ? { queuedCommentRequestId: input.queuedCommentRequestId } : {}), runId: input.successorRunId,
     previousRunId: previous.id, recordedAt: new Date().toISOString() };
-  for (const runId of cancelledStartupIds) {
-    await db.update(nativeRunFinalizations).set({
-      phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null,
-      controlDeadlineAt: null, updatedAt: new Date(),
-    }).where(and(eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId)));
-    await db.update(heartbeatRuns).set({
-      ...(nativeSources.some(run => run.id === runId) ? { nativePhase: "terminal_failure", nativePhaseUpdatedAt: new Date() } : {}),
-      executionControlDeadlineAt: null,
-    }).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId)));
-  }
+  if (cancelledStartupIds.size) await db.transaction(async tx => {
+    await notifyExecutionRunWork(tx);
+    for (const runId of cancelledStartupIds) {
+      await tx.update(nativeRunFinalizations).set({
+        phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null,
+        controlDeadlineAt: null, updatedAt: new Date(),
+      }).where(and(eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId)));
+      await tx.update(heartbeatRuns).set({
+        ...(nativeSources.some(run => run.id === runId) ? { nativePhase: "terminal_failure", nativePhaseUpdatedAt: new Date() } : {}),
+        executionControlDeadlineAt: null,
+      }).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId)));
+    }
+  });
   if (nativeSources.length) await db.update(nativeRunFinalizations).set({
     failureDetail: sql`coalesce(${nativeRunFinalizations.failureDetail}, '{}'::jsonb) || ${JSON.stringify({ replacementDenied: "explicit_user_continuation" })}::jsonb`,
     updatedAt: new Date(),
diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts
index b571364d22d..537510275da 100644
--- a/server/src/services/heartbeat.ts
+++ b/server/src/services/heartbeat.ts
@@ -1,3 +1,5 @@
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import { writeExecutionControl } from "./execution-control-notifications.js";
 import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js";
 import { promptInjectionPreventionService } from "./prompt-injection-prevention.js";
 import { createHeartbeatAgentStatus } from "./heartbeat/agent-status.js";
@@ -3657,10 +3659,10 @@ export function heartbeatService(
       let latestRun = await getRun(run.id).catch(() => null);
       try {
         if (latestRun && isHeartbeatRunTerminalStatus(latestRun.status)) {
-          await db
+          await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx
             .update(heartbeatRuns)
             .set({ executionControlDeadlineAt: null })
-            .where(eq(heartbeatRuns.id, run.id));
+            .where(eq(heartbeatRuns.id, run.id)));
         }
         nativeOwnershipHeld =
           nativeOwnershipHeld ||
diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts
index fbebed7f69f..0ee088a4d51 100644
--- a/server/src/services/heartbeat/run-lifecycle.ts
+++ b/server/src/services/heartbeat/run-lifecycle.ts
@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import os from "node:os";
 import {
   DEFERRED_WAKE_CONTEXT_KEY,
@@ -1011,12 +1012,12 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
   }
 
   async function writeRunStatus<T>(status: string, companyId: string | undefined, write: (tx: Db) => Promise<T>) {
-    if (!isHeartbeatRunTerminalStatus(status)) return write(db);
     // Register on the outer transaction before the terminal write. An unknown
     // commit cannot strand newly-actionable accounting after the live event.
     return db.transaction(async transaction => {
       const tx = transaction as unknown as Db;
-      await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId);
+      if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId);
+      await notifyExecutionRunWork(tx);
       return write(tx);
     });
   }
diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts
index bba22d1442f..cc304a6afba 100644
--- a/server/src/services/heartbeat/runtime-execution.ts
+++ b/server/src/services/heartbeat/runtime-execution.ts
@@ -1,3 +1,5 @@
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import { writeExecutionControl } from "../execution-control-notifications.js";
 import { AsyncLocalStorage } from "node:async_hooks";
 import { measureRunPhase } from "../execution-preparation-timing.js";
 import { parseNativeSessionGoalControl } from "./scheduling.js";
@@ -933,7 +935,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
             })); });
         }
         await ownership?.assertHeld();
-        await db
+        await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx
           .update(heartbeatRuns)
           .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) })
           .where(
@@ -941,7 +943,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
               eq(heartbeatRuns.id, run.id),
               eq(heartbeatRuns.status, "running"),
             ),
-          );
+          ));
         const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded";
         await recordWorkspaceFinalize(workspaceFinalizeStatus);
         if (adapterResult.nativeFinalization) {
@@ -1009,7 +1011,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
       && !(nativeExecution && "access" in nativeExecution.workspace && nativeExecution.workspace.access === "none")) {
       input.effects.onProviderResourceDisposition("stop_and_retain");
     }
-    await db
+    await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx
       .update(heartbeatRuns)
       .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) })
       .where(
@@ -1017,7 +1019,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
           eq(heartbeatRuns.id, run.id),
           eq(heartbeatRuns.status, "running"),
         ),
-      );
+      ));
     if (
       issueRef &&
       context.resumeSessionGoalHeartbeat === true &&
diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts
index 5351722b198..04e2c19a6d1 100644
--- a/server/src/services/issue-recovery-actions.ts
+++ b/server/src/services/issue-recovery-actions.ts
@@ -1,3 +1,5 @@
+import { writeExecutionControl } from "./execution-control-notifications.js";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { and, desc, eq, inArray, isNull, isNotNull, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import { heartbeatRuns, issueRecoveryActions, workspaceOperations } from "@paperclipai/db";
@@ -318,6 +320,8 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
   ): Promise<IssueRecoveryAction> {
     try {
       const created = await db.transaction(async (tx) => {
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition);
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation);
         const [superseded] = await tx
           .update(issueRecoveryActions)
           .set({
@@ -381,10 +385,10 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
         if (existing.cause === "native_workspace_finalization_owner_unverified"
           && input.cause === existing.cause && input.fingerprint === existing.fingerprint
           && !existing.evidence?.workspaceOwnerStop && existing.nextAction !== input.nextAction) {
-          const [updated] = await db.update(issueRecoveryActions).set({ nextAction: input.nextAction, updatedAt: now })
+          const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx.update(issueRecoveryActions).set({ nextAction: input.nextAction, updatedAt: now })
             .where(and(eq(issueRecoveryActions.id, existing.id),
               inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]),
-              sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning();
+              sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning());
           if (updated) return toReadModel(updated);
         }
         return existing;
@@ -404,7 +408,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
           existing.attemptCount,
           Math.min(nextAttemptCount, effectiveMaxAttempts),
         );
-        const [exhausted] = await db
+        const [exhausted] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx
           .update(issueRecoveryActions)
           .set({
             status: "escalated",
@@ -452,13 +456,13 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
               ]),
             ),
           )
-          .returning();
+          .returning());
         if (!exhausted) {
           return retryUpsertSourceScoped(input, retryCount);
         }
         return toReadModel(exhausted);
       }
-      const [updated] = await db
+      const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx
         .update(issueRecoveryActions)
         .set({
           recoveryIssueId: input.preserveExistingOwner
@@ -517,7 +521,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
             inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]),
           ),
         )
-        .returning();
+        .returning());
       if (!updated) {
         return retryUpsertSourceScoped(input, retryCount);
       }
@@ -525,10 +529,10 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
     }
 
     try {
-      const [created] = await db
+      const [created] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx
         .insert(issueRecoveryActions)
         .values(buildInsertValues(input, ownerType, now))
-        .returning();
+        .returning());
       return toReadModel(created!);
     } catch (error) {
       if (!isUniqueRecoveryActionConflict(error)) throw error;
@@ -565,7 +569,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
       predicates.push(eq(issueRecoveryActions.fingerprint, input.fingerprint));
     }
 
-    const [updated] = await dbOrTx
+    const [updated] = await writeExecutionControl(dbOrTx, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx
       .update(issueRecoveryActions)
       .set({
         status: input.status,
@@ -575,7 +579,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
         updatedAt: now,
       })
       .where(and(...predicates))
-      .returning();
+      .returning());
 
     return updated ? toReadModel(updated) : null;
   }
diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts
index 54fe6184b35..f1b16bd0e49 100644
--- a/server/src/services/legacy-execution-recovery.ts
+++ b/server/src/services/legacy-execution-recovery.ts
@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "./execution-control-notifications.js";
 import { hasRequiredWorkspaceRecovery, preserveWorkspaceRestoreRecoveryMetadata, LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js";
 import { hasUnrestoredRemoteWorkspace, preserveLegacyWorkspaceRestoreSources, type LegacyWorkspaceRestoreSource } from "./legacy-workspace-restore-recovery.js";
 import { isPreDispatchReviewWaitVerified } from "./pre-dispatch-review-wait.js";
@@ -108,6 +109,7 @@ export async function terminalizeLegacyExecution(input: {
       resultJson: preserveWorkspaceRestoreRecoveryMetadata(current.resultJson,
         input.recordRestoreFailureOnly ? { ...current.resultJson, ...patch.resultJson } : patch.resultJson),
     };
+    await notifyExecutionRunWork(tx);
     let [updated] = await tx
       .update(heartbeatRuns)
       .set({
diff --git a/server/src/services/local-ai-login.ts b/server/src/services/local-ai-login.ts
index 71482abc4cb..2108a916baf 100644
--- a/server/src/services/local-ai-login.ts
+++ b/server/src/services/local-ai-login.ts
@@ -1,3 +1,4 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { randomUUID } from "node:crypto";
 import { realpathSync } from "node:fs";
 import { mkdir, rm, writeFile } from "node:fs/promises";
@@ -120,6 +121,7 @@ export function localAiLoginService(db: Db) {
       const expiresAt = new Date(Date.now() + ATTEMPT_DURATION_MS);
       await prepareHome(id, intent.provider);
       try {
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.localLoginCleanup);
         await tx.insert(adapterAuthSessions).values({
           id, publicSessionId: id, companyId, environmentId: environment.id,
           adapterType,
diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts
index 76abcdefd31..c8f31024920 100644
--- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts
+++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts
@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm";
 import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
 import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js";
@@ -55,6 +56,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: typeof hea
     let retained = !retainFiles || alreadyRetained ? run : await recordRequiredWorkspaceRestoreFailure(tx as unknown as Db, run, { workspaceRestoreFailure: "restore_failed" }, source);
     if (!retained || retainFiles && !hasRequiredWorkspaceRecovery(retained.resultJson)) return;
     if (options.allowPreterminal && coordinator.phase === "observed") {
+      await notifyExecutionRunWork(tx);
       const [settled] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure",
         leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, nextAttemptAt: null,
         recoveryState: "blocked", failureCode: coordinator.failureCode ?? "native_continuation_requires_reconciliation", updatedAt: new Date(),
diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts
index fb05f82252a..2e3229fe011 100644
--- a/server/src/services/native-runtime/native-restart-recovery.ts
+++ b/server/src/services/native-runtime/native-restart-recovery.ts
@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js";
 import { activeNativeExecutionRunIds, NATIVE_PREPARATION_FENCING_VERSION } from "./native-active-runs.js";
 import {
@@ -855,6 +856,7 @@ export async function claimNativeRestartRecoveries(input: {
           recycledProviderPids: providerProcesses.recycledPids,
           stderrTail: row.run.stderrExcerpt,
         });
+        await notifyExecutionRunWork(tx);
         await tx
           .update(nativeRunFinalizations)
           .set({
diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts
index cf17721d832..61f97aefb8b 100644
--- a/server/src/services/native-runtime/native-run-finalizer.ts
+++ b/server/src/services/native-runtime/native-run-finalizer.ts
@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { reconcileMuseNativeTerminalRun } from "./muse-terminal-reconciliation.js";
 import { assertNativeCaseHandoffResult } from "./native-case-handoff.js";
 import { scheduledIssueMonitorWait } from "../issue-monitors.js";
@@ -420,6 +421,7 @@ async function recordRetryableFailure(input: {
             : "native_finalization_retry_exhausted"
           : input.failureCode;
     if (phase === "terminal_failure") await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
+    await notifyExecutionRunWork(tx);
     await tx
       .update(nativeRunFinalizations)
       .set({
@@ -705,6 +707,7 @@ async function projectCommittedRun(input: {
   const [updatedRun] = await input.db.transaction(async transaction => {
     const tx = transaction as unknown as Db;
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
+    await notifyExecutionRunWork(tx);
     return tx
       .update(heartbeatRuns)
       .set({
@@ -1617,13 +1620,12 @@ export async function finalizeNativeRun(input: {
         })
         .where(eq(heartbeatRuns.id, run.id))
         .returning();
-      const [updatedRun] = input.projectRunStatus
-        ? await input.db.transaction(async transaction => {
-            const tx = transaction as unknown as Db;
-            await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
-            return projectRun(tx);
-          })
-        : await projectRun(input.db);
+      const [updatedRun] = await input.db.transaction(async transaction => {
+        const tx = transaction as unknown as Db;
+        if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
+        await notifyExecutionRunWork(tx);
+        return projectRun(tx);
+      });
       if (
         input.projectRunStatus &&
         !alreadyEmittedByCommittedDecision &&
diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts
index 929cc58c510..03109f56def 100644
--- a/server/src/services/native-runtime/native-safe-replacement.ts
+++ b/server/src/services/native-runtime/native-safe-replacement.ts
@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { logger } from "../../middleware/logger.js";
 import { createHash, randomUUID } from "node:crypto";
 import { stat } from "node:fs/promises";
@@ -34,6 +35,20 @@ const processAlive = (pid: number | null) => {
   }
 };
 
+export function nativeReplacementPending() {
+  return and(
+    eq(heartbeatRuns.status, "failed"),
+    eq(nativeRunFinalizations.phase, "terminal_failure"),
+    inArray(
+      nativeRunFinalizations.failureCode,
+      ["native_provider_terminal_failed", "native_session_cleanup_quarantined", "provider_transport_failed"],
+    ),
+    isNull(nativeRunFinalizations.resultId),
+    sql`coalesce(${nativeRunFinalizations.failureDetail}->>'successorRunId', '') = ''`,
+    sql`coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') = ''`,
+  );
+}
+
 /** Retryable outbox: terminal_failure plus a new-format failure code is the durable candidate. */
 export async function reconcileSafeNativeReplacements(
   db: Db,
@@ -55,17 +70,7 @@ export async function reconcileSafeNativeReplacements(
       eq(nativeRunFinalizations.runId, heartbeatRuns.id),
     )
     .where(
-      and(
-        eq(heartbeatRuns.status, "failed"),
-        eq(nativeRunFinalizations.phase, "terminal_failure"),
-        inArray(
-          nativeRunFinalizations.failureCode,
-          ["native_provider_terminal_failed", "native_session_cleanup_quarantined", "provider_transport_failed"],
-        ),
-        isNull(nativeRunFinalizations.resultId),
-        sql`coalesce(${nativeRunFinalizations.failureDetail}->>'successorRunId', '') = ''`,
-        sql`coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') = ''`,
-      ),
+      nativeReplacementPending(),
     )
     .limit(25);
   let scheduled = 0;
@@ -260,6 +265,7 @@ export async function reconcileSafeNativeReplacements(
             current.failureDetail?.replacementDenied
           )
             return;
+          await notifyExecutionRunWork(tx);
           await tx
             .update(nativeRunFinalizations)
             .set({
@@ -401,6 +407,7 @@ export async function reconcileSafeNativeReplacements(
             idempotencyKey: `native-safe-replacement:${run.id}`,
           })
           .returning();
+        await notifyExecutionRunWork(tx);
         await tx.insert(heartbeatRuns).values({
           id: successorRunId,
           companyId: run.companyId,
diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts
index 67423e01554..682186ee2f3 100644
--- a/server/src/services/native-runtime/native-session-executor.test.ts
+++ b/server/src/services/native-runtime/native-session-executor.test.ts
@@ -9460,13 +9460,24 @@ describe("native session bounded recovery", () => {
       );
       const transact = db.transaction.bind(db);
       const recoveryWriteAttempt = vi.fn();
-      db.transaction = (async (operation) => {
-        if (state.execute.mock.calls.length > 0) {
-          recoveryWriteAttempt();
-          if (ancillaryFailure === "recovery_write") throw secondaryFailure;
-        }
-        return transact(operation);
-      }) as typeof db.transaction;
+      db.transaction = (async (operation) => transact(async tx => operation(new Proxy(tx, {
+        get(target, key) {
+          if (key !== "update") return Reflect.get(target, key);
+          return (table: unknown) => {
+            const query = Reflect.apply(target.update, target, [table]);
+            return new Proxy(query, { get(update, property) {
+              if (property !== "set") return Reflect.get(update, property);
+              return (values: Record<string, unknown>) => {
+                if (table === nativeRunFinalizations && values.phase === "terminal_failure") {
+                  recoveryWriteAttempt();
+                  if (ancillaryFailure === "recovery_write") throw secondaryFailure;
+                }
+                return update.set(values);
+              };
+            } });
+          };
+        },
+      })))) as typeof db.transaction;
       const updateIssue = vi.fn(async () => null);
       const service = vi
         .spyOn(issueServiceModule, "issueService")
diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts
index 73ae25afb56..60d6e9a1d4a 100644
--- a/server/src/services/native-runtime/native-session-executor.ts
+++ b/server/src/services/native-runtime/native-session-executor.ts
@@ -1,3 +1,6 @@
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import { writeExecutionControl } from "../execution-control-notifications.js";
+import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { createRunUsageRecorder } from "../usage-receipts.js";
 import { applyObservabilityEnvironment, type ProviderObservabilityLaunch } from "@paperclipai/adapter-utils/agent-observability";
 import { museRunnerBroker } from "../muse-runner-broker.js";
@@ -9226,7 +9229,7 @@ async function executePaperclipNativeSessionWithinScope(
     };
     const stoppedLeaseRenewal = leaseRenewal.stop().catch(() => undefined);
     try {
-      await input.db
+      await writeExecutionControl(input.db, DELIVERY_QUEUES.executionFinalization, tx => tx
         .update(nativeRunFinalizations)
         .set({
           controlDeadlineAt: new Date(
@@ -9238,7 +9241,7 @@ async function executePaperclipNativeSessionWithinScope(
             eq(nativeRunFinalizations.runId, input.execution.binding.runId),
             eq(nativeRunFinalizations.leaseOwner, leaseOwner),
           ),
-        );
+        ));
 
       const failedAtMs = Date.now();
       const executionFailureMessage = redactIdentityText(
@@ -9280,7 +9283,7 @@ async function executePaperclipNativeSessionWithinScope(
           typeof stopIntent.intentAuditId === "string" && stopIntent.intentAuditId.length > 0 &&
           ["pending", "acknowledged"].includes(String(stopIntent.dispatchState));
         if (stoppedRun && (pendingStartupStop || hasAcknowledgedNativeStopIntent(stoppedRun) || hasAcknowledgedNativeReassignmentStopIntent(stoppedRun))) {
-          const [settled] = await input.db.update(nativeRunFinalizations).set({
+          const [settled] = await writeExecutionControl(input.db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(nativeRunFinalizations).set({
             phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null,
             leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: null,
             updatedAt: new Date(),
@@ -9290,7 +9293,7 @@ async function executePaperclipNativeSessionWithinScope(
             eq(nativeRunFinalizations.leaseOwner, leaseOwner),
             eq(nativeRunFinalizations.attempt, attempt),
             isNull(nativeRunFinalizations.resultId),
-          )).returning({ runId: nativeRunFinalizations.runId });
+          )).returning({ runId: nativeRunFinalizations.runId }));
           if (settled) {
             await stoppedLeaseRenewal;
             if (warmSessionId !== null && lifecyclePolicy.mode === "warm") {
@@ -9377,6 +9380,7 @@ async function executePaperclipNativeSessionWithinScope(
             ),
           )
           .for("update");
+        await notifyExecutionRunWork(tx);
         const updated = await tx
           .update(nativeRunFinalizations)
           .set({
diff --git a/tests/e2e/chat-adapters-ui-providers.spec.ts b/tests/e2e/chat-adapters-ui-providers.spec.ts
index 399a966d1bd..bdde190f2d7 100644
--- a/tests/e2e/chat-adapters-ui-providers.spec.ts
+++ b/tests/e2e/chat-adapters-ui-providers.spec.ts
@@ -180,6 +180,7 @@ test.describe.serial("native chat adapter UI", () => {
     await expect(page.getByLabel("App configuration access token", { exact: true })).toHaveCount(0);
     await expect(page.getByLabel("Signing Secret", { exact: true })).toHaveCount(0);
     await page.reload();
+    await expect(page.getByRole("heading", { name: "Send a message to your agent", exact: true })).toBeVisible();
     expect(mock.slackCreations).toBe(1); expect(mock.slackInstallations).toBe(1);
     mock.setWebhookVerified();
     await expect(page.getByRole("heading", { name: "Success", exact: true })).toBeVisible();