diff --git a/doc/plans/2026-10-30-run-dispatch-sleep.md b/doc/plans/2026-10-30-run-dispatch-sleep.md new file mode 100644 index 00000000000..82dd8692043 --- /dev/null +++ b/doc/plans/2026-10-30-run-dispatch-sleep.md @@ -0,0 +1,34 @@ +# 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. +An idle queue has no recurring read or timer-driven dispatch attempt. + +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. diff --git a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts new file mode 100644 index 00000000000..527309051fa --- /dev/null +++ b/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> | undefined; + let db: ReturnType; + let heartbeat: ReturnType; + let coordinator: ReturnType | undefined; + let worker: ReturnType["registerRunDispatchWork"]> | undefined; + let reads: ReturnType; + 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(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); + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index ae229f03518..f0eeddc6c91 100644 --- a/server/src/index.ts +++ b/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", ); } diff --git a/server/src/modules/agent-lifecycle/adapters/records.ts b/server/src/modules/agent-lifecycle/adapters/records.ts index 1c59758dd91..db842941836 100644 --- a/server/src/modules/agent-lifecycle/adapters/records.ts +++ b/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"); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index cf9946bafe9..63559e7701d 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/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 { + 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 { const rows = await db .select() @@ -1414,6 +1422,7 @@ export function createPostgresRunDispatchAdapter( return { evaluateScheduledRetryGate, + nextScheduledRetryAt, listDueRetries, cancelStaleQueuedRun, dispatchResolvedInteractionIfCurrent, diff --git a/server/src/modules/run-dispatch/application/ports.ts b/server/src/modules/run-dispatch/application/ports.ts index 83e41fd3da6..be6faf39ca7 100644 --- a/server/src/modules/run-dispatch/application/ports.ts +++ b/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; evaluateScheduledRetryGate(input: EvaluateScheduledRetryGateInput): Promise; listDueRetries(input: ListDueRetriesInput): Promise; } diff --git a/server/src/modules/run-dispatch/application/use-cases.test.ts b/server/src/modules/run-dispatch/application/use-cases.test.ts index 58e5ca6f8df..1cce6637ba0 100644 --- a/server/src/modules/run-dispatch/application/use-cases.test.ts +++ b/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 }; diff --git a/server/src/modules/run-dispatch/index.ts b/server/src/modules/run-dispatch/index.ts index 5293ca0ae64..d44d559618a 100644 --- a/server/src/modules/run-dispatch/index.ts +++ b/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({ diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index b888ee51fb3..c773b903dec 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/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({ diff --git a/server/src/services/agent-configuration.ts b/server/src/services/agent-configuration.ts index 4139dc6dfe0..88a2c636bdb 100644 --- a/server/src/services/agent-configuration.ts +++ b/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); } diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index 0a6dcaf0fc9..700c6ba8ffa 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/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 }); diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 73535f5434a..159055c10d3 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/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 }); diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 0f2ed547a79..cc10ffa3bb8 100644 --- a/server/src/services/companies.ts +++ b/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); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 9195f1fcb12..7388bdb8861 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/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", diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index 22c8913bbcd..6fef650fe42 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/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( write: (tx: Db) => PromiseLike, ): Promise { 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); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 670e4a2755b..9e3348ddace 100644 --- a/server/src/services/heartbeat.ts +++ b/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[0], canRun: () => boolean) => + registerRunDispatchWork(coordinator, { + promoteDueScheduledRetries, + nextScheduledRetryAt, + resumeQueuedRunsAfterDrain, + schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed, + }, () => canRun() && readTaskDrain(new Date()) === null), promoteDueScheduledRetries, retryScheduledRetryNow, diff --git a/server/src/services/heartbeat/dispatch-work.test.ts b/server/src/services/heartbeat/dispatch-work.test.ts new file mode 100644 index 00000000000..2ba36f38ef2 --- /dev/null +++ b/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> = []; +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) { + 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(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); + }); +}); diff --git a/server/src/services/heartbeat/dispatch-work.ts b/server/src/services/heartbeat/dispatch-work.ts new file mode 100644 index 00000000000..3291008dd9a --- /dev/null +++ b/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, + owner: { + promoteDueScheduledRetries(): Promise; + nextScheduledRetryAt(): Promise; + resumeQueuedRunsAfterDrain(): Promise; + schedulingSuppressed(): Promise; + }, + 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()]); }, + }; +} diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index a41b486b647..376530b2f0a 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/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`, ); diff --git a/server/src/services/heartbeat/retries.test.ts b/server/src/services/heartbeat/retries.test.ts index a557871c553..a6382081167 100644 --- a/server/src/services/heartbeat/retries.test.ts +++ b/server/src/services/heartbeat/retries.test.ts @@ -35,6 +35,7 @@ function callbacks() { getWorktreeExecutionCutoff: vi.fn(async (): Promise => null), applyRunDispatchPostCommitEffects: vi.fn<(effects: PostCommitEffect[]) => void>(), runDispatch: { + nextScheduledRetryAt: vi.fn(async () => null), evaluateScheduledRetryGate: vi.fn(async () => ({ allowed: true })), promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [], postCommitEffects: [] })), promoteScheduledRetry: vi.fn(async () => ({ outcome: "not_promoted" })), diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 946f5e309f8..5f4df5ed74e 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/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) => Promise; runDispatch: Pick, - "evaluateScheduledRetryGate" | "promoteDueScheduledRetries" | "promoteScheduledRetry">; + "evaluateScheduledRetryGate" | "promoteDueScheduledRetries" | "promoteScheduledRetry" | "nextScheduledRetryAt">; getAgent: (agentId: string) => Promise; 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 ( diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 00a27c1faf2..07bb54dfdf7 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/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({ diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 0208653a86f..174c39d6125 100644 --- a/server/src/services/instance-settings.ts +++ b/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); diff --git a/server/src/services/issue-review-state-notifications.ts b/server/src/services/issue-review-state-notifications.ts index 679b7e37475..b769be64056 100644 --- a/server/src/services/issue-review-state-notifications.ts +++ b/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); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 42c65bb5d02..a7c8bb55023 100644 --- a/server/src/services/issues.ts +++ b/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) diff --git a/server/src/services/native-runtime/native-provider-capacity-retry.ts b/server/src/services/native-runtime/native-provider-capacity-retry.ts index 5b54bd0cdfc..7d007c383dd 100644 --- a/server/src/services/native-runtime/native-provider-capacity-retry.ts +++ b/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, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 991fa8f63b5..4393e0ad5cf 100644 --- a/server/src/services/recovery/service.ts +++ b/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({ diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts new file mode 100644 index 00000000000..9b1bb175894 --- /dev/null +++ b/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); +} diff --git a/server/src/services/scheduled-retry-cancellation-notifications.ts b/server/src/services/scheduled-retry-cancellation-notifications.ts index 8a8b4fe1a2f..c5548996607 100644 --- a/server/src/services/scheduled-retry-cancellation-notifications.ts +++ b/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) : "*"); }