perf(heartbeat): drive retry promotion from durable deadlines (#18387)

Reviewed and verified at f1c48c924d.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/plans/2026-10-30-run-dispatch-sleep.md

@@ -0,0 +1,43 @@
+# Run dispatch without idle retry scans
+
+The run-dispatch module owns retry admission, its durable `scheduled_retry_at`
+deadline, and atomic promotion. The heartbeat queue owns concurrency, current
+execution authority, and dispatch. The shared coordinator owns notification
+coalescing, nonoverlap, transaction uncertainty, drain/resume and shutdown.
+The heartbeat facade and `index.ts` only connect these owners.
+
+A scheduled retry is a future obligation. Startup reconstructs its next stored
+deadline through the run-dispatch adapter. Writes that create, move, cancel or
+invalidate retries notify the retry worker in their transaction. Promotion uses
+the existing gate and compare-and-set, and its committed write wakes ordinary
+queued dispatch. A queued run behind occupied capacity waits for the owning
+run's lifecycle transition. Configuration and authority changes also wake it.
+Retry-schedule notifications are separate from ordinary queue eligibility.
+`notifyRunDispatch` wakes only queued dispatch; ordinary issue, agent, budget
+and company edits cannot repeatedly scan the global retry schedule.
+`notifyScheduledRetryDispatch` wakes retry reconciliation as well as dispatch
+for actual retry creation, rescheduling and cancellation. Its producers include
+bounded and operator-requested retries, recovery retries, native capacity and
+safe-replacement successors, and the existing retry-cancellation transition
+boundary. Shared runtime changes do not impose a spacing floor on a saved deadline.
+
+An idle queue has no recurring read or timer-driven dispatch attempt. Notifications batch for one second and carry the changed company to queued dispatch. Known retry deadlines still preempt batching. Cancellation-owned retries stay with their cancellation worker and do not retain a promotion deadline.
+
+The supported writer contract is service writes through the shared database
+connection scope, including caller-owned and nested transactions. Existing
+explicit domain notification boundaries cover lifecycle, issue review state,
+agent configuration/status, company lifecycle and scheduling settings. Direct
+SQL writers and independent replicas must provide an explicit ingress signal or
+restart reconstruction; a process-local signal does not observe their commits.
+This change adds no permanent database listener or write interception.
+
+The remaining periodic recovery chain still owns deferred comment receipts,
+execution-wait cancellation recovery, orphan/resource recovery, stranded issue
+assignments, dependency wake repair, task watchdogs, silent active runs and stale
+locks. Those duties retain their backstop until their separate producer/deadline
+migrations are qualified. This slice removes only the periodic scheduled-retry
+promotion call. Ordinary queued dispatch can still be reached by that backstop;
+the existing atomic claim and per-agent start lock prevent duplicate execution.
+
+These workers use internal process deadlines. They do not register Cloud wake
+alarms and do not relax the durable idle-sleep safety proof.

packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts

@@ -47,6 +47,9 @@ async function openOwnedWire(url: string): Promise<{ wire: PrpWireConnection; cl
           const size = initial === 126 ? buffer.readUInt16BE(2) : initial === 127 ? Number(buffer.readBigUInt64BE(2)) : initial;
           const offset = initial === 126 ? 4 : initial === 127 ? 10 : 2;
           if (buffer.length < offset + size) return;
+          // Upgrade head bytes can precede the caller's message subscription.
+          // Retain complete data frames until the control-plane listener exists.
+          if ((opcode === 1 || opcode === 2) && listeners.length === 0) return;
           const payload = buffer.subarray(offset, offset + size); buffer = buffer.subarray(offset + size);
           if (opcode === 8) { socket.destroy(); finish(); return; }
           if (opcode === 9) send(payload, 10);
@@ -56,12 +59,45 @@ async function openOwnedWire(url: string): Promise<{ wire: PrpWireConnection; cl
       socket.on("data", bytes => { buffer = Buffer.concat([buffer, bytes]); parse(); }); socket.on("close", finish); socket.on("error", finish);
       const close = () => { socket.destroy(); finish(); };
       resolvePeer({ wire: { sendJson: value => send(Buffer.from(JSON.stringify(value)), 1), close,
-        onJson: listener => { listeners.push(listener); }, onClose: listener => { closes.push(() => listener({})); } }, closed, close });
+        onJson: listener => { listeners.push(listener); parse(); }, onClose: listener => { closes.push(() => listener({})); } }, closed, close });
       queueMicrotask(parse);
     }); req.end();
   });
 }
 
+it("delivers a WebSocket message coalesced with the real HTTP upgrade response", async () => {
+  const message = { type: "hello", sentinel: "coalesced-upgrade" };
+  const server = createServer();
+  const sockets = new Set<import("node:net").Socket>();
+  server.on("connection", socket => { sockets.add(socket); socket.on("close", () => sockets.delete(socket)); });
+  server.on("upgrade", (req, socket) => {
+    const accept = createHash("sha1").update(String(req.headers["sec-websocket-key"]) + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").digest("base64");
+    const payload = Buffer.from(JSON.stringify(message));
+    socket.write(Buffer.concat([Buffer.from(`HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`),
+      Buffer.from([0x81, payload.length]), payload]));
+  });
+  let peer: Awaited<ReturnType<typeof openOwnedWire>> | undefined;
+  const messages: unknown[] = [];
+  try {
+    await new Promise<void>((resolve, reject) => {
+      server.once("error", reject);
+      server.listen(0, "127.0.0.1", () => { server.off("error", reject); resolve(); });
+    });
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("Missing coalesced-upgrade endpoint");
+    peer = await openOwnedWire(`ws://127.0.0.1:${address.port}/`);
+    peer.wire.onJson(value => messages.push(value));
+    await vi.waitFor(() => expect(messages).toEqual([message]), { timeout: 1000 });
+  } finally {
+    peer?.close();
+    const closing = server.listening
+      ? new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve()))
+      : Promise.resolve();
+    for (const socket of sockets) socket.destroy();
+    await closing;
+  }
+});
+
 // Real encrypted PRP and real Runner/provider processes; no model or server imports.
 it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every durable boundary without another process", async mode => {
   const root = await mkdtemp(join(tmpdir(), "runner-staged-handoff-"));
@@ -226,6 +262,7 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d
           coreWarm: currentCore?.store.state.warmTransition?.phase, header: checkpoint && inspectStagedRunAttachmentHeader(checkpoint.records.get("header")!)?.phase,
           runnerRun: physical.runId, runnerWarm: physical.warmTransition?.phase, lifecycle: physical.lifecycle,
           failureCode: physical.recoverableFailure?.code, failureKeys: Object.keys(physical.recoverableFailure ?? {}),
+          physicalFailure: physical.recoverableFailure,
         })}`, { cause: error });
       } finally {
         release();

server/src/__tests__/agent-timer-deadlines.test.ts

@@ -597,8 +597,10 @@ describe.skipIf(!support.supported)("agent timer owned deadlines", () => {
     await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } } });
     await vi.advanceTimersByTimeAsync(2000);
 
-
-    expect(idleWorkSnapshot().active).toBe(1);
+    // The producer committed: retain its resume deadline without keeping the
+    // instance busy throughout the drain. Uncertain writes still hold admission.
+    expect(idleWorkSnapshot().active).toBe(0);
+    expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now());
     const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction");
     try {
       await vi.advanceTimersByTimeAsync(60_000);
@@ -608,6 +610,11 @@ describe.skipIf(!support.supported)("agent timer owned deadlines", () => {
     stopTaskDrain();
     await vi.advanceTimersByTimeAsync(1000);
     await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 });
+    await vi.advanceTimersByTimeAsync(coordinator!.nextWakeAt()! - Date.now());
+    await vi.waitFor(() => expect(f.enqueueWakeup).toHaveBeenCalledOnce());
+    expect(f.enqueueWakeup).toHaveBeenCalledWith(f.agent.id, expect.objectContaining({ source: "timer", reason: "heartbeat_timer" }));
+    await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { enabled: false } } });
+    await quiet();
   });
 
   it("does not inspect timer rows during startup drain, then reconstructs them on resume", async () => {

server/src/__tests__/deadline-work.test.ts

@@ -1,4 +1,5 @@
-import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { installDatabaseWorkSignals, signalDatabaseWork, databaseWorkPending } from "../../../packages/db/src/work-signals.js";
 import { afterEach, expect, it, vi } from "vitest";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 import { registerDeadlineWork } from "../services/deadline-work.js";
@@ -39,3 +40,132 @@ it.each([
   expect(run).toHaveBeenCalledTimes(2);
   expect(onError).not.toHaveBeenCalled();
 });
+
+it("releases a settled first-write hold during drain and retains a probe-free resume wake", async () => {
+  vi.useFakeTimers();
+  let draining = false;
+  const owner = installDatabaseWorkSignals({
+    execute: vi.fn(async () => [{ status: "committed" }]),
+    transaction: async (callback: (tx: object) => Promise<unknown>) => callback({ execute: async () => [{ xid: "1" }] }),
+  });
+  const run = vi.fn(async () => {}), next = vi.fn(async () => null);
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !draining, canReconcile: () => true, onError: vi.fn() });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next, canRun: () => !draining, retryMs: 30_000 });
+  await worker.ready;
+  draining = true;
+  const before = idleWorkSnapshot().active;
+  await owner.transaction(async tx => { await signalDatabaseWork(tx, DELIVERY_QUEUES.agentTimers); });
+  await vi.advanceTimersByTimeAsync(0);
+  expect(idleWorkSnapshot().active).toBe(before);
+  expect(coordinator.nextWakeAt()).not.toBeNull();
+  await vi.advanceTimersByTimeAsync(3 * 30_000);
+  expect(run).toHaveBeenCalledOnce(); expect(next).toHaveBeenCalledOnce(); expect(owner.execute).not.toHaveBeenCalled();
+  draining = false;
+  await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull();
+});
+
+it("retains failed work through drain and preserves a future domain deadline", async () => {
+  vi.useFakeTimers();
+  let draining = false;
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  const run = vi.fn(async () => { if (run.mock.calls.length === 1) { draining = true; throw new Error("retry debt"); } });
+  const deadline = Date.now() + 600_000;
+  const next = vi.fn(async () => deadline);
+  const onError = vi.fn();
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !draining, canReconcile: () => true, onError });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next, canRun: () => !draining, retryMs: 30_000 });
+  await worker.ready;
+  await vi.advanceTimersByTimeAsync(90_000);
+  expect(run).toHaveBeenCalledOnce(); expect(next).not.toHaveBeenCalled(); expect(owner.execute).not.toHaveBeenCalled();
+  draining = false; await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBe(deadline);
+  draining = true; await vi.advanceTimersByTimeAsync(30_000);
+  expect(coordinator.nextWakeAt()).toBe(deadline); expect(onError).toHaveBeenCalledOnce();
+});
+
+it("does no domain probes in warm standby and reconstructs after admission opens", async () => {
+  vi.useFakeTimers();
+  let standby = true;
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  const run = vi.fn(async () => {}), next = vi.fn(async () => null);
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !standby, canReconcile: () => !standby, onError: vi.fn() });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next, canRun: () => !standby, retryMs: 30_000 });
+  await worker.ready; await vi.advanceTimersByTimeAsync(90_000);
+  expect(run).not.toHaveBeenCalled(); expect(next).not.toHaveBeenCalled(); expect(owner.execute).not.toHaveBeenCalled();
+  standby = false; await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull();
+});
+
+it("does not release an ambiguous database intent while dispatch is drained", async () => {
+  vi.useFakeTimers();
+  let draining = false, committed = false;
+  const owner = installDatabaseWorkSignals({
+    execute: vi.fn(async () => [{ status: committed ? "committed" : "in progress" }]),
+    transaction: async (callback: (tx: object) => Promise<unknown>) => {
+      await callback({ execute: async () => [{ xid: "2" }] });
+      throw new Error("lost acknowledgement");
+    },
+  });
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => !draining, canReconcile: () => true, onError: vi.fn() });
+  const run = vi.fn(async () => {});
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run, next: async () => null, canRun: () => !draining, retryMs: 30_000 });
+  await worker.ready; draining = true;
+  const before = idleWorkSnapshot().active;
+  await expect(owner.transaction(async tx => { await signalDatabaseWork(tx, DELIVERY_QUEUES.agentTimers); })).rejects.toThrow("lost acknowledgement");
+  await vi.advanceTimersByTimeAsync(0);
+  expect(idleWorkSnapshot().active).toBe(before + 1);
+  expect(databaseWorkPending(owner, DELIVERY_QUEUES.agentTimers)).toBe(true);
+  committed = true; await vi.advanceTimersByTimeAsync(30_000);
+  expect(databaseWorkPending(owner, DELIVERY_QUEUES.agentTimers)).toBe(false);
+  expect(idleWorkSnapshot().active).toBe(before);
+  draining = false; await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull();
+});
+
+it("backs off an expired deadline when only the domain is drained and resumes promptly", async () => {
+  vi.useFakeTimers();
+  let paused = false;
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  const run = vi.fn(async () => {});
+  const domainCanRun = vi.fn(() => !paused);
+  const deadline = Date.now() + 1_000;
+  const next = vi.fn(async () => run.mock.calls.length === 1 ? deadline : null);
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, {
+    run, next, canRun: domainCanRun, retryMs: 30_000,
+  });
+  await worker.ready;
+  paused = true;
+  const before = idleWorkSnapshot();
+  await vi.advanceTimersByTimeAsync(1_000);
+  expect(coordinator.nextWakeAt()).toBe(Date.now() + 30_000);
+  const attempts = domainCanRun.mock.calls.length;
+  await vi.advanceTimersByTimeAsync(90_000);
+  expect(domainCanRun.mock.calls.length - attempts).toBeLessThanOrEqual(12);
+  // Four attempts (expiry plus three retries), each begins and ends tracked work.
+  expect(idleWorkSnapshot().generation - before.generation).toBe(8);
+  expect(idleWorkSnapshot().active).toBe(before.active);
+  expect(run).toHaveBeenCalledOnce(); expect(next).toHaveBeenCalledOnce(); expect(owner.execute).not.toHaveBeenCalled();
+  paused = false;
+  await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull();
+});
+
+it("retains a prompt paused wake when a new request preempts a far-future deadline", async () => {
+  vi.useFakeTimers();
+  let paused = false;
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  const run = vi.fn(async () => {});
+  const deadline = Date.now() + 86_400_000;
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() });
+  const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, {
+    run, next: async () => run.mock.calls.length === 1 ? deadline : null, canRun: () => !paused, retryMs: 30_000,
+  });
+  await worker.ready; paused = true; worker.wake();
+  await vi.advanceTimersByTimeAsync(0);
+  expect(coordinator.nextWakeAt()).toBe(Date.now() + 30_000);
+  expect(owner.execute).not.toHaveBeenCalled();
+  paused = false; await vi.advanceTimersByTimeAsync(30_000);
+  expect(run).toHaveBeenCalledTimes(2); expect(coordinator.nextWakeAt()).toBeNull();
+});

server/src/__tests__/execution-control-work.test.ts

@@ -383,11 +383,11 @@ describe.skipIf(!support.supported)("execution-control durable work", () => {
     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)));
+        .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id)), { companyId });
       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)));
+      .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id)), { companyId });
     await quiet(c);
   });
   it("delivers a reconciled continuation when marking and resolving commit together", async () => {

server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts

@@ -0,0 +1,128 @@
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { PgDialect } from "drizzle-orm/pg-core";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { notifyRunDispatch, notifyScheduledRetryDispatch } from "../services/run-dispatch-notifications.js";
+import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js";
+
+const cleanup: Array<() => Promise<void>> = [];
+afterEach(async () => {
+  for (const stop of cleanup.splice(0)) await stop();
+  vi.useRealTimers();
+});
+function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean } = {}) {
+  vi.useFakeTimers();
+  vi.setSystemTime(1_000_000);
+  const dialect = new PgDialect();
+  const execute = vi.fn(async (query: any) => {
+    if (dialect.sqlToQuery(query).sql.includes("pg_current_xact_id")) return [{ xid: "42" }];
+    throw new Error("unexpected database query");
+  });
+  const db: { execute: typeof execute; transaction: (callback: (tx: any) => Promise<any>) => Promise<any> } = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise<any>): Promise<any> {
+    return callback({ execute, transaction: db.transaction });
+  } });
+  const state = { due: input.due ?? null as number | null, suppressed: input.suppressed ?? false,
+    draining: input.draining ?? false, standby: false, fail: false };
+  const promote = vi.fn(async () => {
+    if (state.fail) throw new Error("database unavailable");
+    if (state.due !== null && state.due <= Date.now()) state.due = null;
+  });
+  const next = vi.fn(async () => state.due);
+  const dispatch = vi.fn(async (_scope?: { companyId: string }) => {});
+  const coordinator = createDeliveryWorkCoordinator({ owner: db,
+    canRun: () => !state.draining && !state.standby, canReconcile: () => !state.standby, onError: vi.fn() });
+  const worker = registerRunDispatchWork(coordinator, {
+    promoteDueScheduledRetries: promote, nextScheduledRetryAt: next,
+    resumeQueuedRunsAfterDrain: dispatch, schedulingSuppressed: async () => state.suppressed,
+  }, () => !state.draining && !state.standby);
+  cleanup.push(() => coordinator.stop());
+  return { db, state, promote, next, dispatch, coordinator, worker,
+    notifyRetry: () => db.transaction(tx => notifyScheduledRetryDispatch(tx)),
+    notify: (scope?: { companyId: string }) => db.transaction(tx => notifyRunDispatch(tx, scope)),
+    settle: () => vi.advanceTimersByTimeAsync(1_000) };
+}
+
+describe("run dispatch obligations", () => {
+  it.each([false, true])("settles empty/feature-suppressed state and stops all owner work (suppressed=%s)", async suppressed => {
+    const f = fixture({ suppressed }); await f.worker.ready;
+    const calls = [f.promote.mock.calls.length, f.next.mock.calls.length, f.dispatch.mock.calls.length];
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+    await vi.advanceTimersByTimeAsync(20 * 60_000);
+    expect([f.promote.mock.calls.length, f.next.mock.calls.length, f.dispatch.mock.calls.length]).toEqual(calls);
+    expect(f.db.execute).not.toHaveBeenCalled();
+  });
+  it("restores an exact deadline and does no earlier recovery work", async () => {
+    const f = fixture({ due: 1_010_000 }); await f.worker.ready;
+    expect(f.coordinator.nextWakeAt()).toBe(1_010_000);
+    await vi.advanceTimersByTimeAsync(9_999); expect(f.promote).toHaveBeenCalledOnce();
+    await vi.advanceTimersByTimeAsync(1); expect(f.promote).toHaveBeenCalledTimes(2);
+    expect(f.state.due).toBeNull(); expect(f.coordinator.nextWakeAt()).toBeNull();
+    expect(f.dispatch).toHaveBeenCalledOnce();
+  });
+  it("rearms earlier work and cancels a future deadline on a committed producer change", async () => {
+    const f = fixture({ due: 1_600_000 }); await f.worker.ready;
+    f.state.due = 1_005_000; await f.notifyRetry(); await f.settle();
+    expect(f.coordinator.nextWakeAt()).toBe(1_005_000);
+    f.state.due = null; await f.notifyRetry(); await f.settle();
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+    const before = f.next.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.next).toHaveBeenCalledTimes(before);
+  });
+  it("batches repeated tenant writes and passes only their company to queued dispatch", async () => {
+    const f = fixture(); await f.worker.ready;
+    for (let index = 0; index < 100; index++) {
+      const source = { companyId: "company-a", issueId: `issue-${index % 2}`,
+        runId: `run-${index}`, agentId: `agent-${index}`, interactionBody: "not a dispatch key" };
+      await f.notify(source);
+      await vi.advanceTimersByTimeAsync(10);
+    }
+    expect(f.promote).toHaveBeenCalledOnce();
+    expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2);
+    await f.settle();
+    expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" });
+    for (const [scope] of f.dispatch.mock.calls.slice(1)) expect(scope).toEqual({ companyId: "company-a" });
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+    const count = f.dispatch.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.dispatch).toHaveBeenCalledTimes(count);
+  });
+  it("remembers the first producer arriving during drain without owner SQL", async () => {
+    const f = fixture(); await f.worker.ready; f.state.draining = true;
+    await f.notifyRetry(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000);
+    expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).toHaveBeenCalledOnce();
+    expect(idleWorkSnapshot().active).toBe(0);
+    f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000);
+    expect(f.promote).toHaveBeenCalledTimes(2); expect(f.dispatch).toHaveBeenCalledTimes(2);
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("releases a failed producer hold during drain and retains recovery for resume", async () => {
+    const f = fixture(); await f.worker.ready;
+    f.state.fail = true; await f.notifyRetry(); await f.settle();
+    expect(f.promote).toHaveBeenCalledTimes(2);
+    f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000);
+    expect(f.promote).toHaveBeenCalledTimes(2); expect(f.next).toHaveBeenCalledOnce();
+    expect(idleWorkSnapshot().active).toBe(0);
+    f.state.fail = false; f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000);
+    expect(f.promote).toHaveBeenCalledTimes(3); expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("preserves a notification during a nonoverlapping pass", async () => {
+    const f = fixture(); await f.worker.ready;
+    let release!: () => void;
+    const held = new Promise<void>(resolve => { release = resolve; });
+    f.promote.mockImplementationOnce(() => held);
+    await f.notifyRetry(); await f.settle();
+    await f.notifyRetry(); await f.settle();
+    expect(f.promote).toHaveBeenCalledTimes(2);
+    release(); await f.settle();
+    await vi.advanceTimersByTimeAsync(30_000);
+    expect(f.promote).toHaveBeenCalledTimes(3); expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("does not dispatch on warm standby and cancels deadlines at shutdown", async () => {
+    const f = fixture({ due: 1_010_000 }); f.state.standby = true; await f.worker.ready;
+    await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).not.toHaveBeenCalled();
+    f.state.standby = false; await vi.advanceTimersByTimeAsync(30_000);
+    expect(f.promote).toHaveBeenCalledOnce(); expect(f.state.due).toBeNull();
+    f.state.due = Date.now() + 60_000; await f.notifyRetry(); await f.settle(); await f.worker.stop();
+    await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2);
+  });
+});

server/src/__tests__/heartbeat-run-dispatch-work.test.ts

@@ -0,0 +1,284 @@
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { randomUUID } from "node:crypto";
+import { and, eq, sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns, issues } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { issueService } from "../services/issues.js";
+import { materializeNativeProviderCapacityRetry } from "../services/native-runtime/native-provider-capacity-retry.js";
+import { agentService } from "../services/agents.js";
+import { budgetService } from "../services/budgets.js";
+import { heartbeatService } from "../services/heartbeat.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
+import { notifyRunDispatch } from "../services/run-dispatch-notifications.js";
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RUN_DISPATCH_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+
+describe.skipIf(!support.supported)("run dispatch production wiring (controlled local process adapter)", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>;
+  let heartbeat: ReturnType<typeof heartbeatService>;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let worker: ReturnType<ReturnType<typeof heartbeatService>["registerRunDispatchWork"]> | undefined;
+  let reads: ReturnType<typeof vi.spyOn>;
+  let draining = false;
+  const errors: unknown[] = [];
+  const retryQueries: string[] = [];
+  beforeAll(async () => {
+    if (!external) database = await startEmbeddedPostgresTestDatabase("run-dispatch-work-");
+    db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+    db.$client.options.debug = (_connection, query) => {
+      if (query.startsWith("select") && query.includes('"heartbeat_runs"') &&
+        query.includes("startupCancellation") && query.includes('"scheduled_retry_at"')) retryQueries.push(query);
+    };
+    await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
+  }, 90_000);
+  beforeEach(async () => {
+    await db.execute(sql`TRUNCATE companies CASCADE`);
+    heartbeat = heartbeatService(db); draining = false; errors.length = 0;
+    reads = vi.spyOn(db, "select");
+  });
+  afterEach(async () => {
+    await worker?.stop(); await coordinator?.stop();
+    await heartbeat.drainActiveRunExecutions();
+    worker = undefined; coordinator = undefined;
+    vi.useRealTimers(); vi.restoreAllMocks();
+    expect(errors).toEqual([]);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
+
+  async function start() {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    worker = heartbeat.registerRunDispatchWork(coordinator, () => !draining);
+    await worker.ready;
+  }
+  async function fixture() {
+    const companyId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Dispatch", issuePrefix: `D${companyId.slice(0, 7)}`,
+      requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "dispatch-user" });
+    const [agent] = await db.insert(agents).values({ companyId, name: "Controlled process", role: "engineer", status: "idle",
+      adapterType: "process", adapterConfig: { command: process.execPath, args: ["-e", "console.log('dispatch-proof')"], cwd: process.cwd() },
+      runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, wakeOnDemand: true } },
+    }).returning();
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, status: "failed",
+      responsibleUserId: "dispatch-user", errorCode: "codex_transient_upstream", finishedAt: new Date(),
+      resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, contextSnapshot: {},
+    }).returning();
+    return { companyId, agentId: agent.id, runId: run.id };
+  }
+  async function quiet() {
+    await heartbeat.drainActiveRunExecutions();
+    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull(), { timeout: 5_000 });
+    const count = reads.mock.calls.length;
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try {
+      await vi.advanceTimersByTimeAsync(20 * 60_000);
+      expect(reads).toHaveBeenCalledTimes(count);
+    } finally { vi.useRealTimers(); }
+  }
+  async function status(id: string) {
+    return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]);
+  }
+  async function completed(id: string) {
+    await vi.waitFor(async () => expect(["running", "succeeded"]).toContain((await status(id))?.status), { timeout: 10_000 });
+    await heartbeat.drainActiveRunExecutions();
+    const completed = await status(id);
+    expect(completed.status).toBe("succeeded");
+    expect(completed.stdoutExcerpt).toContain("dispatch-proof");
+  }
+  it("keeps fresh empty, configured idle and retained terminal history quiet", async () => {
+    await start(); await quiet();
+    await fixture(); await worker!.stop(); await coordinator!.stop();
+    await start(); await quiet();
+  });
+  it.each(["agent", "issue"] as const)("does not scan global scheduled retries for repeated ordinary %s edits", async kind => {
+    const f = await fixture();
+    const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Edit source", status: "backlog" }).returning();
+    await start(); await quiet();
+    const before = retryQueries.length;
+    expect(before).toBeGreaterThan(0); // The passive observer saw real startup retry SQL.
+    for (let index = 0; index < 12; index++) {
+      if (kind === "agent") await agentService(db).update(f.agentId, { name: `Edited agent ${index}` });
+      else await issueService(db).update(issue.id, { title: `Edited issue ${index}` });
+      await new Promise(resolve => setTimeout(resolve, 100));
+    }
+    await quiet();
+    expect(retryQueries).toHaveLength(before);
+  });
+  it.each([false, true])("observes a native capacity retry only after its transaction settles (rollback=%s)", async rollback => {
+    const f = await fixture();
+    const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Capacity retry", status: "in_progress",
+      assigneeAgentId: f.agentId }).returning();
+    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.runId));
+    await start(); await quiet();
+    let retryId = "";
+    const writing = db.transaction(async tx => {
+      retryId = await materializeNativeProviderCapacityRetry({ tx: tx as unknown as typeof db, ...f, issueId: issue.id });
+      expect(databaseWorkPending(db, DELIVERY_QUEUES.scheduledRetryDispatch)).toBe(true);
+      expect(await status(retryId)).toBeUndefined();
+      if (rollback) throw new Error("native capacity rollback");
+    });
+    if (rollback) {
+      await expect(writing).rejects.toThrow("native capacity rollback");
+      await quiet(); expect(await status(retryId)).toBeUndefined();
+    } else {
+      await writing;
+      const retry = await status(retryId);
+      expect(retry.status).toBe("scheduled_retry");
+      await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(retry.scheduledRetryAt!.getTime()), { timeout: 5_000 });
+      await heartbeat.cancelRun(retryId, "fixture teardown", { suppressImmediateRecovery: true });
+      await quiet();
+    }
+  });
+  it("ordinary retry creation wakes the worker, respects the exact deadline, dispatches and settles", async () => {
+    const f = await fixture(); await start(); await quiet();
+    const result = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 4_000, random: () => 0 });
+    expect(result.outcome).toBe("scheduled");
+    if (result.outcome !== "scheduled") throw new Error("Expected retry creation");
+    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(result.dueAt.getTime()), { timeout: 5_000 });
+    expect((await status(result.run.id)).status).toBe("scheduled_retry");
+    await new Promise(resolve => setTimeout(resolve, 100));
+    expect(Date.now()).toBeLessThan(result.dueAt.getTime());
+    expect((await status(result.run.id)).status).toBe("scheduled_retry");
+    await completed(result.run.id);
+    expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.retryOfRunId, f.runId)))).toHaveLength(1);
+    await quiet();
+  });
+  it("reconstructs an overdue retry after process restart without a periodic tick", async () => {
+    const f = await fixture();
+    const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { now: new Date(Date.now() - 120_000), delayMs: 60_000 });
+    if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation");
+    heartbeat = heartbeatService(db);
+    await start(); await completed(scheduled.run.id); await quiet();
+  });
+  it("cancels a future retry and removes its old deadline", async () => {
+    const f = await fixture(); await start();
+    const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 600_000 });
+    if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation");
+    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(scheduled.dueAt.getTime()), { timeout: 5_000 });
+    await heartbeat.cancelRun(scheduled.run.id, "operator stop", { suppressImmediateRecovery: true });
+    await quiet(); expect((await status(scheduled.run.id)).status).toBe("cancelled");
+  });
+  it("retains the first retry written during drain and catches it up after resume", async () => {
+    const f = await fixture(); await start(); await quiet(); draining = true;
+    const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 200 });
+    if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation");
+    await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+    const count = reads.mock.calls.length;
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try {
+      await vi.advanceTimersByTimeAsync(90_000);
+      expect(reads).toHaveBeenCalledTimes(count);
+      expect(idleWorkSnapshot().active).toBe(0);
+    } finally { vi.useRealTimers(); }
+    expect((await status(scheduled.run.id)).status).toBe("scheduled_retry");
+    draining = false; worker!.wake();
+    await completed(scheduled.run.id); await quiet();
+  });
+  it("uses a committed budget raise to resume a queued run in the company it paused", async () => {
+    const f = await fixture();
+    const budgets = budgetService(db);
+    await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId, amount: 100,
+      metric: "billed_cents", windowKind: "lifetime", hardStopEnabled: true }, "dispatch-user");
+    const [cost] = await db.insert(costEvents).values({ companyId: f.companyId, agentId: f.agentId,
+      provider: "fixture", biller: "fixture", model: "controlled", inputTokens: 0, outputTokens: 0,
+      costCents: 125, occurredAt: new Date() }).returning();
+    await budgets.evaluateCostEvent(cost);
+    expect((await db.select().from(companies).where(eq(companies.id, f.companyId)))[0].status).toBe("paused");
+    const [queued] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
+      status: "queued", responsibleUserId: "dispatch-user", invocationSource: "automation", contextSnapshot: {} }).returning();
+    await start(); await quiet(); expect((await status(queued.id)).status).toBe("queued");
+    await budgets.upsertPolicy(f.companyId, { scopeType: "company", scopeId: f.companyId, amount: 10_000,
+      metric: "billed_cents", windowKind: "lifetime", hardStopEnabled: true }, "dispatch-user");
+    await completed(queued.id); await quiet();
+  });
+  it("clearing an agent error wakes saved queue work only after its own transaction commits", async () => {
+    const f = await fixture(); await start(); await quiet();
+    await db.update(agents).set({ status: "error" }).where(eq(agents.id, f.agentId));
+    const [queued] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
+      status: "queued", responsibleUserId: "dispatch-user", invocationSource: "automation", contextSnapshot: {} }).returning();
+    let writing: ReturnType<ReturnType<typeof agentService>["clearError"]> | undefined;
+    try {
+      await db.transaction(async blocker => {
+        await blocker.select().from(agents).where(eq(agents.id, f.agentId)).for("update");
+        writing = agentService(db).clearError(f.agentId);
+        void writing.catch(() => {});
+        await new Promise(resolve => setTimeout(resolve, 100));
+        expect(databaseWorkPending(db, DELIVERY_QUEUES.queuedRunDispatch)).toBe(false);
+        expect((await status(queued.id)).status).toBe("queued");
+        expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0].status).toBe("error");
+      });
+    } finally { await writing; }
+    await completed(queued.id); await quiet();
+  });
+  it("leaves cancellation-owned retries quiet even when their old promotion time is overdue", async () => {
+    const f = await fixture();
+    const scheduled = await heartbeat.scheduleBoundedRetry(f.runId, { now: new Date(Date.now() - 60_000), delayMs: 1_000 });
+    if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation");
+    await db.update(heartbeatRuns).set({ resultJson: { startupCancellation: { requestedAt: new Date().toISOString() },
+      cancellation: { expected: true, reason: "operator stop", source: "operator" } } }).where(eq(heartbeatRuns.id, scheduled.run.id));
+    await start(); await quiet();
+    expect((await status(scheduled.run.id)).status).toBe("scheduled_retry");
+    expect(await heartbeat.promoteDueScheduledRetries()).toEqual({ promoted: 0, runIds: [] });
+    await quiet();
+  });
+  it("rejects an error clear without publishing work when the lifecycle is no longer ready", async () => {
+    const f = await fixture(); await start(); await quiet();
+    await db.update(agents).set({ status: "error", lifecycleState: "paused" }).where(eq(agents.id, f.agentId));
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try {
+      await expect(agentService(db).clearError(f.agentId)).rejects.toThrow("Only agents in error status");
+      expect(coordinator!.nextWakeAt()).toBeNull();
+      const count = reads.mock.calls.length;
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(reads).toHaveBeenCalledTimes(count);
+    } finally { vi.useRealTimers(); }
+    expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0].status).toBe("error");
+    await quiet();
+  });
+  it("keeps a foreign company's future retry intact while concurrent workers claim one due retry", async () => {
+    const selected = await fixture(), foreign = await fixture(); await start();
+    const later = await heartbeat.scheduleBoundedRetry(foreign.runId, { delayMs: 600_000 });
+    const due = await heartbeat.scheduleBoundedRetry(selected.runId, { delayMs: 500 });
+    if (later.outcome !== "scheduled" || due.outcome !== "scheduled") throw new Error("Expected retries");
+    const other = heartbeatService(db);
+    const competing = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const competingWorker = other.registerRunDispatchWork(competing, () => true);
+    try {
+      await competingWorker.ready; await completed(due.run.id); await other.drainActiveRunExecutions();
+      expect((await status(later.run.id)).status).toBe("scheduled_retry");
+      expect((await status(later.run.id)).scheduledRetryAt).toEqual(later.dueAt);
+      expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, selected.runId))).toHaveLength(1);
+      await heartbeat.cancelRun(later.run.id, "fixture teardown", { suppressImmediateRecovery: true });
+      await quiet();
+    } finally { await competingWorker.stop(); await competing.stop(); await other.drainActiveRunExecutions(); }
+  });
+  it("holds nested producer work until outer commit and rolls back without a successor", async () => {
+    const f = await fixture(); await start(); await quiet();
+    let release!: () => void; const held = new Promise<void>(resolve => { release = resolve; });
+    let queuedId = "";
+    const writing = db.transaction(async tx => {
+      await tx.transaction(async nested => {
+        await notifyRunDispatch(nested);
+        const [row] = await nested.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
+          responsibleUserId: "dispatch-user", status: "queued", contextSnapshot: {} }).returning();
+        queuedId = row.id;
+      });
+      await held;
+    });
+    await vi.waitFor(() => expect(queuedId).not.toBe(""));
+    await new Promise(resolve => setTimeout(resolve, 100));
+    expect(await status(queuedId)).toBeUndefined();
+    release(); await writing; await completed(queuedId); await quiet();
+    await expect(db.transaction(async tx => {
+      await tx.transaction(async nested => { await notifyRunDispatch(nested); await nested.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "queued" }); });
+      throw new Error("rollback outer");
+    })).rejects.toThrow("rollback outer");
+    await quiet();
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(2);
+  });
+});

server/src/__tests__/managed-skill-content.test.ts

@@ -1,3 +1,5 @@
+import { createServer } from "node:http";
+import { createS3StorageProvider } from "../storage/s3-provider.js";
 import { skillInstallationMaintenance } from "../services/skill-installation-maintenance.js";
 import { removeRuntimeAssetTree } from "../services/native-runtime/runtime-asset-retention.js";
 import { eq } from "drizzle-orm";
@@ -77,6 +79,93 @@ describe.skipIf(!support.supported)("managed skills use committed versions", ()
       .toMatchObject({ content: "complete root guide" });
   });
 
+  it("loads persisted Cloud configuration for read-only maintenance without seeding it", async () => {
+    const companyId = await company();
+    const objects = new Map<string, Buffer>();
+    const server = createServer((request, response) => {
+      void (async () => {
+        const key = new URL(request.url!, "http://fixture").pathname;
+        if (request.method === "PUT") {
+          const chunks: Buffer[] = [];
+          for await (const chunk of request) chunks.push(Buffer.from(chunk));
+          let body = Buffer.concat(chunks);
+          if (request.headers["content-encoding"]?.includes("aws-chunked")) {
+            const decoded: Buffer[] = [];
+            let offset = 0;
+            while (offset < body.length) {
+              const end = body.indexOf("\r\n", offset);
+              const size = Number.parseInt(body.subarray(offset, end).toString().split(";")[0]!, 16);
+              if (size === 0) break;
+              decoded.push(body.subarray(end + 2, end + 2 + size)); offset = end + 2 + size + 2;
+            }
+            body = Buffer.concat(decoded);
+          }
+          objects.set(key, body); response.writeHead(200, { etag: '"fixture"' }).end(); return;
+        }
+        const body = objects.get(key);
+        if (!body) { response.writeHead(404).end(); return; }
+        response.writeHead(200, { "content-length": body.length, "content-type": "application/octet-stream", etag: '"fixture"' });
+        response.end(request.method === "HEAD" ? undefined : body);
+      })().catch(() => { response.writeHead(500).end(); });
+    });
+    await new Promise<void>(resolve => { server.listen(0, "127.0.0.1", resolve); });
+    const address = server.address(); if (!address || typeof address === "string") throw new Error("Expected fixture address");
+    const endpoint = `http://127.0.0.1:${address.port}`;
+    const credentials = { AWS_ACCESS_KEY_ID: "fixture-access-key", AWS_SECRET_ACCESS_KEY: "fixture-secret-key", AWS_SESSION_TOKEN: "" };
+    const previousCredentials = Object.fromEntries(Object.keys(credentials).map(key => [key, process.env[key]]));
+    Object.assign(process.env, credentials);
+    vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createS3StorageProvider({
+      bucket: "managed-skill-fixture", region: "us-east-1", endpoint, forcePathStyle: true,
+    })));
+    const key = "cloud-runtime-config:v1";
+    try {
+      await companySkillService(db).createLocalSkill(companyId, { name: "Cloud CLI", markdown: "# Cloud CLI" });
+      const run = (phase: string) => promisify(execFile)(process.execPath, [
+        "--import", fileURLToPath(import.meta.resolve("tsx")),
+        fileURLToPath(new URL("../../scripts/migrate-skill-packages.ts", import.meta.url)),
+        `--phase=${phase}`, `--company=${companyId}`, "--limit=50",
+      ], { env: { ...process.env, ...credentials, DATABASE_URL: database.connectionString,
+        PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "disposable-cloud-cli-fixture",
+        PAPERCLIP_SECRETS_MASTER_KEY: "a".repeat(64), PAPERCLIP_STORAGE_PROVIDER: "s3",
+        PAPERCLIP_STORAGE_S3_BUCKET: "managed-skill-fixture", PAPERCLIP_STORAGE_S3_REGION: "us-east-1",
+        PAPERCLIP_STORAGE_S3_ENDPOINT: endpoint, PAPERCLIP_STORAGE_S3_FORCE_PATH_STYLE: "true", PAPERCLIP_STORAGE_S3_PREFIX: "" },
+        timeout: 60_000, maxBuffer: 1024 * 1024 });
+      await expect(run("eligibility")).rejects.toThrow("Initialize Cloud runtime configuration");
+      expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key))).toHaveLength(0);
+      await db.insert(instanceSettings).values({ singletonKey: key, general: { version: 1, revision: 0, config: null } });
+      const eligibility = JSON.parse((await run("eligibility")).stdout.trim().split("\n").at(-1)!);
+      expect(eligibility.results).toEqual([expect.objectContaining({ status: "version_backed" })]);
+      const before = [...objects.entries()].map(([key, body]) => [key, Buffer.from(body)] as const);
+      const verification = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
+      expect(verification.results).toEqual([expect.objectContaining({ status: "remote_storage" })]);
+      expect([...objects.entries()]).toEqual(before);
+      const archiveKey = [...objects.keys()].find(key => key.endsWith("/files.tar.gz"))!;
+      const archive = objects.get(archiveKey)!;
+      const corrupted = Buffer.from(archive); corrupted[0] = corrupted[0]! ^ 1;
+      objects.set(archiveKey, corrupted);
+      const corrupt = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
+      expect(corrupt.results).toEqual([expect.objectContaining({ status: "corrupt_package" })]);
+      objects.set(archiveKey, archive);
+      const manifestKey = [...objects.keys()].find(key => key.endsWith("/manifest.json"))!;
+      const manifest = objects.get(manifestKey)!; objects.delete(manifestKey);
+      const unavailable = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
+      expect(unavailable.results).toEqual([expect.objectContaining({ status: "unavailable_or_corrupt_package" })]);
+      objects.set(manifestKey, manifest);
+      const recovered = JSON.parse((await run("verification")).stdout.trim().split("\n").at(-1)!);
+      expect(recovered.results).toEqual([expect.objectContaining({ status: "remote_storage" })]);
+      expect([...objects.entries()].sort()).toEqual([...before].sort());
+      expect(await db.select().from(instanceSettings).where(eq(instanceSettings.singletonKey, key)))
+        .toEqual([expect.objectContaining({ general: { version: 1, revision: 0, config: null } })]);
+    } finally {
+      await db.delete(instanceSettings).where(eq(instanceSettings.singletonKey, key));
+      vi.mocked(storageModule.getStorageService).mockReturnValue(createStorageService(createLocalDiskStorageProvider(path.join(root, "objects"))));
+      for (const [key, value] of Object.entries(previousCredentials)) {
+        if (value === undefined) delete process.env[key]; else process.env[key] = value;
+      }
+      server.closeAllConnections(); await new Promise<void>(resolve => { server.close(() => resolve()); });
+    }
+  }, 180_000);
+
   it.each([
     { storage: "legacy local", env: { PAPERCLIP_STORAGE_PROVIDER: "local_disk" }, expectedStatus: "local_storage" },
     { storage: "shared S3", env: { PAPERCLIP_STORAGE_PROVIDER: "s3",

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -134,6 +134,7 @@ const {
     })),
     reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })),
     reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })),
+    registerRunDispatchWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerRetainedNativeSessionCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })),
@@ -923,6 +924,18 @@ describe("startServer feedback export wiring", () => {
     if (enabled) expect(heartbeatServiceMock.registerCapturedInstructionCommitWork.mock.calls[0][1]()).toBe(true);
   });
 
+  it.each([false, true])("registers retry and queued dispatch only when maintenance is enabled (%s)", async enabled => {
+    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
+    await startServer(); await settleStartupHousekeeping();
+    expect(heartbeatServiceMock.registerRunDispatchWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
+    if (enabled) {
+      const [coordinator, admitted] = heartbeatServiceMock.registerRunDispatchWork.mock.calls[0];
+      const app = await createAppMock.mock.results.at(-1)!.value;
+      expect(coordinator).toBe(app.locals.deliveryWork);
+      expect(admitted()).toBe(true);
+    }
+  });
+
   it.each([false, true])("registers retry cancellation recovery only when maintenance is enabled (%s)", async enabled => {
     loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
     await startServer(); await settleStartupHousekeeping();
@@ -1328,6 +1341,7 @@ describe("startServer feedback export wiring", () => {
       expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
       expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true });
       expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
+      expect(heartbeatServiceMock.promoteDueScheduledRetries).not.toHaveBeenCalled();
       expect(retiredDetector).not.toHaveBeenCalled();
     } finally {
       delete (runtime as Partial<typeof runtime>).reconcileProductivityReviews;

server/src/index.ts

@@ -1778,7 +1778,6 @@ async function startServerWithDatabaseTeardown(
           }
         }
 
-        const promotion = await timeStartupRecovery("scheduled_retries", () => heartbeat.promoteDueScheduledRetries());
         await timeStartupRecovery("queued_runs", () => heartbeat.resumeQueuedRuns());
         const recoveredGoalActions = await timeStartupRecovery("goal_actions", () => heartbeat.recoverPendingSessionGoalActions());
         if (
@@ -1799,7 +1798,6 @@ async function startServerWithDatabaseTeardown(
         }
         const reconciled = await heartbeat.reconcileStrandedAssignedIssues();
         if (
-          promotion.promoted > 0 ||
           reconciled.assignmentDispatched > 0 ||
           reconciled.dispatchRequeued > 0 ||
           reconciled.continuationRequeued > 0 ||
@@ -1808,7 +1806,7 @@ async function startServerWithDatabaseTeardown(
           reconciled.escalated > 0
         ) {
           logger.warn(
-            { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled },
+            { ...reconciled },
             "startup heartbeat recovery changed assigned issue state",
           );
         }
@@ -1847,6 +1845,8 @@ async function startServerWithDatabaseTeardown(
       await startupHeartbeatRecovery;
     }
 
+    if (config.heartbeatSchedulerEnabled) await heartbeat.registerRunDispatchWork(deliveryWork,
+      () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready;
     if (config.heartbeatSchedulerEnabled) await heartbeat.registerScheduledRetryCancellationWork(deliveryWork,
       () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
 
@@ -2049,12 +2049,10 @@ async function startServerWithDatabaseTeardown(
           // persisted queued work is still being driven forward.
           runPeriodicRecovery(() => heartbeat
             .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true })
-            .then(() => heartbeat.promoteDueScheduledRetries())
-            .then(async (promotion) => {
+            .then(async () => {
               await heartbeat.resumeQueuedRuns();
               const reconciled = await heartbeat.reconcileStrandedAssignedIssues({ skipExternalConversationSettlement: true });
               if (
-                promotion.promoted > 0 ||
                 reconciled.assignmentDispatched > 0 ||
                 reconciled.dispatchRequeued > 0 ||
                 reconciled.continuationRequeued > 0 ||
@@ -2063,7 +2061,7 @@ async function startServerWithDatabaseTeardown(
                 reconciled.escalated > 0
               ) {
                 logger.warn(
-                  { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled },
+                  { ...reconciled },
                   "periodic heartbeat recovery changed assigned issue state",
                 );
               }

server/src/modules/agent-lifecycle/adapters/records.ts

@@ -80,22 +80,20 @@ export function agentRecords(db: Db, effects: LifecycleEffects) {
         throw conflict("Only agents in error status can have their error cleared");
       }
 
-      const updated = await db
-        .update(agents)
-        .set({
-          status: "idle",
-          pauseReason: null,
-          pausedAt: null,
-          errorReason: null,
-          updatedAt: new Date(),
-        })
-        .where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready")))
-        .returning()
-        .then((rows) => rows[0] ?? null);
+      const updated = await effects.transaction(db, existing.companyId, async tx => {
+        const [current] = await tx.select({ status: agents.status, lifecycleState: agents.lifecycleState })
+          .from(agents).where(and(eq(agents.id, id), eq(agents.companyId, existing.companyId))).for("no key update");
+        if (current?.status !== "error" || current.lifecycleState !== "ready") {
+          throw conflict("Only agents in error status can have their error cleared");
+        }
+        await effects.beforeStatusChange(tx, existing.companyId, id, "error", "idle");
+        const [saved] = await tx.update(agents).set({
+          status: "idle", pauseReason: null, pausedAt: null, errorReason: null, updatedAt: new Date(),
+        }).where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready"))).returning();
+        if (!saved) throw conflict("Only agents in error status can have their error cleared");
+        return saved;
+      });
 
-      if (!updated) {
-        throw conflict("Only agents in error status can have their error cleared");
-      }
       return getById(updated.id);
     },
 

server/src/modules/run-dispatch/adapters/postgres.ts

@@ -511,6 +511,15 @@ export function createPostgresRunDispatchAdapter(
     return decideScheduledRetryGate(factsResult.facts, input.now);
   }
 
+  async function nextScheduledRetryAt(input: { cutoff: Date | null }): Promise<number | null> {
+    const [row] = await db.select({ at: heartbeatRuns.scheduledRetryAt }).from(heartbeatRuns)
+      .where(and(eq(heartbeatRuns.status, "scheduled_retry"),
+        sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') is distinct from 'object'`,
+        input.cutoff ? gte(heartbeatRuns.createdAt, input.cutoff) : undefined))
+      .orderBy(asc(heartbeatRuns.scheduledRetryAt)).limit(1);
+    return row?.at?.getTime() ?? null;
+  }
+
   async function listDueRetries(input: ListDueRetriesInput): Promise<DueRetryRun[]> {
     const rows = await db
       .select()
@@ -518,6 +527,7 @@ export function createPostgresRunDispatchAdapter(
       .where(
         and(
           eq(heartbeatRuns.status, "scheduled_retry"),
+          sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') is distinct from 'object'`,
           lte(heartbeatRuns.scheduledRetryAt, input.now),
           input.cutoff ? gte(heartbeatRuns.createdAt, input.cutoff) : undefined,
         ),
@@ -1414,6 +1424,7 @@ export function createPostgresRunDispatchAdapter(
 
   return {
     evaluateScheduledRetryGate,
+    nextScheduledRetryAt,
     listDueRetries,
     cancelStaleQueuedRun,
     dispatchResolvedInteractionIfCurrent,

server/src/modules/run-dispatch/application/ports.ts

@@ -24,6 +24,7 @@ export type EvaluateScheduledRetryGateInput = {
 
 /** Read-only scheduled-retry operations exposed to application use cases. */
 export interface ScheduledRetryReader {
+  nextScheduledRetryAt(input: { cutoff: Date | null }): Promise<number | null>;
   evaluateScheduledRetryGate(input: EvaluateScheduledRetryGateInput): Promise<GateDecision>;
   listDueRetries(input: ListDueRetriesInput): Promise<DueRetryRun[]>;
 }

server/src/modules/run-dispatch/application/use-cases.test.ts

@@ -12,6 +12,7 @@ function fakeReader(dueRuns: DueRetryRun[] = []): ScheduledRetryReader & { evalu
   const evaluateCalls: unknown[] = [];
   return {
     evaluateCalls,
+    async nextScheduledRetryAt() { return null; },
     async evaluateScheduledRetryGate(input) {
       evaluateCalls.push(input);
       return { allowed: true };

server/src/modules/run-dispatch/index.ts

@@ -67,6 +67,7 @@ export function createRunDispatch(db: Db, deps: RunDispatchDeps = {}) {
   });
 
   return {
+    nextScheduledRetryAt: (input: { cutoff: Date | null }) => adapter.nextScheduledRetryAt(input),
     evaluateScheduledRetryGate: createEvaluateScheduledRetryGate({ reader: adapter }),
     promoteScheduledRetry,
     promoteDueScheduledRetries: createPromoteDueScheduledRetries({

server/src/modules/wake-queue/adapters/postgres.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "../../../services/run-dispatch-notifications.js";
 import { notifyExternalConversationSettlement } from "../../../services/external-conversation-settlement-notifications.js";
 import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js";
@@ -637,6 +638,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
         .returning()
         .then((rows) => rows[0]);
 
+      await notifyRunDispatch(tx, { companyId });
       const queuedRun = await tx
         .insert(heartbeatRuns)
         .values({
@@ -715,6 +717,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
         .returning()
         .then((rows) => rows[0]);
 
+      await notifyRunDispatch(tx, { companyId });
       const queuedRun = await tx
         .insert(heartbeatRuns)
         .values({

server/src/services/agent-configuration.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifySkillVersionPinChange } from "./skill-version-retention-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
@@ -83,6 +84,7 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura
       existing.adapterConfig,
     );
   }
+  await notifyRunDispatch(db, { companyId: existing.companyId });
   if (normalizedPatch.runtimeConfig !== undefined || normalizedPatch.reportsTo !== undefined) {
     await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
   }

server/src/services/agent-lifecycle.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
 import { isAgentStatusInvokable } from "@paperclipai/shared";
 import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js";
@@ -38,7 +39,10 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li
     },
     notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle),
     async beforeStatusChange(tx, companyId, agentId, before, after) {
-      if (before !== after) await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+      if (before !== after) {
+        await notifyRunDispatch(tx, { companyId });
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+      }
       if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) {
         await notifyNativeResponsePresentation(tx, { companyId, agentId });
         await notifyNativeReviewResponse(tx, { companyId, agentId });

server/src/services/budgets.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyApprovalWork } from "./approval-work-notifications.js";
 import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
 import { addCents, compareCents, normalizeCents } from "@paperclipai/shared";
@@ -300,6 +301,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
     }
 
     if (policy.scopeType === "project") {
+      await notifyRunDispatch(db, { companyId: policy.companyId });
       await db
         .update(projects)
         .set({
@@ -313,6 +315,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
 
     const scope = await resolveScopeRecord(db, "company", policy.scopeId);
     if (scope.paused) return;
+    await notifyRunDispatch(db, { companyId: policy.companyId });
     await notifyCompanyRetentionWork(db, policy.companyId);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
@@ -359,6 +362,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
     }
 
     if (policy.scopeType === "project") {
+      await notifyRunDispatch(db, { companyId: policy.companyId });
       await db
         .update(projects)
         .set({
@@ -373,6 +377,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
     await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync);
     const scope = await resolveScopeRecord(db, "company", policy.scopeId);
     if (scope.pauseReason !== "budget") return;
+    await notifyRunDispatch(db, { companyId: policy.companyId });
     await notifyCompanyRetentionWork(db, policy.companyId);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);

server/src/services/chat-run-milestone-notifications.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyIssueReviewState } from "./issue-review-state-notifications.js";
 import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js";
 import { notifyChatResponseAvailability } from "./chat-response-availability.js";
@@ -11,6 +12,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type ChatRunMilestoneScope = { companyId: string; issueId?: string; runId?: string };
 
 export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) {
+  if (!scope.issueId && !scope.runId) await notifyRunDispatch(tx, scope);
   if (scope.issueId || scope.runId) {
     await notifyIssueReviewState(tx, { companyId: scope.companyId, issueId: scope.issueId,
       runId: scope.issueId ? undefined : scope.runId });

server/src/services/companies.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
 import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -332,6 +333,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
         const renamedPrefix = await resolveRenamedIssuePrefix(tx, id, companyPatch);
 
         if (companyPatch.status !== undefined && companyPatch.status !== existing.status) {
+          await notifyRunDispatch(tx, { companyId: id });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
@@ -488,6 +490,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
         const wasAlreadyArchived = existing.status === "archived";
 
         if (!wasAlreadyArchived) {
+          await notifyRunDispatch(tx, { companyId: id });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);

server/src/services/deadline-work.ts

@@ -26,7 +26,7 @@ export function registerDeadlineWork(
     pausePassiveMaintenance: true,
     onSettled: hints => { requested++; task.onSettled?.(hints); },
     run: async signal => {
-      if (!task.canRun()) { requested++; next = null; return; }
+      if (!task.canRun()) { requested++; return; }
       if (Date.now() < earliest) { next = earliest; return; }
       if (Date.now() < coalesceUntil && (next === null || next > Date.now())) {
         next = next === null ? coalesceUntil : Math.min(next, coalesceUntil);
@@ -39,8 +39,6 @@ export function registerDeadlineWork(
       if (signal.aborted || !task.canRun()) { next = null; return; }
       const deadline = await task.next();
 
-
-
       if (!task.canRun()) { next = null; return; }
       earliest = deadline === null || deadline > Date.now()
         ? (task.minimumSpacingMs ? Date.now() + task.minimumSpacingMs : 0)
@@ -49,8 +47,13 @@ export function registerDeadlineWork(
       coalesceUntil = Date.now() + (task.notificationCoalesceMs ?? 0);
       reconciled = generation;
     },
-    hasPending: async () => requested !== reconciled,
-    nextRunAt: () => next,
+    // A settled producer must release its admission hold during drain. Keep
+    // the unreconciled generation and an in-memory resume wake, without asking
+    // the domain to probe its database while dispatch is closed.
+    hasPending: async () => task.canRun() && requested !== reconciled,
+    nextRunAt: () => !task.canRun() && requested !== reconciled
+      ? Date.now() + task.retryMs
+      : next,
   });
   return { ...worker, wake: () => { requested++; worker.wake(); } };
 }

server/src/services/delivery-work-notifications.ts

@@ -2,6 +2,8 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  scheduledRetryDispatch: "scheduled-retry-dispatch",
+  queuedRunDispatch: "queued-run-dispatch",
   runEvidenceMigration: "run-evidence-migration",
   providerTraceRetention: "provider-trace-retention",
   nativeRunnerEvidenceRetention: "native-runner-evidence-retention",

server/src/services/execution-control-notifications.ts

@@ -1,7 +1,11 @@
+import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js";
 
-export async function notifyExecutionRunWork(transaction: object) {
+export type ExecutionWorkScope = { companyId: string; agentId?: string; runId?: string };
+
+export async function notifyExecutionRunWork(transaction: object, scope?: ExecutionWorkScope) {
+  if (scope) await notifyRunDispatch(transaction, scope);
   for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
     DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
     await notifyDeliveryWork(transaction, queue);
@@ -12,8 +16,10 @@ export function writeExecutionControl<T>(
   db: Db | Parameters<Parameters<Db["transaction"]>[0]>[0],
   queue: DeliveryQueue | readonly DeliveryQueue[],
   write: (tx: Db) => PromiseLike<T>,
+  scope: ExecutionWorkScope | undefined,
 ): Promise<T> {
   return db.transaction(async transaction => {
+    await notifyRunDispatch(transaction, scope);
     await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
     for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
     return write(transaction as unknown as Db);

server/src/services/execution-control-reconciliation.ts

@@ -196,7 +196,7 @@ export async function reconcileAbandonedExecutionControl(
           await notifyNativeEnvironmentSettlement(tx, { companyId: run.companyId, runId: run.id });
           await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
           await notifyUnsafeWorkspaceExportRunChange(tx, run);
-          await notifyExecutionRunWork(tx);
+          await notifyExecutionRunWork(tx, { companyId: run.companyId });
           const [updatedRun] = await tx
             .update(heartbeatRuns)
             .set({

server/src/services/execution-recovery-resolution.ts

@@ -546,7 +546,7 @@ export async function settleUnrecoverableExecutions(
             companyId: run.companyId, issueId: task.id, runId: run.id,
           }).limit(1).for("update");
           if (!hold) return;
-          await notifyExecutionRunWork(tx);
+          await notifyExecutionRunWork(tx, { companyId: run.companyId });
           const note = "Exact session recovery stopped. The recorded workspace recovery holds this execution without replaying its actions.";
           await tx.update(issueRecoveryActions).set({ status: "cancelled", outcome: "cancelled",
             resolvedAt: now, updatedAt: now, wakePolicy: null, monitorPolicy: null,
@@ -654,7 +654,7 @@ export async function settleUnrecoverableExecutions(
             replay: "not_authorized",
           },
         });
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: run.companyId });
         await tx
           .update(heartbeatRuns)
           .set({ executionStatusDeliveryId: randomUUID() })

server/src/services/explicit-native-continuation.ts

@@ -475,7 +475,7 @@ export async function admitExplicitNativeContinuation(input: {
   }).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.queuedCommentInterruptId),
     eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution")));
   if (cancelledStartupIds.size) await db.transaction(async tx => {
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId, agentId });
     for (const runId of cancelledStartupIds) {
       await tx.update(nativeRunFinalizations).set({
         phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null,

server/src/services/heartbeat.ts

@@ -1,3 +1,4 @@
+import { registerRunDispatchWork } from "./heartbeat/dispatch-work.js";
 import { NATIVE_RETAINED_OWNER_WAIT } from "./native-runtime/native-controller-recovery.js";
 import { externalConversationSettlementWork } from "./external-conversation-settlement-work.js";
 import { nativeRecoveryNoticeWork } from "./native-runtime/native-recovery-notice-work.js";
@@ -1003,6 +1004,7 @@ export function heartbeatService(
     finalizeAiConnectionBusyDeferral,
     finalizeWorkspaceBusyDeferral,
     promoteDueScheduledRetries,
+    nextScheduledRetryAt,
     retryScheduledRetryNow,
     scheduleBoundedRetry,
   } = createHeartbeatRetries(db, {
@@ -1337,11 +1339,11 @@ export function heartbeatService(
     return heartbeatQueue.dispatchPendingNativeStatusWakeups(...args);
   }
 
-  function resumeQueuedRunsAfterDrain() {
+  function resumeQueuedRunsAfterDrain(...args: Parameters<typeof heartbeatQueue.resumeQueuedRunsAfterDrain>) {
     for (const pending of nativeRecoveryDeadlines.values()) {
       pending.rearm();
     }
-    const promise = heartbeatQueue.resumeQueuedRunsAfterDrain();
+    const promise = heartbeatQueue.resumeQueuedRunsAfterDrain(...args);
     activeWakeupPromises.add(promise);
     void promise.then(
       () => { activeWakeupPromises.delete(promise); },
@@ -3712,7 +3714,7 @@ export function heartbeatService(
           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)), { companyId: run.companyId });
         }
         nativeOwnershipHeld =
           nativeOwnershipHeld ||
@@ -4234,6 +4236,13 @@ export function heartbeatService(
     computeTaskDrain,
     applyTaskDrain,
 
+    registerRunDispatchWork: (coordinator: Parameters<typeof registerRunDispatchWork>[0], canRun: () => boolean) =>
+      registerRunDispatchWork(coordinator, {
+        promoteDueScheduledRetries,
+        nextScheduledRetryAt,
+        resumeQueuedRunsAfterDrain,
+        schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed,
+      }, () => canRun() && readTaskDrain(new Date()) === null),
     promoteDueScheduledRetries,
     retryScheduledRetryNow,
 

server/src/services/heartbeat/dispatch-work.ts

@@ -0,0 +1,58 @@
+import type { RunDispatchScope } from "../run-dispatch-notifications.js";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { registerDeadlineWork } from "../deadline-work.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+
+export function registerRunDispatchWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  owner: {
+    promoteDueScheduledRetries(): Promise<unknown>;
+    nextScheduledRetryAt(): Promise<number | null>;
+    resumeQueuedRunsAfterDrain(scope?: RunDispatchScope): Promise<unknown>;
+    schedulingSuppressed(): Promise<boolean>;
+  },
+  canRun: () => boolean,
+) {
+  let suppressed = false;
+  const retries = registerDeadlineWork(coordinator, DELIVERY_QUEUES.scheduledRetryDispatch, {
+    retryMs: 30_000,
+    continuationMs: 1_000,
+    notificationCoalesceMs: 1_000,
+    canRun,
+    run: async signal => {
+      suppressed = await owner.schedulingSuppressed();
+      if (!signal.aborted && !suppressed && canRun()) await owner.promoteDueScheduledRetries();
+    },
+    next: () => suppressed ? Promise.resolve(null) : owner.nextScheduledRetryAt(),
+  });
+  const scopes = new Set<string>(["*"]);
+  const queue = registerDeadlineWork(coordinator, DELIVERY_QUEUES.queuedRunDispatch, {
+    retryMs: 30_000,
+    notificationCoalesceMs: 1_000,
+    canRun,
+    onSettled: hints => { for (const hint of hints ?? ["*"]) scopes.add(hint); },
+    run: async signal => {
+      const requested = scopes.has("*") ? ["*"] : [...scopes];
+      if (requested[0] === "*") scopes.clear();
+      else for (const hint of requested) scopes.delete(hint);
+      for (let index = 0; index < requested.length; index++) {
+        if (signal.aborted || !canRun()) {
+          for (const hint of requested.slice(index)) scopes.add(hint);
+          return;
+        }
+        try {
+          await owner.resumeQueuedRunsAfterDrain(requested[index] === "*" ? undefined : JSON.parse(requested[index]) as RunDispatchScope);
+        } catch (error) {
+          for (const hint of requested.slice(index)) scopes.add(hint);
+          throw error;
+        }
+      }
+    },
+    next: async () => null,
+  });
+  return {
+    ready: Promise.all([retries.ready, queue.ready]),
+    wake() { scopes.add("*"); retries.wake(); queue.wake(); },
+    async stop() { await Promise.all([retries.stop(), queue.stop()]); },
+  };
+}

server/src/services/heartbeat/queue.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "../run-dispatch-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
 import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
 import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
@@ -1972,7 +1973,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
 
   // Release admission promptly without first scanning unrelated recovery work.
   // Every run still passes the ordinary per-agent claim and dispatch gates.
-  async function resumeQueuedRunsAfterDrain() {
+  async function resumeQueuedRunsAfterDrain(scope?: { companyId: string }) {
     if ((await getSchedulingSuppression()).suppressed) return;
     const cutoff = await getWorktreeExecutionCutoff();
     const queuedRuns = await db
@@ -1982,6 +1983,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
       .where(
         and(
           eq(heartbeatRuns.status, "queued"),
+          scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
           eq(companies.status, "active"),
           cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined,
         ),
@@ -4632,6 +4634,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
     }
 
     const queueOutcome = await db.transaction(async (tx) => {
+      await notifyRunDispatch(tx, { companyId: agent.companyId });
       await tx.execute(
         sql`select id from agents where id = ${agentId} and company_id = ${agent.companyId} for update`,
       );

server/src/services/heartbeat/retries.test.ts

@@ -35,6 +35,7 @@ function callbacks() {
     getWorktreeExecutionCutoff: vi.fn(async (): Promise<Date | null> => null),
     applyRunDispatchPostCommitEffects: vi.fn<(effects: PostCommitEffect[]) => void>(),
     runDispatch: {
+      nextScheduledRetryAt: vi.fn(async () => null),
       evaluateScheduledRetryGate: vi.fn<HeartbeatRetryDependencies["runDispatch"]["evaluateScheduledRetryGate"]>(async () => ({ allowed: true })),
       promoteDueScheduledRetries: vi.fn<HeartbeatRetryDependencies["runDispatch"]["promoteDueScheduledRetries"]>(async () => ({ promoted: 0, runIds: [], postCommitEffects: [] })),
       promoteScheduledRetry: vi.fn<HeartbeatRetryDependencies["runDispatch"]["promoteScheduledRetry"]>(async () => ({ outcome: "not_promoted" })),
@@ -183,7 +184,9 @@ describe("heartbeat retry module callbacks", () => {
   });
 });
 
-const support = await getEmbeddedPostgresTestSupport();
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HEARTBEAT_RETRIES_TEST_DATABASE_URL");
+const support = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
 const describePostgres = support.supported ? describe : describe.skip;
 if (!support.supported) console.warn(`Skipping retry PostgreSQL tests: ${support.reason ?? "unsupported host"}`);
 
@@ -193,11 +196,11 @@ describePostgres("heartbeat retry module database wiring", () => {
   const now = new Date("2026-10-08T12:00:00Z");
 
   beforeAll(async () => {
-    database = await startEmbeddedPostgresTestDatabase("heartbeat-retries-");
-    db = createDb(database.connectionString);
+    if (!external) database = await startEmbeddedPostgresTestDatabase("heartbeat-retries-");
+    db = createDb(external ?? database.connectionString, { maxConnections: 6 });
   }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
 
-  afterAll(async () => { await database?.cleanup(); });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
 
   afterEach(async () => {
     await db.delete(issues);

server/src/services/heartbeat/retries.ts

@@ -1,3 +1,4 @@
+import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
 import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js";
 import { infrastructureRetryDelayMs } from "../system-recovery-state.js";
 import {
@@ -77,7 +78,7 @@ type PlanApprovalResumeFailure = {
 export interface HeartbeatRetryDependencies {
   hasCaseHandoffForRun: (run: Pick<HeartbeatRun, "id" | "companyId">) => Promise<boolean>;
   runDispatch: Pick<ReturnType<typeof createRunDispatch>,
-    "evaluateScheduledRetryGate" | "promoteDueScheduledRetries" | "promoteScheduledRetry">;
+    "evaluateScheduledRetryGate" | "promoteDueScheduledRetries" | "promoteScheduledRetry" | "nextScheduledRetryAt">;
   getAgent: (agentId: string) => Promise<typeof agents.$inferSelect | null | undefined>;
   getRun: (
     runId: string,
@@ -997,6 +998,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
           retryContextSnapshot.previousRunId = continuation.previousRunId;
         }
 
+        await notifyScheduledRetryDispatch(tx, { companyId: run.companyId });
         const wakeupRequest = await tx
           .insert(agentWakeupRequests)
           .values({
@@ -1643,6 +1645,10 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
     });
   }
 
+  async function nextScheduledRetryAt() {
+    return runDispatch.nextScheduledRetryAt({ cutoff: await getWorktreeExecutionCutoff() });
+  }
+
   async function promoteDueScheduledRetries(now = new Date()) {
     const cutoff = await getWorktreeExecutionCutoff();
     const result = await runDispatch.promoteDueScheduledRetries({
@@ -1750,6 +1756,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
     };
 
     const updated = await db.transaction(async (tx) => {
+      await notifyScheduledRetryDispatch(tx, { companyId: scheduled.run.companyId });
       const row = await tx
         .update(heartbeatRuns)
         .set({
@@ -1901,6 +1908,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
     finalizeAiConnectionBusyDeferral,
     finalizeWorkspaceBusyDeferral,
     promoteDueScheduledRetries,
+    nextScheduledRetryAt,
     retryScheduledRetryNow,
 
     scheduleBoundedRetry: async (

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

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "../run-dispatch-notifications.js";
 import { preserveAcceptedRunUsageProjectionSql } from "../usage-receipts.js";
 import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
 import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
@@ -1044,7 +1045,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
       if (previous && isHeartbeatRunTerminalStatus(status))
         await notifyRetainedNativeSessionCleanup(tx, { companyId: previous.companyId, runId: previous.id });
       if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId);
-      await notifyExecutionRunWork(tx);
+      await notifyExecutionRunWork(tx, companyId ? { companyId } : undefined);
       if (previous) await notifyChatRunMilestones(tx, { companyId: previous.companyId, runId: previous.id,
         issueId: previous.nativeIssueId ?? previous.issueId ?? readNonEmptyString(parseObject(previous.contextSnapshot).issueId) ?? undefined });
       return write(tx);
@@ -2447,6 +2448,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
         .returning()
         .then((rows) => rows[0]);
 
+      await notifyRunDispatch(tx, { companyId: run.companyId });
       const queuedRun = await tx
         .insert(heartbeatRuns)
         .values({

server/src/services/heartbeat/runtime-execution.ts

@@ -983,7 +983,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
               eq(heartbeatRuns.id, run.id),
               eq(heartbeatRuns.status, "running"),
             ),
-          ));
+          ), { companyId: run.companyId });
         const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded";
         await recordWorkspaceFinalize(workspaceFinalizeStatus);
         if (adapterResult.nativeFinalization) {
@@ -1063,7 +1063,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
           eq(heartbeatRuns.id, run.id),
           eq(heartbeatRuns.status, "running"),
         ),
-      ));
+      ), { companyId: run.companyId });
     if (
       issueRef &&
       context.resumeSessionGoalHeartbeat === true &&

server/src/services/instance-settings.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifySkillVersionRetentionReconstruction } from "./skill-version-retention-notifications.js";
 import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js";
 import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js";
@@ -630,6 +631,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
         if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx);
         if (patch.enableWorktreeRunExecution !== undefined) {
+          await notifyRunDispatch(tx);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           await notifyExternalAgentSettlement(tx);
           await notifyInstructionCommit(tx);

server/src/services/issue-recovery-actions.ts

@@ -392,7 +392,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
           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(), { companyId: input.companyId });
           if (updated) return toReadModel(updated);
         }
         return existing;
@@ -467,7 +467,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
               ),
             )
             .returning();
-        });
+        }, { companyId: input.companyId });
         if (!exhausted) {
           return retryUpsertSourceScoped(input, retryCount);
         }
@@ -539,7 +539,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
             ),
           )
           .returning();
-      });
+      }, { companyId: input.companyId });
       if (!updated) {
         return retryUpsertSourceScoped(input, retryCount);
       }
@@ -554,7 +554,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
           .insert(issueRecoveryActions)
           .values(buildInsertValues(input, ownerType, now))
           .returning();
-      });
+      }, { companyId: input.companyId });
       return toReadModel(created!);
     } catch (error) {
       if (!isUniqueRecoveryActionConflict(error)) throw error;
@@ -608,7 +608,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
         })
         .where(and(...predicates))
         .returning();
-    });
+    }, { companyId: input.companyId });
 
     return updated ? toReadModel(updated) : null;
   }

server/src/services/issue-review-state-notifications.ts

@@ -1,9 +1,11 @@
+import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 
 export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) {
+  await notifyRunDispatch(tx, scope);
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
   await notifyExternalConversationSettlement(tx, scope);
   await notifyNativeReviewResponse(tx, scope);

server/src/services/issues.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
@@ -10606,6 +10607,7 @@ export function issueService(db: Db) {
             : values.parentId ? `issue:${values.parentId}` : undefined);
         }
         if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
+        await notifyRunDispatch(tx, { companyId: values.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         const [issue] = await tx.insert(issues).values(values).returning();
         await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId });
@@ -11494,6 +11496,7 @@ export function issueService(db: Db) {
         }
         if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined
           || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) {
+          await notifyRunDispatch(tx, { companyId: receiptExisting.companyId });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         }
         if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt &&
@@ -11958,6 +11961,7 @@ export function issueService(db: Db) {
           }
           await notifyDecisionTargetIssueRemoval(tx, id);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
+          await notifyRunDispatch(tx, owner ? { companyId: owner.companyId } : undefined);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           removedIssue = await tx
             .delete(issues)
@@ -12086,6 +12090,7 @@ export function issueService(db: Db) {
         : isNull(issues.executionRunId);
       const updated = await db.transaction(async tx => {
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
+        await notifyRunDispatch(tx, { companyId: issueCompany.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         return tx
           .update(issues)
@@ -12215,6 +12220,7 @@ export function issueService(db: Db) {
           const staleExecutionRunId = current.executionRunId;
           const adopted = await db.transaction(async tx => {
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
+            await notifyRunDispatch(tx, { companyId: issueCompany.companyId });
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
             return tx
               .update(issues)
@@ -12470,6 +12476,7 @@ export function issueService(db: Db) {
         await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
         await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
+        await notifyRunDispatch(tx, { companyId: existing.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         const updated = await tx
           .update(issues)
@@ -12530,6 +12537,7 @@ export function issueService(db: Db) {
           await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: existing.companyId, issueId: id });
         }
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
+        await notifyRunDispatch(tx, { companyId: existing.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         const updated = await tx
           .update(issues)

server/src/services/legacy-execution-recovery.ts

@@ -135,7 +135,7 @@ export async function terminalizeLegacyExecution(input: {
     };
     if (issueId) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: current.companyId, issueId });
     await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id });
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId: current.companyId });
     let [updated] = await tx
       .update(heartbeatRuns)
       .set({

server/src/services/native-runtime/native-failed-workspace-recovery.ts

@@ -65,7 +65,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeo
     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);
+      await notifyExecutionRunWork(tx, { companyId: run.companyId });
       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(),

server/src/services/native-runtime/native-provider-capacity-retry.ts

@@ -1,3 +1,4 @@
+import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
 import { randomUUID } from "node:crypto";
 import { and, eq } from "drizzle-orm";
 import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db";
@@ -49,6 +50,7 @@ export async function materializeNativeProviderCapacityRetry(input: {
     scheduledRetryAt: dueAt.toISOString(),
     executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt),
   };
+  await notifyScheduledRetryDispatch(input.tx, { companyId: run.companyId });
   const [wake] = await input.tx.insert(agentWakeupRequests).values({
     companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system",
     reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId,

server/src/services/native-runtime/native-restart-recovery.ts

@@ -924,7 +924,7 @@ export async function claimNativeRestartRecoveries(input: {
           recycledProviderPids: providerProcesses.recycledPids,
           stderrTail: row.run.stderrExcerpt,
         });
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: row.run.companyId });
         await tx
           .update(nativeRunFinalizations)
           .set({

server/src/services/native-runtime/native-run-finalizer.ts

@@ -401,7 +401,7 @@ async function recordRetryableFailure(input: {
     if (phase === "terminal_failure" && failureCode === "native_workspace_sync_out_unsafe_archive") {
       await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     }
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId: input.run.companyId });
     await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     await tx
       .update(nativeRunFinalizations)
@@ -1659,7 +1659,7 @@ export async function finalizeNativeRun(input: {
         if (input.projectRunStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
         if (input.projectRunStatus) await notifyUnsafeWorkspaceExportRunChange(tx, run);
         if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: run.companyId });
         await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId });
         const [after] = await projectRun(tx);
         return after ? { before, after } : null;

server/src/services/native-runtime/native-run-projection.ts

@@ -138,7 +138,7 @@ export async function projectCommittedRun(input: {
     await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id });
     await notifyUnsafeWorkspaceExportRunChange(tx, input.run);
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId: input.run.companyId });
     await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     const [after] = await tx
       .update(heartbeatRuns)

server/src/services/native-runtime/native-safe-replacement.test.ts

@@ -1,3 +1,4 @@
+import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { getExecutionBlocker } from "../execution-blocker.js";
 import { createRunDispatch, deriveCommentId } from "../../modules/run-dispatch/index.js";
 import { buildExecutionContinuation } from "../execution-continuation.js";
@@ -577,7 +578,12 @@ const support = externalDatabaseUrl
     });
     it("persists exactly one linked successor under competing sweepers and restarts", async () => {
       const source = await seed(2);
+      const owner = heartbeatService(db);
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
+      const worker = owner.registerRunDispatchWork(coordinator, () => true);
+      await worker.ready;
       const now = new Date();
+      try {
       await Promise.all([
         reconcileSafeNativeReplacements(db, now),
         reconcileSafeNativeReplacements(db, now),
@@ -608,6 +614,10 @@ const support = externalDatabaseUrl
             .where(eq(nativeRunFinalizations.runId, source.runId))
         )[0]?.failureDetail?.successorRunId,
       ).toBe(children[0]!.id);
+      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(children[0]!.scheduledRetryAt!.getTime()), { timeout: 5_000 });
+      } finally {
+        await worker.stop(); await coordinator.stop(); await owner.drainActiveRunExecutions();
+      }
     });
     it("retries status publication after a crash without dispatching provider work", async () => {
       await deliverExecutionStatuses(db);

server/src/services/native-runtime/native-safe-replacement.ts

@@ -1,3 +1,4 @@
+import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
 import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
 import { notifyExecutionRunWork } from "../execution-control-notifications.js";
@@ -267,7 +268,7 @@ export async function reconcileSafeNativeReplacements(
             current.failureDetail?.replacementDenied
           )
             return;
-          await notifyExecutionRunWork(tx);
+          await notifyExecutionRunWork(tx, { companyId: run.companyId });
           await tx
             .update(nativeRunFinalizations)
             .set({
@@ -409,7 +410,8 @@ export async function reconcileSafeNativeReplacements(
             idempotencyKey: `native-safe-replacement:${run.id}`,
           })
           .returning();
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: run.companyId });
+        await notifyScheduledRetryDispatch(tx, { companyId: run.companyId });
         await tx.insert(heartbeatRuns).values({
           id: successorRunId,
           companyId: run.companyId,

server/src/services/native-runtime/native-session-executor.ts

@@ -9902,7 +9902,7 @@ async function executePaperclipNativeSessionWithinScope(
             eq(nativeRunFinalizations.runId, input.execution.binding.runId),
             eq(nativeRunFinalizations.leaseOwner, leaseOwner),
           ),
-        ));
+        ), { companyId: input.execution.binding.companyId });
 
       const failedAtMs = Date.now();
       const executionFailureMessage = redactIdentityText(
@@ -9954,7 +9954,7 @@ async function executePaperclipNativeSessionWithinScope(
             eq(nativeRunFinalizations.leaseOwner, leaseOwner),
             eq(nativeRunFinalizations.attempt, attempt),
             isNull(nativeRunFinalizations.resultId),
-          )).returning({ runId: nativeRunFinalizations.runId }));
+          )).returning({ runId: nativeRunFinalizations.runId }), { companyId: input.execution.binding.companyId });
           if (settled) {
             await stoppedLeaseRenewal;
             if (warmSessionId !== null && lifecyclePolicy.mode === "warm") {
@@ -10041,7 +10041,7 @@ async function executePaperclipNativeSessionWithinScope(
             ),
           )
           .for("update");
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: input.execution.binding.companyId });
         await notifyNativeFinalizationWork(tx, input.execution.binding);
         const updated = await tx
           .update(nativeRunFinalizations)

server/src/services/native-runtime/native-superseded-retry.ts

@@ -39,7 +39,7 @@ export async function retireSupersededNativeRetry(db: Db, input: {
   )).limit(1);
   if (!source) return false;
 
-  await notifyExecutionRunWork(db);
+  await notifyExecutionRunWork(db, { companyId: run.companyId });
   await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", nextAttemptAt: null,
     leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: "blocked",
     failureCode: "native_retry_superseded", failureDetail: { ...coordinator.failureDetail,

server/src/services/recovery/service.ts

@@ -1,3 +1,4 @@
+import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
 import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
 import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
 import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
@@ -2716,6 +2717,7 @@ export function recoveryService(
     const now = new Date();
     const retryAt = readProviderQuotaRetryAt(input.latestRun, now);
     return database.transaction(async (tx) => {
+      await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId });
       const wakeup = await tx
         .insert(agentWakeupRequests)
         .values({
@@ -3316,6 +3318,7 @@ export function recoveryService(
           created = Boolean(enqueuedRun);
         } else {
           scheduledRun = await db.transaction(async (tx) => {
+            await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId });
             const wakeup = await tx
               .insert(agentWakeupRequests)
               .values({

server/src/services/run-dispatch-notifications.ts

@@ -0,0 +1,14 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type RunDispatchScope = { companyId: string };
+
+export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) {
+  const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint);
+}
+
+/** Only a retry row or deadline change needs to re-read the saved retry schedule. */
+export async function notifyScheduledRetryDispatch(transaction: object, scope?: RunDispatchScope) {
+  await notifyRunDispatch(transaction, scope);
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch);
+}

server/src/services/scheduled-retry-cancellation-notifications.ts

@@ -1,7 +1,9 @@
+import { notifyScheduledRetryDispatch } from "./run-dispatch-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
 export type ScheduledRetryCancellationScope = { companyId: string; runId?: string };
 
 export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) {
+  await notifyScheduledRetryDispatch(transaction, scope);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*");
 }

server/src/services/system-recovery.ts

@@ -71,7 +71,7 @@ export async function updateSystemRecovery(db: Db, source: WorkspaceRecoverySour
       sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,revision}' = ${String(previous.revision)}`))
       .returning({ id: issueRecoveryActions.id });
     return Boolean(updated);
-  });
+  }, { companyId: source.companyId });
 }
 
 export async function loadSystemRecovery(db: Db, source: WorkspaceRecoverySourceIdentity) {

server/src/services/task-retry.ts

@@ -334,7 +334,7 @@ export async function requestTaskRetry(input: {
       details: { requestId: active?.id ?? id, sourceRunId: latest.id, attachedToExistingWork: Boolean(active || busy) } });
     return { requestId: active?.id ?? id, accepted: true, stale: false, observedRunId: latest.id,
       chatActionId: intent.chatActionId };
-  });
+  }, { companyId: input.companyId });
 }
 
 /** Chat owns admission and authorization. The task receipt follows that