fix: wake retry scheduling only when retry state changes

Separate retry-schedule notifications from ordinary queued-run eligibility. Preserve exact deadlines and prove native producer commit/rollback behavior without global retry scans on ordinary agent or issue edits.

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

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

@@ -12,6 +12,15 @@ 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

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

@@ -3,7 +3,7 @@ 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 } from "../services/run-dispatch-notifications.js";
+import { notifyRunDispatch, notifyScheduledRetryDispatch } from "../services/run-dispatch-notifications.js";
 import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js";
 
 const cleanup: Array<() => Promise<void>> = [];
@@ -38,6 +38,7 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean
   }, () => !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) };
 }
@@ -61,9 +62,9 @@ describe("run dispatch obligations", () => {
   });
   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.notify(); await f.settle();
+    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.notify(); await f.settle();
+    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);
@@ -76,7 +77,7 @@ describe("run dispatch obligations", () => {
       await f.notify(source);
       await vi.advanceTimersByTimeAsync(10);
     }
-    expect(f.promote.mock.calls.length).toBeLessThanOrEqual(2);
+    expect(f.promote).toHaveBeenCalledOnce();
     expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2);
     await f.settle();
     expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" });
@@ -87,7 +88,7 @@ describe("run dispatch obligations", () => {
   });
   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.notify(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000);
+    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);
@@ -96,7 +97,7 @@ describe("run dispatch obligations", () => {
   });
   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.notify(); await f.settle();
+    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();
@@ -109,8 +110,8 @@ describe("run dispatch obligations", () => {
     let release!: () => void;
     const held = new Promise<void>(resolve => { release = resolve; });
     f.promote.mockImplementationOnce(() => held);
-    await f.notify(); await f.settle();
-    await f.notify(); await f.settle();
+    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);
@@ -121,7 +122,7 @@ describe("run dispatch obligations", () => {
     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.notify(); await f.settle(); await f.worker.stop();
+    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

@@ -2,9 +2,11 @@ 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 } from "@paperclipai/db";
+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";
@@ -24,9 +26,14 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
   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 () => {
@@ -88,6 +95,45 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
     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 });

server/src/services/heartbeat/retries.ts

@@ -1,4 +1,4 @@
-import { notifyRunDispatch } from "../run-dispatch-notifications.js";
+import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
 import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js";
 import { infrastructureRetryDelayMs } from "../system-recovery-state.js";
 import {
@@ -998,7 +998,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
           retryContextSnapshot.previousRunId = continuation.previousRunId;
         }
 
-        await notifyRunDispatch(tx, { companyId: run.companyId });
+        await notifyScheduledRetryDispatch(tx, { companyId: run.companyId });
         const wakeupRequest = await tx
           .insert(agentWakeupRequests)
           .values({
@@ -1756,7 +1756,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
     };
 
     const updated = await db.transaction(async (tx) => {
-      await notifyRunDispatch(tx, { companyId: scheduled.run.companyId });
+      await notifyScheduledRetryDispatch(tx, { companyId: scheduled.run.companyId });
       const row = await tx
         .update(heartbeatRuns)
         .set({

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

@@ -1,4 +1,4 @@
-import { notifyRunDispatch } from "../run-dispatch-notifications.js";
+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";
@@ -50,7 +50,7 @@ export async function materializeNativeProviderCapacityRetry(input: {
     scheduledRetryAt: dueAt.toISOString(),
     executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt),
   };
-  await notifyRunDispatch(input.tx, { companyId: run.companyId });
+  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-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";
@@ -410,6 +411,7 @@ export async function reconcileSafeNativeReplacements(
           })
           .returning();
         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/recovery/service.ts

@@ -1,4 +1,4 @@
-import { notifyRunDispatch } from "../run-dispatch-notifications.js";
+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";
@@ -2717,7 +2717,7 @@ export function recoveryService(
     const now = new Date();
     const retryAt = readProviderQuotaRetryAt(input.latestRun, now);
     return database.transaction(async (tx) => {
-      await notifyRunDispatch(tx, { companyId: input.issue.companyId });
+      await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId });
       const wakeup = await tx
         .insert(agentWakeupRequests)
         .values({
@@ -3318,7 +3318,7 @@ export function recoveryService(
           created = Boolean(enqueuedRun);
         } else {
           scheduledRun = await db.transaction(async (tx) => {
-            await notifyRunDispatch(tx, { companyId: input.issue.companyId });
+            await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId });
             const wakeup = await tx
               .insert(agentWakeupRequests)
               .values({

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

@@ -4,6 +4,11 @@ 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.scheduledRetryDispatch, hint);
   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,9 +1,9 @@
-import { notifyRunDispatch } from "./run-dispatch-notifications.js";
+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 notifyRunDispatch(transaction, scope);
+  await notifyScheduledRetryDispatch(transaction, scope);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*");
 }