From d820f2bae4a8c00e1361b6504df2c7018639205d Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 16:53:01 -0500 Subject: [PATCH] refactor(recovery): expose scoped issue recovery obligations Co-Authored-By: Paperclip --- .../modules/wake-queue/adapters/postgres.ts | 4 +- server/src/services/agent-configuration.ts | 2 +- server/src/services/agent-lifecycle.ts | 2 +- server/src/services/budgets.ts | 8 +-- .../chat-run-milestone-notifications.ts | 2 +- server/src/services/companies.ts | 4 +- .../services/delivery-work-notifications.ts | 4 ++ server/src/services/heartbeat.ts | 27 +++++---- server/src/services/heartbeat/queue.ts | 2 +- server/src/services/heartbeat/retries.ts | 4 +- .../src/services/heartbeat/run-lifecycle.ts | 2 +- server/src/services/issue-comment-writes.ts | 2 + .../services/issue-recovery-notifications.ts | 11 ++++ .../issue-review-state-notifications.ts | 2 +- server/src/services/issues.ts | 18 +++--- .../src/services/legacy-controller-lease.ts | 12 ++-- .../native-provider-capacity-retry.ts | 2 +- server/src/services/recovery/service.ts | 41 ++++++++++--- server/src/services/recovery/work.ts | 58 +++++++++++++++++++ .../services/run-dispatch-notifications.ts | 4 +- ...eduled-retry-cancellation-notifications.ts | 2 +- server/src/services/task-watchdog-work.ts | 36 ++++++++++++ server/src/services/task-watchdogs.ts | 46 +++++++++++---- 23 files changed, 233 insertions(+), 62 deletions(-) create mode 100644 server/src/services/issue-recovery-notifications.ts create mode 100644 server/src/services/recovery/work.ts create mode 100644 server/src/services/task-watchdog-work.ts diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index c773b903dec..1df4420615f 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -638,7 +638,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru .returning() .then((rows) => rows[0]); - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId }); const queuedRun = await tx .insert(heartbeatRuns) .values({ @@ -717,7 +717,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru .returning() .then((rows) => rows[0]); - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId }); const queuedRun = await tx .insert(heartbeatRuns) .values({ diff --git a/server/src/services/agent-configuration.ts b/server/src/services/agent-configuration.ts index 88a2c636bdb..dc10efaa082 100644 --- a/server/src/services/agent-configuration.ts +++ b/server/src/services/agent-configuration.ts @@ -84,7 +84,7 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura existing.adapterConfig, ); } - await notifyRunDispatch(db); + await notifyRunDispatch(db, { companyId: existing.companyId }); 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 700c6ba8ffa..59357fcc5de 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -40,7 +40,7 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle), async beforeStatusChange(tx, companyId, agentId, before, after) { if (before !== after) { - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); } if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) { diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 86d00c4d108..186253b19bb 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -288,7 +288,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica } if (policy.scopeType === "project") { - await notifyRunDispatch(db); + await notifyRunDispatch(db, { companyId: policy.companyId }); await db .update(projects) .set({ @@ -302,7 +302,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; - await notifyRunDispatch(db); + await notifyRunDispatch(db, { companyId: policy.companyId }); await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); @@ -349,7 +349,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica } if (policy.scopeType === "project") { - await notifyRunDispatch(db); + await notifyRunDispatch(db, { companyId: policy.companyId }); await db .update(projects) .set({ @@ -364,7 +364,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; - await notifyRunDispatch(db); + await notifyRunDispatch(db, { companyId: policy.companyId }); await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 159055c10d3..524936395b5 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -12,7 +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 notifyRunDispatch(tx, scope); 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 cc10ffa3bb8..a9a8f59998c 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -333,7 +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 notifyRunDispatch(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); @@ -490,7 +490,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { const wasAlreadyArchived = existing.status === "archived"; if (!wasAlreadyArchived) { - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId: id }); 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 7388bdb8861..1d9ac79f520 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,10 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + strandedAssignments: "stranded-assignments", + resolvedDependencies: "resolved-dependencies", + taskWatchdogs: "task-watchdogs", + staleIssueLocks: "stale-issue-locks", scheduledRetryDispatch: "scheduled-retry-dispatch", queuedRunDispatch: "queued-run-dispatch", runEvidenceMigration: "run-evidence-migration", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 0772b920023..8c73880ce57 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerIssueRecoveryWork } from "./recovery/work.js"; 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"; @@ -1611,15 +1612,15 @@ export function heartbeatService( void reportRunFailure(db, updated, failureReport); } - async function reconcileStrandedAssignedIssues(opts?: { skipExternalConversationSettlement?: boolean }) { + async function reconcileStrandedAssignedIssues(opts?: Parameters[0]) { return recovery.reconcileStrandedAssignedIssues({ - skipExternalConversationSettlement: opts?.skipExternalConversationSettlement, + ...opts, issueCreatedAtGte: await getWorktreeExecutionCutoff(), }); } - async function sweepStaleIssueLocks() { - return recovery.sweepStaleIssueLocks(); + async function sweepStaleIssueLocks(opts?: Parameters[0]) { + return recovery.sweepStaleIssueLocks(opts); } function issueIdFromRunContext(contextSnapshot: unknown) { @@ -1649,10 +1650,7 @@ export function heartbeatService( }); } - async function reconcileTaskWatchdogs(opts?: { - companyId?: string | null; - runId?: string | null; - }) { + async function reconcileTaskWatchdogs(opts?: Parameters[0]) { return taskWatchdogs.reconcileTaskWatchdogs({ ...opts, issueCreatedAtGte: await getWorktreeExecutionCutoff(), @@ -1677,10 +1675,7 @@ export function heartbeatService( return recovery.buildRunOutputSilence(run, now); } - async function reconcileResolvedDependencyWakes(opts?: { - runId?: string | null; - companyId?: string | null; - }) { + async function reconcileResolvedDependencyWakes(opts?: Parameters[0]) { return recovery.reconcileResolvedDependencyWakeBackstop(opts); } @@ -4248,6 +4243,14 @@ export function heartbeatService( computeTaskDrain, applyTaskDrain, + registerIssueStateRecoveryWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerIssueRecoveryWork(db, coordinator, { + stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true }), + dependencies: reconcileResolvedDependencyWakes, + locks: sweepStaleIssueLocks, + watchdogs: reconcileTaskWatchdogs, + schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed, + }, () => canRun() && readTaskDrain(new Date()) === null), registerRunDispatchWork: (coordinator: Parameters[0], canRun: () => boolean) => registerRunDispatchWork(coordinator, { promoteDueScheduledRetries, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 376530b2f0a..9b08eece201 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -4633,7 +4633,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende } const queueOutcome = await db.transaction(async (tx) => { - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId: agent.companyId }); await tx.execute( sql`select id from agents where id = ${agentId} and company_id = ${agent.companyId} for update`, ); diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 5f4df5ed74e..32333c0ebec 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -998,7 +998,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen retryContextSnapshot.previousRunId = continuation.previousRunId; } - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId: run.companyId }); const wakeupRequest = await tx .insert(agentWakeupRequests) .values({ @@ -1756,7 +1756,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen }; const updated = await db.transaction(async (tx) => { - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId: scheduled.run.companyId }); const row = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 07bb54dfdf7..e44906e72fb 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -2448,7 +2448,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl .returning() .then((rows) => rows[0]); - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId: run.companyId }); const queuedRun = await tx .insert(heartbeatRuns) .values({ diff --git a/server/src/services/issue-comment-writes.ts b/server/src/services/issue-comment-writes.ts index e24e78e9855..a968ecb1fec 100644 --- a/server/src/services/issue-comment-writes.ts +++ b/server/src/services/issue-comment-writes.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import type { PgInsertValue } from "drizzle-orm/pg-core"; import { issueComments, type Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -6,6 +7,7 @@ type WriteDb = Db | Parameters[0]>[0]; export function insertIssueComment(db: WriteDb, input: PgInsertValue & { companyId: string }) { return db.transaction(async tx => { + await notifyIssueRecovery(tx, { companyId: input.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`); return tx.insert(issueComments).values(input).returning(); }); diff --git a/server/src/services/issue-recovery-notifications.ts b/server/src/services/issue-recovery-notifications.ts new file mode 100644 index 00000000000..db04ca801ba --- /dev/null +++ b/server/src/services/issue-recovery-notifications.ts @@ -0,0 +1,11 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type IssueRecoveryScope = { companyId: string }; + +export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) { + const hint = scope ? JSON.stringify(scope) : "*"; + for (const queue of [DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies, + DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) { + await notifyDeliveryWork(transaction, queue, hint); + } +} diff --git a/server/src/services/issue-review-state-notifications.ts b/server/src/services/issue-review-state-notifications.ts index b769be64056..d9d4d756816 100644 --- a/server/src/services/issue-review-state-notifications.ts +++ b/server/src/services/issue-review-state-notifications.ts @@ -5,7 +5,7 @@ import { notifyNativeReviewResponse } from "./native-review-response-notificatio 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 notifyRunDispatch(tx, scope); 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 a7c8bb55023..575648ccc37 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -10606,7 +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 notifyRunDispatch(tx, { companyId: values.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const [issue] = await tx.insert(issues).values(values).returning(); await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId }); @@ -11494,8 +11494,10 @@ export function issueService(db: Db) { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); } if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined - || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) { - await notifyRunDispatch(tx); + || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined || patch.parentId !== undefined + || patch.executionState !== undefined || patch.executionRunId !== undefined || patch.checkoutRunId !== undefined + || blockedByIssueIds !== undefined) { + await notifyRunDispatch(tx, { companyId: receiptExisting.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); } if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt && @@ -11960,7 +11962,7 @@ export function issueService(db: Db) { } await notifyDecisionTargetIssueRemoval(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, owner ? { companyId: owner.companyId } : undefined); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); removedIssue = await tx .delete(issues) @@ -12089,7 +12091,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 notifyRunDispatch(tx, { companyId: issueCompany.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx .update(issues) @@ -12219,7 +12221,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 notifyRunDispatch(tx, { companyId: issueCompany.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx .update(issues) @@ -12475,7 +12477,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 notifyRunDispatch(tx, { companyId: existing.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const updated = await tx .update(issues) @@ -12536,7 +12538,7 @@ export function issueService(db: Db) { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: existing.companyId, issueId: id }); } await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId: existing.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const updated = await tx .update(issues) diff --git a/server/src/services/legacy-controller-lease.ts b/server/src/services/legacy-controller-lease.ts index 55a8439debc..22fa078f459 100644 --- a/server/src/services/legacy-controller-lease.ts +++ b/server/src/services/legacy-controller-lease.ts @@ -77,14 +77,18 @@ async function renewOnLivenessPool( }); } -export async function hasLiveLegacyController(db: Db, run: Run): Promise { - if (run.runtimeMode === "native" || !run.controllerBootId) return false; - const [owner] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( +export async function liveLegacyControllerExpiresAt(db: Db, run: Run): Promise { + if (run.runtimeMode === "native" || !run.controllerBootId) return null; + const [owner] = await db.select({ expiresAt: heartbeatRuns.controllerLeaseExpiresAt }).from(heartbeatRuns).where(and( eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.status, "running"), gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`), )); - return Boolean(owner); + return owner?.expiresAt ?? null; +} + +export async function hasLiveLegacyController(db: Db, run: Run): Promise { + return (await liveLegacyControllerExpiresAt(db, run)) !== null; } export type RevokedLegacyControllerState = Pick; 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 7d007c383dd..83314ff9c51 100644 --- a/server/src/services/native-runtime/native-provider-capacity-retry.ts +++ b/server/src/services/native-runtime/native-provider-capacity-retry.ts @@ -50,7 +50,7 @@ export async function materializeNativeProviderCapacityRetry(input: { scheduledRetryAt: dueAt.toISOString(), executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt), }; - await notifyRunDispatch(input.tx); + await notifyRunDispatch(input.tx, { companyId: run.companyId }); const [wake] = await input.tx.insert(agentWakeupRequests).values({ companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system", reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 4393e0ad5cf..7888b3e23fb 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -27,7 +27,7 @@ import { decideLegacyContinuation, legacyDispositionEpisode, legacyDispositionFingerprint, LEGACY_DISPOSITION_REPAIR_INSTRUCTION, type LegacyDispositionEpisode, } from "./legacy-continuation.js"; -import { CONTROLLER_LEASE_LOST_FAILURE_CODE, hasLiveLegacyController } from "../legacy-controller-lease.js"; +import { CONTROLLER_LEASE_LOST_FAILURE_CODE, hasLiveLegacyController, liveLegacyControllerExpiresAt } from "../legacy-controller-lease.js"; import { isCompletedOnboardingHandoffWake } from "../chat-completion-delivery.js"; import { instanceSettingsService } from "../instance-settings.js"; import { isWaitingConversation, settleConversationTurnForRun, deliverConversationComments } from "../agent-conversations.js"; @@ -245,6 +245,7 @@ type ResolvedDependencyWakeBackstopOptions = { runId?: string | null; companyId?: string | null; blockerIssueId?: string | null; + issueId?: string; source?: ResolvedDependencyWakeBackstopSource; }; @@ -2717,7 +2718,7 @@ export function recoveryService( const now = new Date(); const retryAt = readProviderQuotaRetryAt(input.latestRun, now); return database.transaction(async (tx) => { - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ @@ -3318,7 +3319,7 @@ export function recoveryService( created = Boolean(enqueuedRun); } else { scheduledRun = await db.transaction(async (tx) => { - await notifyRunDispatch(tx); + await notifyRunDispatch(tx, { companyId: input.issue.companyId }); const wakeup = await tx .insert(agentWakeupRequests) .values({ @@ -4526,6 +4527,9 @@ export function recoveryService( } async function reconcileStrandedAssignedIssues(opts?: { + companyId?: string; + issueId?: string; + onDeferredDeadline?: (at: number) => void; issueCreatedAtGte?: Date | null; skipExternalConversationSettlement?: boolean; }) { @@ -4534,6 +4538,8 @@ export function recoveryService( .from(issues) .where( and( + opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined, + opts?.issueId ? eq(issues.id, opts.issueId) : undefined, isNull(issues.assigneeUserId), inArray(issues.status, ["todo", "in_progress", "in_review"]), or( @@ -5714,6 +5720,7 @@ export function recoveryService( classification.baseBackoffMs * Math.pow(2, Math.max(0, consecutive - 1)); if (elapsed < requiredDelay) { + opts?.onDeferredDeadline?.(latestFinishedAt.getTime() + requiredDelay); result.skipped += 1; continue; } @@ -5815,7 +5822,7 @@ export function recoveryService( source === "workspace.finalize" ? "workspace_finalize_reconciliation" : "issue_graph_liveness_reconciliation"; - const useCursor = !opts?.blockerIssueId; + const useCursor = !opts?.blockerIssueId && !opts?.issueId; const queryCandidates = (afterIssueId: string | null) => { const filters = [ @@ -5825,6 +5832,7 @@ export function recoveryService( sql`${issues.assigneeAgentId} is not null`, ]; if (opts?.companyId) filters.push(eq(issues.companyId, opts.companyId)); + if (opts?.issueId) filters.push(eq(issues.id, opts.issueId)); if (afterIssueId) filters.push(gt(issues.id, afterIssueId)); if (opts?.blockerIssueId) { @@ -6115,6 +6123,7 @@ export function recoveryService( // fallback below. Without it, a terminal issue named in the run context // snapshot would still terminalize the shared run and defeat the guard. runReferencedByActiveIssue?: boolean; + onDeferredDeadline?: (at: number) => void; }, ): Promise<{ terminalized: boolean; status: string }> { // Act only on a run in "running" status. A "queued" run has no process yet, @@ -6130,8 +6139,11 @@ export function recoveryService( // Another controller may own a sandbox run whose PID has no meaning on // this host. Its live lease owns both execution and finalization, even if // the issue is already terminal or this process has no in-memory handle. - if (await hasLiveLegacyController(db, run)) + const controllerDeadline = await liveLegacyControllerExpiresAt(db, run); + if (controllerDeadline) { + options?.onDeferredDeadline?.(controllerDeadline.getTime()); return { terminalized: false, status: run.status }; + } const pid = run.processPid ?? null; const processGroupId = run.processGroupId ?? null; @@ -6352,7 +6364,7 @@ export function recoveryService( // still claims to be live but can no longer reach a terminal status on its // own, so a stuck "running" run can no longer block the sweep. Idempotent and // safe: clears at most one row's worth of lock columns per candidate. - async function sweepStaleIssueLocks() { + async function sweepStaleIssueLocks(opts?: { companyId?: string; issueId?: string; onDeferredDeadline?: (at: number) => void }) { const result = { cleared: 0, issueIds: [] as string[], @@ -6369,7 +6381,9 @@ export function recoveryService( }) .from(issues) .where( - sql`(${issues.checkoutRunId} is not null or ${issues.executionRunId} is not null)`, + and(sql`(${issues.checkoutRunId} is not null or ${issues.executionRunId} is not null)`, + opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined, + opts?.issueId ? eq(issues.id, opts.issueId) : undefined), ); const referencedRunIds = [ @@ -6384,7 +6398,8 @@ export function recoveryService( ? await db .select() .from(heartbeatRuns) - .where(inArray(heartbeatRuns.id, referencedRunIds)) + .where(and(inArray(heartbeatRuns.id, referencedRunIds), + opts?.companyId ? eq(heartbeatRuns.companyId, opts.companyId) : undefined)) : []; const runStatusById = new Map(); for (const row of runRows) runStatusById.set(row.id, row.status); @@ -6395,8 +6410,15 @@ export function recoveryService( // the same run id in a stale lock column. The terminal reference alone must // not terminalize a run that an active issue still owns, so exclude these // runs from the issue-terminal authority below. + const ownershipReferences = opts?.issueId && referencedRunIds.length > 0 + ? await db.select({ companyId: issues.companyId, status: issues.status, + checkoutRunId: issues.checkoutRunId, executionRunId: issues.executionRunId }).from(issues).where(and( + opts.companyId ? eq(issues.companyId, opts.companyId) : undefined, + or(inArray(issues.checkoutRunId, referencedRunIds), inArray(issues.executionRunId, referencedRunIds)), + )) + : candidates; const runIdsReferencedByActiveIssue = new Set(); - for (const issue of candidates) { + for (const issue of ownershipReferences) { if (issue.status === "done" || issue.status === "cancelled") continue; for (const runId of [issue.checkoutRunId, issue.executionRunId]) { if (runId) runIdsReferencedByActiveIssue.add(runId); @@ -6437,6 +6459,7 @@ export function recoveryService( referencingIssueTerminalStatus: issueTerminalStatusByRunId.get(row.id) ?? null, runReferencedByActiveIssue: runIdsReferencedByActiveIssue.has(row.id), + onDeferredDeadline: opts?.onDeferredDeadline, }); runStatusById.set(row.id, outcome.status); if (outcome.terminalized) result.terminalizedRunIds.push(row.id); diff --git a/server/src/services/recovery/work.ts b/server/src/services/recovery/work.ts new file mode 100644 index 00000000000..93679e80423 --- /dev/null +++ b/server/src/services/recovery/work.ts @@ -0,0 +1,58 @@ +import { registerTaskWatchdogWork } from "../task-watchdog-work.js"; +import { and, eq, inArray, isNotNull, isNull, or } from "drizzle-orm"; +import { issues, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, type DeliveryQueue } from "../delivery-work-notifications.js"; +import type { IssueRecoveryScope } from "../issue-recovery-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; + +type Entry = { id: string; companyId: string; dueAt: number }; +type Scope = { companyId: string; issueId: string; onDeferredDeadline: (at: number) => void }; + +export function registerIssueRecoveryWork( + db: Db, + coordinator: ReturnType, + owner: { + stranded(scope: Scope): Promise; + dependencies(scope: Scope): Promise<{ enqueueFailed: number }>; + locks(scope: Scope): Promise; + watchdogs(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>; + schedulingSuppressed(): Promise; + }, + canRun: () => boolean, +) { + function register(queue: DeliveryQueue, statuses: string[] | null, check: (scope: Scope) => Promise) { + return registerScopedDeadlineWork(coordinator, { + queue, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, + matches: (entry, scope) => !scope || entry.companyId === scope.companyId, + load: async scope => { + const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and( + scope ? eq(issues.companyId, scope.companyId) : undefined, + statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)), + statuses ? isNull(issues.hiddenAt) : undefined, + )); + return rows.map(row => ({ ...row, dueAt: Date.now() })); + }, + check: async entry => { + if (await owner.schedulingSuppressed()) return null; + let next: number | null = null; + await check({ companyId: entry.companyId, issueId: entry.id, + onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } }); + return next === null ? null : { ...entry, dueAt: next }; + }, + }, canRun); + } + const assignments = register(DELIVERY_QUEUES.strandedAssignments, ["todo", "in_progress", "in_review"], owner.stranded); + const dependencies = register(DELIVERY_QUEUES.resolvedDependencies, ["blocked"], async scope => { + const result = await owner.dependencies(scope); + if (result.enqueueFailed > 0) throw new Error("Resolved dependency wake remains undelivered"); + }); + const locks = register(DELIVERY_QUEUES.staleIssueLocks, null, owner.locks); + const watchdogs = registerTaskWatchdogWork(db, coordinator, { + reconcile: owner.watchdogs, enabled: async () => !(await owner.schedulingSuppressed()), + }, canRun); + return { ready: Promise.all([assignments.ready, dependencies.ready, locks.ready, watchdogs.ready]), + wake() { assignments.invalidate(); dependencies.invalidate(); locks.invalidate(); watchdogs.invalidate(); }, + async stop() { await Promise.all([assignments.stop(), dependencies.stop(), locks.stop(), watchdogs.stop()]); } }; +} diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index 9b1bb175894..a7a3382872a 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -1,6 +1,8 @@ +import { notifyIssueRecovery, type IssueRecoveryScope } from "./issue-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export async function notifyRunDispatch(transaction: object) { +export async function notifyRunDispatch(transaction: object, scope?: IssueRecoveryScope) { + await notifyIssueRecovery(transaction, scope); 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 c5548996607..24e57de66cb 100644 --- a/server/src/services/scheduled-retry-cancellation-notifications.ts +++ b/server/src/services/scheduled-retry-cancellation-notifications.ts @@ -4,6 +4,6 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type ScheduledRetryCancellationScope = { companyId: string; runId?: string }; export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) { - await notifyRunDispatch(transaction); + await notifyRunDispatch(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*"); } diff --git a/server/src/services/task-watchdog-work.ts b/server/src/services/task-watchdog-work.ts new file mode 100644 index 00000000000..091c8c6595f --- /dev/null +++ b/server/src/services/task-watchdog-work.ts @@ -0,0 +1,36 @@ +import { and, eq } from "drizzle-orm"; +import { issueWatchdogs, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { IssueRecoveryScope } from "./issue-recovery-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; + +type Entry = { id: string; companyId: string; dueAt: number }; + +export function registerTaskWatchdogWork( + db: Db, + coordinator: ReturnType, + owner: { + enabled(): Promise; + reconcile(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>; + }, + canRun: () => boolean, +) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.taskWatchdogs, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, + matches: (entry, scope) => !scope || entry.companyId === scope.companyId, + load: async scope => (await db.select({ id: issueWatchdogs.id, companyId: issueWatchdogs.companyId }) + .from(issueWatchdogs).where(and(eq(issueWatchdogs.status, "active"), + scope ? eq(issueWatchdogs.companyId, scope.companyId) : undefined))) + .map(row => ({ ...row, dueAt: Date.now() })), + check: async entry => { + if (!(await owner.enabled())) return null; + let next: number | null = null; + const result = await owner.reconcile({ companyId: entry.companyId, watchdogId: entry.id, + onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } }); + if (result.incomplete > 0) throw new Error("Task watchdog subtree remains incomplete"); + return next === null ? null : { ...entry, dueAt: next }; + }, + }, canRun); +} diff --git a/server/src/services/task-watchdogs.ts b/server/src/services/task-watchdogs.ts index cf8d92d39df..6f94e5520a7 100644 --- a/server/src/services/task-watchdogs.ts +++ b/server/src/services/task-watchdogs.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { createHash } from "node:crypto"; import { and, asc, eq, gte, inArray, isNull, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -831,7 +832,7 @@ async function updateIssueWatchdogRow( return updated; } -export async function upsertIssueWatchdogForIssue( +async function upsertIssueWatchdogInTransaction( dbOrTx: any, companyId: string, issueId: string, @@ -852,7 +853,7 @@ export async function upsertIssueWatchdogForIssue( return { watchdog: toIssueWatchdog(updated), created: false }; } - const insertResult: { row: IssueWatchdogRow; created: boolean } = await dbOrTx + const insertResult: { row: IssueWatchdogRow; created: boolean } = await dbOrTx.transaction(async (tx: Db) => tx .insert(issueWatchdogs) .values({ companyId, @@ -870,7 +871,7 @@ export async function upsertIssueWatchdogForIssue( updatedAt: now, }) .returning() - .then((rows: IssueWatchdogRow[]) => ({ row: rows[0], created: true })) + .then((rows: IssueWatchdogRow[]) => ({ row: rows[0], created: true }))) .catch(async (error: unknown) => { if (!isIssueWatchdogUniqueConflict(error)) throw error; const winner = await dbOrTx @@ -885,6 +886,18 @@ export async function upsertIssueWatchdogForIssue( return { watchdog: toIssueWatchdog(insertResult.row), created: insertResult.created }; } +export function upsertIssueWatchdogForIssue( + dbOrTx: Db | Parameters[0]>[0], + companyId: string, + issueId: string, + input: IssueWatchdogUpsertInput, +): Promise<{ watchdog: IssueWatchdog; created: boolean }> { + return dbOrTx.transaction(async tx => { + await notifyIssueRecovery(tx, { companyId }); + return upsertIssueWatchdogInTransaction(tx, companyId, issueId, input); + }); +} + export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) { const issuesSvc = issueService(db); @@ -1478,7 +1491,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) return created; } - async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null } = {}) { + async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null; onDeferredDeadline?: (at: number) => void } = {}) { const watchdog = await markTerminalWatchdogIssueReviewed(row, opts); const sourceIssue = await db .select() @@ -1491,6 +1504,13 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) const input = await collectClassifierInput(watchdog.companyId, watchdog); const classification = classifyTaskWatchdogSubtree(input); + if (classification.state === "pending_first_run") { + const pending = new Set(classification.pendingIssueIds); + for (const issue of input.issues) if (pending.has(issue.id)) { + const createdAt = toEpochMs(issue.createdAt); + if (createdAt !== null) opts.onDeferredDeadline?.(createdAt + TASK_WATCHDOG_FIRST_RUN_GRACE_MS); + } + } if (classification.state !== "stopped") { return { state: classification.state, reason: classification.reason, classification }; } @@ -1610,13 +1630,14 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) }; } - async function listActiveWatchdogsForCompany(companyId?: string | null) { + async function listActiveWatchdogsForCompany(companyId?: string | null, watchdogId?: string) { return db .select() .from(issueWatchdogs) .where(and( eq(issueWatchdogs.status, "active"), ...(companyId ? [eq(issueWatchdogs.companyId, companyId)] : []), + watchdogId ? eq(issueWatchdogs.id, watchdogId) : undefined, )); } @@ -1746,14 +1767,16 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) issueId: string, actor: ActorFields = {}, ): Promise => { - await assertWatchedIssue(db, companyId, issueId); - const existing = await db + return db.transaction(async tx => { + await assertWatchedIssue(tx, companyId, issueId); + const existing = await tx .select() .from(issueWatchdogs) .where(and(eq(issueWatchdogs.companyId, companyId), eq(issueWatchdogs.issueId, issueId))) .then((rows) => rows[0] ?? null); if (!existing || existing.status === "disabled") return null; - const [updated] = await db + await notifyIssueRecovery(tx, { companyId }); + const [updated] = await tx .update(issueWatchdogs) .set({ status: "disabled", @@ -1765,14 +1788,17 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) .where(eq(issueWatchdogs.id, existing.id)) .returning(); return toIssueWatchdog(updated); + }); }, reconcileTaskWatchdogs: async (opts: { companyId?: string | null; runId?: string | null; + watchdogId?: string; + onDeferredDeadline?: (at: number) => void; issueCreatedAtGte?: Date | null; } = {}) => { - let rows = await listActiveWatchdogsForCompany(opts.companyId ?? null); + let rows = await listActiveWatchdogsForCompany(opts.companyId ?? null, opts.watchdogId); if (opts.issueCreatedAtGte) { const watchdogIssueIds = [...new Set(rows.map((row) => row.issueId))]; const eligibleIssueIds = new Set( @@ -1802,7 +1828,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) }; for (const row of rows) { result.checked += 1; - const evaluated = await evaluateWatchdog(row, { runId: opts.runId ?? null }); + const evaluated = await evaluateWatchdog(row, { runId: opts.runId ?? null, onDeferredDeadline: opts.onDeferredDeadline }); if (evaluated.state === "triggered") { result.triggered += 1; result.watchdogIssueIds.push(evaluated.watchdogIssueId);