perf(heartbeat): schedule retry promotion from durable deadlines

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

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

@@ -0,0 +1,146 @@
+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, createDb, heartbeatRuns } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { heartbeatService } from "../services/heartbeat.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.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[] = [];
+  beforeAll(async () => {
+    if (!external) database = await startEmbeddedPostgresTestDatabase("run-dispatch-work-");
+    db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+  }, 90_000);
+  beforeEach(async () => {
+    await db.execute(sql`TRUNCATE companies CASCADE`);
+    heartbeat = heartbeatService(db); draining = false; errors.length = 0;
+    reads = vi.spyOn(db, "select");
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+  });
+  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.advanceTimersByTimeAsync(1);
+    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull(), { timeout: 5_000 });
+    const count = reads.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(20 * 60_000);
+    expect(reads).toHaveBeenCalledTimes(count);
+  }
+  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((await status(id))?.status).toBe("succeeded"), { timeout: 10_000 });
+    await heartbeat.drainActiveRunExecutions();
+  }
+  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("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: 60_000, random: () => 0 });
+    expect(result.outcome).toBe("scheduled");
+    if (result.outcome !== "scheduled") throw new Error("Expected retry creation");
+    await vi.advanceTimersByTimeAsync(1);
+    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(result.dueAt.getTime()));
+    expect((await status(result.run.id)).status).toBe("scheduled_retry");
+    await vi.advanceTimersByTimeAsync(result.dueAt.getTime() - Date.now() - 1);
+    expect((await status(result.run.id)).status).toBe("scheduled_retry");
+    await vi.advanceTimersByTimeAsync(1);
+    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.advanceTimersByTimeAsync(1);
+    await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(scheduled.dueAt.getTime()));
+    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: 10_000 });
+    if (scheduled.outcome !== "scheduled") throw new Error("Expected retry creation");
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect((await status(scheduled.run.id)).status).toBe("scheduled_retry");
+    draining = false; worker!.wake(); await vi.advanceTimersByTimeAsync(1);
+    await completed(scheduled.run.id); await quiet();
+  });
+  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 vi.advanceTimersByTimeAsync(1_000);
+    expect(await status(queuedId)).toBeUndefined();
+    release(); await writing; await vi.advanceTimersByTimeAsync(1); 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 vi.advanceTimersByTimeAsync(1); await quiet();
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(2);
+  });
+});

server/src/index.ts

@@ -1842,6 +1842,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;
 
@@ -2044,12 +2046,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 ||
@@ -2058,7 +2058,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,18 +80,19 @@ 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({
+      const updated = await effects.transaction(db, existing.companyId, async tx => {
+        await effects.beforeStatusChange(tx, existing.companyId, id, "error", "idle");
+        return 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()
-        .then((rows) => rows[0] ?? null);
+          .where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready")))
+          .returning()
+          .then((rows) => rows[0] ?? null);
+      });
 
       if (!updated) {
         throw conflict("Only agents in error status can have their error cleared");

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

@@ -511,6 +511,14 @@ 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"),
+        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()
@@ -1414,6 +1422,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);
       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);
       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);
   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);
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+      }
       if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) {
         await notifyNativeResponsePresentation(tx, { companyId, agentId });
         await notifyNativeReviewResponse(tx, { companyId, agentId });

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);
   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);
           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);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);

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,9 @@
+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) {
+  await notifyRunDispatch(transaction);
   for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
     DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
     await notifyDeliveryWork(transaction, queue);
@@ -14,6 +16,7 @@ export function writeExecutionControl<T>(
   write: (tx: Db) => PromiseLike<T>,
 ): Promise<T> {
   return db.transaction(async transaction => {
+    await notifyRunDispatch(transaction);
     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/heartbeat.ts

@@ -1,3 +1,4 @@
+import { registerRunDispatchWork } from "./heartbeat/dispatch-work.js";
 import { externalConversationSettlementWork } from "./external-conversation-settlement-work.js";
 import { nativeRecoveryNoticeWork } from "./native-runtime/native-recovery-notice-work.js";
 import { nativeRunProjectionWork } from "./native-runtime/native-run-projection-work.js";
@@ -999,6 +1000,7 @@ export function heartbeatService(
     finalizeAiConnectionBusyDeferral,
     finalizeWorkspaceBusyDeferral,
     promoteDueScheduledRetries,
+    nextScheduledRetryAt,
     retryScheduledRetryNow,
     scheduleBoundedRetry,
   } = createHeartbeatRetries(db, {
@@ -4246,6 +4248,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.test.ts

@@ -0,0 +1,105 @@
+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 "../delivery-work-coordinator.js";
+import { notifyRunDispatch } from "../run-dispatch-notifications.js";
+import { registerRunDispatchWork } from "./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 = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: 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 () => {});
+  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,
+    notify: () => db.transaction(tx => notifyRunDispatch(tx)),
+    settle: () => vi.advanceTimersByTimeAsync(0) };
+}
+
+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.notify(); await f.settle();
+    expect(f.coordinator.nextWakeAt()).toBe(1_005_000);
+    f.state.due = null; await f.notify(); 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("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);
+    expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).toHaveBeenCalledOnce();
+    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("retains failed work across drain and settles after recovery", async () => {
+    const f = fixture(); f.state.fail = true; await f.worker.ready;
+    f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000);
+    expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).not.toHaveBeenCalled();
+    f.state.fail = false; f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000);
+    expect(f.promote).toHaveBeenCalledTimes(2); 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.notify(); await f.settle();
+    await f.notify(); 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.notify(); await f.settle(); await f.worker.stop();
+    await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2);
+  });
+});

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

@@ -0,0 +1,39 @@
+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(): Promise<unknown>;
+    schedulingSuppressed(): Promise<boolean>;
+  },
+  canRun: () => boolean,
+) {
+  let suppressed = false;
+  const retries = registerDeadlineWork(coordinator, DELIVERY_QUEUES.scheduledRetryDispatch, {
+    retryMs: 30_000,
+    continuationMs: 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 queue = registerDeadlineWork(coordinator, DELIVERY_QUEUES.queuedRunDispatch, {
+    retryMs: 30_000,
+    canRun,
+    run: async signal => {
+      if (!signal.aborted && canRun()) await owner.resumeQueuedRunsAfterDrain();
+    },
+    next: async () => null,
+  });
+  return {
+    ready: Promise.all([retries.ready, queue.ready]),
+    wake() { 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";
@@ -4632,6 +4633,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
     }
 
     const queueOutcome = await db.transaction(async (tx) => {
+      await notifyRunDispatch(tx);
       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" })),

server/src/services/heartbeat/retries.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } 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 notifyRunDispatch(tx);
         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 notifyRunDispatch(tx);
       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";
@@ -2447,6 +2448,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
         .returning()
         .then((rows) => rows[0]);
 
+      await notifyRunDispatch(tx);
       const queuedRun = await tx
         .insert(heartbeatRuns)
         .values({

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-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);
   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";
@@ -10605,6 +10606,7 @@ export function issueService(db: Db) {
             : values.parentId ? `issue:${values.parentId}` : undefined);
         }
         if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
+        await notifyRunDispatch(tx);
         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 });
@@ -11493,6 +11495,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);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         }
         if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt &&
@@ -11957,6 +11960,7 @@ export function issueService(db: Db) {
           }
           await notifyDecisionTargetIssueRemoval(tx, id);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
+          await notifyRunDispatch(tx);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           removedIssue = await tx
             .delete(issues)
@@ -12085,6 +12089,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);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         return tx
           .update(issues)
@@ -12214,6 +12219,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);
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
             return tx
               .update(issues)
@@ -12469,6 +12475,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);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         const updated = await tx
           .update(issues)
@@ -12529,6 +12536,7 @@ export function issueService(db: Db) {
           await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: existing.companyId, issueId: id });
         }
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
+        await notifyRunDispatch(tx);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         const updated = await tx
           .update(issues)

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

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } 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 notifyRunDispatch(input.tx);
   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/recovery/service.ts

@@ -1,3 +1,4 @@
+import { notifyRunDispatch } 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 notifyRunDispatch(tx);
       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 notifyRunDispatch(tx);
             const wakeup = await tx
               .insert(agentWakeupRequests)
               .values({

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

@@ -0,0 +1,6 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export async function notifyRunDispatch(transaction: object) {
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch);
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch);
+}

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

@@ -1,7 +1,9 @@
+import { notifyRunDispatch } 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);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*");
 }