diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 0f74d6637777..77d0f96f0889 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -1319,19 +1319,16 @@ describeEmbeddedPostgres("issue recovery actions", () => { } // Fixed here: the escalation body's own `getCompanyIssuePrefix`, `getAgent` - // and `getLatestIssueRun` reads now run on `tx`. Still pooled under the - // lock, tracked on BLO-34207: owner resolution - // (`resolveStrandedIssueRecoveryOwnerAgentId` -> `getAgent` / - // `isAgentInvokable` / `budgets.getInvocationBlock` / instance settings). - // This ratchet fails on any new pooled call site and on a regression of the - // three fixed ones. - const knownPooledUnderLock = [ - "resolveStrandedIssueRecoveryOwnerAgentId", - "isAgentInvokable", - "evaluateAgentInvokabilityFromDb", - "getInvocationBlock", - "getOrCreateRow", - ]; + // and `getLatestIssueRun` reads now run on `tx`, and so does owner + // resolution (BLO-34207: `resolveStrandedIssueRecoveryOwnerAgentId` -> + // `getAgent` / `isAgentInvokable` / `budgets.getInvocationBlock` / + // instance settings). This ratchet fails on any new pooled call site and + // on a regression of any of them. + // Empty on purpose: nothing may run pooled under the lock any more. Master + // dropped `getLatestIssueRun` (29cdd6ab3) and this PR moved the last one, + // `getOrCreateRow`, onto the caller tx (`readInstanceSettingsOn`); it is + // asserted forbidden by name below, so it must not be allowlisted here. + const knownPooledUnderLock: string[] = []; const unexpected = pooledInsideTransaction.filter( (entry) => !knownPooledUnderLock.some((name) => entry.includes(name)), ); @@ -1343,6 +1340,22 @@ describeEmbeddedPostgres("issue recovery actions", () => { // Substring match, so this also covers `getLatestIssueRunForAgentStage` and // `getLatestIssueRunSince` — none of the three may run pooled under the lock. expect(pooledInsideTransaction.filter((entry) => entry.includes("getLatestIssueRun"))).toEqual([]); + // BLO-34207. Named individually rather than relying on the `unexpected` + // filter above: an allowlist entry is a substring match, so a future entry + // that happens to contain one of these names would silently re-admit it. + for (const name of [ + // BLO-34207: `issuesSvc.update` / `addComment` read instance settings on + // the caller handle now (`instanceSettingsOn`), so the singleton read no + // longer takes a second pool connection under the graph lock. + "getOrCreateRow", + "resolveStrandedIssueRecoveryOwnerAgentId", + "resolveInvokableRecoveryAgentId", + "isAgentInvokable", + "evaluateAgentInvokabilityFromDb", + "getInvocationBlock", + ]) { + expect(pooledInsideTransaction.filter((entry) => entry.includes(name))).toEqual([]); + } const actionRows = await db .select() .from(issueRecoveryActions) diff --git a/server/src/__tests__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index bf5e15ed81da..22e3287a6c5b 100644 --- a/server/src/__tests__/issues-service.test.ts +++ b/server/src/__tests__/issues-service.test.ts @@ -8896,6 +8896,38 @@ describeEmbeddedPostgres("issueService.create workspace inheritance", () => { }); }); + // BLO-34207: issue mutations read instance settings on the caller's handle so + // they do not take a second pool connection while holding the company graph + // lock. That read must stay a pure read: bootstrapping the singleton row from + // inside the caller's tx takes an instance-wide row lock BEFORE + // `lockIssueParentMutationCompany` and holds it to commit, so two callers can + // take the two locks in opposite orders. Assert the write never happens. + it("does not write the instance settings singleton from an update running on a caller transaction", async () => { + const companyId = randomUUID(); + const issueId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Settings read must not write", + status: "todo", + priority: "medium", + }); + await db.delete(instanceSettings); + + await db.transaction(async (tx) => { + await svc.update(issueId, { title: "Renamed under a caller tx" }, tx); + }); + + expect(await db.select().from(instanceSettings)).toHaveLength(0); + }); + it("returns cycle validation instead of deadlocking intersecting multi-level reparent updates", async () => { const companyId = randomUUID(); const issueAId = randomUUID(); diff --git a/server/src/index.ts b/server/src/index.ts index d753662ec8b2..a90ae9898933 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1086,6 +1086,29 @@ export async function startServer(): Promise { // crashed run and handing that lock to the retry — exactly the interleaving // the in-tick `await` was added to remove. let crashReconcileSweepInFlight = false; + // BLO-34207: the same latch, for the reap → retry-promotion → queued-resume → + // stranded-reconcile → … chain below. That chain is sequential over every + // stranded candidate in the estate (147 distinct issues per pass, measured + // 2026-09-16) and each candidate takes the company-wide issue-graph advisory + // lock. One slow candidate makes the pass outlive the 30 s interval, the next + // tick starts a second pass, and the passes then contend with EACH OTHER on + // that lock: 8 waiters against `POSTGRES_POOL_MAX=10` starved the pool, so + // the holder could not get the second connection it needed to finish and only + // a waiter's 15 s `lock_timeout` broke the cycle. Every pass in the chain is + // idempotent, so a tick that finds one still running skips it. + // + // The latch inherits `crashReconcileSweepInFlight`'s failure mode: a pass that + // HANGS rather than rejects never clears it, and periodic recovery then stops + // estate-wide until restart with no output at all — the symptom is an absence. + // `heartbeatRecoveryChainStartedAt` makes that absence visible. It is not a + // timeout: clearing the latch without cancelling the work would re-admit the + // overlap the latch exists to remove, so this reports and does not act. + let heartbeatRecoveryChainInFlight = false; + let heartbeatRecoveryChainStartedAt = 0; + // 10 ticks. Above the normal case (a sweep routinely outlives one interval — + // that is what the latch is for) and far below the hours a wedged chain would + // otherwise sit silent. + const HEARTBEAT_RECOVERY_CHAIN_STALL_WARN_MS = 10 * config.heartbeatSchedulerIntervalMs; const heartbeatSchedulerInFlight = new Set>(); const trackHeartbeatSchedulerWork = (work: Promise) => { let tracked: Promise; @@ -1648,102 +1671,153 @@ export async function startServer(): Promise { // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. + // + // Deliberately NOT under `heartbeatRecoveryChainInFlight`. These four + // passes are the dispatch path — `resumeQueuedRuns` is what actually + // starts a queued run — and they do not take + // `lockIssueParentMutationCompany`, which is the contention the latch + // exists to remove. Gating them on the latched tail would couple + // dispatch to that tail's slowest pass: it ends in + // `reconcileContendedPrReviewerWakes`, which crosses the + // plugin-worker RPC bridge that logged ~976 x 30 s timeouts in the + // BLO-34207 incident. Resumption would then run once per whole chain + // instead of once per 30 s tick, reproducing the very symptom the + // latch is deployed against ("runs sat in `running` 15-20 min before + // their k8s Job was created"). trackHeartbeatSchedulerWork(heartbeat .resumeRunningExternalRuntimeRuns() .then(() => heartbeat.reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000 })) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); - const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); - if ( - promotion.promoted > 0 || - reconciled.assignmentDispatched > 0 || - reconciled.dispatchRequeued > 0 || - reconciled.continuationRequeued > 0 || - reconciled.successfulRunHandoffEscalated > 0 || - reconciled.reviewWaitingParked > 0 || - reconciled.waitingOnReviewResolved > 0 || - reconciled.escalated > 0 - ) { - logger.warn( - { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled }, - "periodic heartbeat recovery changed assigned issue state", - ); - } - }) - .then(async () => { - const reconciled = await heartbeat.reconcileIssueGraphLiveness(); - // BLO-29601: auto-resolving a dead escalation is a change to the issue - // graph too. Without it in this gate the drain runs silently and the only - // evidence it happened at all is the cancelled rows themselves. - if ( - reconciled.escalationsCreated > 0 || - reconciled.dependencyWakesHealed > 0 || - reconciled.staleEscalationsAutoResolved > 0 - ) { - logger.warn({ ...reconciled }, "periodic issue-graph liveness reconciliation changed issue graph state"); - } - }) - .then(async () => { - const reconciled = await heartbeat.reconcileTaskWatchdogs(); - if (reconciled.triggered > 0) { - logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work"); - } - }) - .then(async () => { - const scanned = await heartbeat.scanSilentActiveRuns(); - if (scanned.created > 0 || scanned.escalated > 0) { - logger.warn({ ...scanned }, "periodic active-run output watchdog created review work"); - } - }) - .then(async () => { - const reviewed = await heartbeat.reconcileProductivityReviews(); - // BLO-30303 AC4: unconditional — see the startup pass above. - logger.info({ ...reviewed }, "periodic productivity reconciliation funnel"); - if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) { - logger.warn({ ...reviewed }, "periodic productivity reconciliation created or updated review work"); - } - }) - .then(async () => { - const swept = await heartbeat.reconcileResolvedBlockerDependents(); - if (swept.woken > 0 || swept.failed > 0) { - logger.warn({ ...swept }, "periodic resolved-blocker-dependents sweep enqueued wakes"); - } - }) - .then(async () => { - const failedWakeDispatches = await heartbeat.reconcileFailedWakeDispatches(); - if (failedWakeDispatches.recovered > 0 || failedWakeDispatches.exhausted > 0) { + if (promotion.promoted > 0) { logger.warn( - { ...failedWakeDispatches }, - "periodic failed-wake-dispatch reconciliation retried durable wake failures (BLO-14395)", - ); - } - }) - .then(async () => { - // BLO-21995: replay PR-reviewer wakes that lost their PR-scope - // advisory lock at webhook time. GitHub never redelivers a 200, - // so this pass is the only path back for a sanctioned review - // request that lost that race. - const contendedReviewerWakes = await reconcileContendedPrReviewerWakes(db as any, { - webhookSecret: config.githubWebhookSecret || null, - pluginWorkerManager, - heartbeatOptions: { paperclipNodeRole: config.paperclipNodeRole }, - prReviewerAgentIds: config.githubPrReviewerAgentIds, - prReviewerBotLogin: config.prReviewerBotLogin || null, - }); - if ( - contendedReviewerWakes.recovered > 0 || - contendedReviewerWakes.exhausted > 0 - ) { - logger.warn( - { ...contendedReviewerWakes }, - "periodic contended PR-reviewer wake reconciliation replayed lock-contended review requests (BLO-21995)", + { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds }, + "periodic heartbeat dispatch promoted due scheduled retries", ); } }) .catch((err) => { - logger.error({ err }, "periodic heartbeat recovery failed"); + logger.error({ err }, "periodic heartbeat dispatch resumption failed"); })); + + if (heartbeatSchedulerStopped) return; + + // The lock-taking tail, single-flighted across ticks. Block form (not + // an early `return`) so a pass appended after it still runs while a + // chain is in flight — at 147 sequential candidates against a 30 s + // tick, "in flight" is the steady state, so an early return would + // make any later sibling silently dead. Matches + // `crashReconcileSweepInFlight` above. + if (!heartbeatRecoveryChainInFlight) { + heartbeatRecoveryChainInFlight = true; + heartbeatRecoveryChainStartedAt = Date.now(); + trackHeartbeatSchedulerWork(heartbeat + .reconcileStrandedAssignedIssues() + .then(async (reconciled) => { + if ( + reconciled.assignmentDispatched > 0 || + reconciled.dispatchRequeued > 0 || + reconciled.continuationRequeued > 0 || + reconciled.successfulRunHandoffEscalated > 0 || + reconciled.reviewWaitingParked > 0 || + reconciled.waitingOnReviewResolved > 0 || + reconciled.escalated > 0 + ) { + logger.warn({ ...reconciled }, "periodic heartbeat recovery changed assigned issue state"); + } + }) + .then(async () => { + const reconciled = await heartbeat.reconcileIssueGraphLiveness(); + // BLO-29601: auto-resolving a dead escalation is a change to the issue + // graph too. Without it in this gate the drain runs silently and the only + // evidence it happened at all is the cancelled rows themselves. + if ( + reconciled.escalationsCreated > 0 || + reconciled.dependencyWakesHealed > 0 || + reconciled.staleEscalationsAutoResolved > 0 + ) { + logger.warn({ ...reconciled }, "periodic issue-graph liveness reconciliation changed issue graph state"); + } + }) + .then(async () => { + const reconciled = await heartbeat.reconcileTaskWatchdogs(); + if (reconciled.triggered > 0) { + logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work"); + } + }) + .then(async () => { + const scanned = await heartbeat.scanSilentActiveRuns(); + if (scanned.created > 0 || scanned.escalated > 0) { + logger.warn({ ...scanned }, "periodic active-run output watchdog created review work"); + } + }) + .then(async () => { + const reviewed = await heartbeat.reconcileProductivityReviews(); + // BLO-30303 AC4: unconditional — see the startup pass above. + logger.info({ ...reviewed }, "periodic productivity reconciliation funnel"); + if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) { + logger.warn({ ...reviewed }, "periodic productivity reconciliation created or updated review work"); + } + }) + .then(async () => { + const swept = await heartbeat.reconcileResolvedBlockerDependents(); + if (swept.woken > 0 || swept.failed > 0) { + logger.warn({ ...swept }, "periodic resolved-blocker-dependents sweep enqueued wakes"); + } + }) + .then(async () => { + const failedWakeDispatches = await heartbeat.reconcileFailedWakeDispatches(); + if (failedWakeDispatches.recovered > 0 || failedWakeDispatches.exhausted > 0) { + logger.warn( + { ...failedWakeDispatches }, + "periodic failed-wake-dispatch reconciliation retried durable wake failures (BLO-14395)", + ); + } + }) + .then(async () => { + // BLO-21995: replay PR-reviewer wakes that lost their PR-scope + // advisory lock at webhook time. GitHub never redelivers a 200, + // so this pass is the only path back for a sanctioned review + // request that lost that race. + const contendedReviewerWakes = await reconcileContendedPrReviewerWakes(db as any, { + webhookSecret: config.githubWebhookSecret || null, + pluginWorkerManager, + heartbeatOptions: { paperclipNodeRole: config.paperclipNodeRole }, + prReviewerAgentIds: config.githubPrReviewerAgentIds, + prReviewerBotLogin: config.prReviewerBotLogin || null, + }); + if ( + contendedReviewerWakes.recovered > 0 || + contendedReviewerWakes.exhausted > 0 + ) { + logger.warn( + { ...contendedReviewerWakes }, + "periodic contended PR-reviewer wake reconciliation replayed lock-contended review requests (BLO-21995)", + ); + } + }) + .catch((err) => { + logger.error({ err }, "periodic heartbeat recovery failed"); + }) + .finally(() => { + heartbeatRecoveryChainInFlight = false; + heartbeatRecoveryChainStartedAt = 0; + })); + } else if ( + Date.now() - heartbeatRecoveryChainStartedAt > + HEARTBEAT_RECOVERY_CHAIN_STALL_WARN_MS + ) { + // Skipping is normal and silent; a chain that has been in flight for + // many ticks is not. Reported, not acted on — see the latch decl. + logger.warn( + { + inFlightMs: Date.now() - heartbeatRecoveryChainStartedAt, + warnAfterMs: HEARTBEAT_RECOVERY_CHAIN_STALL_WARN_MS, + }, + "periodic heartbeat recovery chain still in flight across many ticks; recovery passes are not running", + ); + } } })(); }, config.heartbeatSchedulerIntervalMs); diff --git a/server/src/services/agent-invokability.ts b/server/src/services/agent-invokability.ts index b9b39d59f5d3..155e2a9c35e4 100644 --- a/server/src/services/agent-invokability.ts +++ b/server/src/services/agent-invokability.ts @@ -4,6 +4,7 @@ import { getAgentWorkEligibility, type AgentEligibilityAgent, type AgentOrgChain import { eq } from "drizzle-orm"; type AgentStatus = (typeof agents.$inferSelect)["status"]; +type DbTransaction = Parameters[0]>[0]; export type AgentOrgRow = Pick< typeof agents.$inferSelect, @@ -116,7 +117,9 @@ export function evaluateAgentInvokability( } export async function evaluateAgentInvokabilityFromDb( - db: Db, + // Callers holding an advisory lock MUST pass their own transaction — a second + // pool connection here is what convoyed on BLO-34207. + db: Db | DbTransaction, agent: AgentOrgRow | null | undefined, ): Promise { if (!agent) return evaluateAgentInvokability(agent, []); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index ce7662bac52a..7d4b4028ba88 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -291,6 +291,35 @@ function toInstanceSettings(row: typeof instanceSettings.$inferSelect): Instance } as InstanceSettings; } +/** + * Read-only settings view for callers that already hold a transaction. + * + * BLO-34207: `getOrCreateRow` bootstraps the singleton row with + * `insert ... on conflict do update`. On the pool that is harmless — its own + * autocommitted statement — but run on a caller's tx it takes a row lock on a + * single instance-wide row and holds it to commit. Issue mutations take that + * read BEFORE `lockIssueParentMutationCompany`, so two transactions can acquire + * the singleton row and the company graph lock in opposite orders: a global + * serialization point with a deadlock edge, which is strictly worse than the + * convoy this change set exists to remove. + * + * A missing row and a freshly bootstrapped row normalize to exactly the same + * values (`general: {}` / `experimental: {}`), so reading defaults is + * equivalent to bootstrapping — minus the write. Writers still go through + * `instanceSettingsService`, which keeps creating the row. + */ +export async function readInstanceSettingsOn(dbOrTx: Db) { + const row = await dbOrTx + .select() + .from(instanceSettings) + .where(eq(instanceSettings.singletonKey, DEFAULT_SINGLETON_KEY)) + .then((rows) => rows[0] ?? null); + return { + general: normalizeGeneralSettings(row?.general), + experimental: normalizeExperimentalSettings(row?.experimental), + }; +} + export function instanceSettingsService(db: Db, options: InstanceSettingsServiceOptions = {}) { async function getOrCreateRow() { const existing = await db diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 85be68377bf6..8eacc9b19017 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -117,7 +117,7 @@ import { TERMINAL_HEARTBEAT_RUN_STATUSES, runOwnsIssueExecutionLock, } from "./issue-execution-lock.js"; -import { instanceSettingsService } from "./instance-settings.js"; +import { instanceSettingsService, readInstanceSettingsOn } from "./instance-settings.js"; import { assertNotDuplicatePrReviewIssue, lockPrReviewIssueScopes, @@ -5673,6 +5673,19 @@ function alertmanagerAggregateCreationFingerprint( export function issueService(db: Db) { const instanceSettings = instanceSettingsService(db); + // BLO-34207: `update` and `addComment` are called from inside a transaction + // that already holds `lockIssueParentMutationCompany` (recovery's + // `escalateStrandedAssignedIssue`). Reading instance settings off the pooled + // handle there takes a SECOND pool connection while the lock is held, and + // with `POSTGRES_POOL_MAX=10` against 8-9 waiters on that same key the read + // only gets a connection when a waiter hits its `lock_timeout` — the convoy. + // So bind the settings reads to the caller's handle when there is one. + // + // `readInstanceSettingsOn` and not `instanceSettingsService(tx)`: the latter + // bootstraps the singleton row, which on a caller's tx is a row lock taken + // BEFORE the company graph lock and held to commit — a deadlock edge. These + // are pure reads, so they take the read-only view on either handle. + const instanceSettingsOn = (dbOrTx: unknown) => readInstanceSettingsOn(dbOrTx as Db); const treeControlSvc = issueTreeControlService(db); async function lockIssueBlockerRelations( @@ -10677,7 +10690,7 @@ export function issueService(db: Db) { issueId: id, }); } - const experimental = await instanceSettings.getExperimental(); + const experimental = (await instanceSettingsOn(dbOrTx)).experimental; const isolatedWorkspacesEnabled = experimental.enableIsolatedWorkspaces; if (!isolatedWorkspacesEnabled) { delete issueData.executionWorkspaceId; @@ -12836,7 +12849,7 @@ export function issueService(db: Db) { if (!comment) return null; const currentUserRedactionOptions = { - enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs, + enabled: (await instanceSettingsOn(dbOrTx)).general.censorUsernameInLogs, }; return redactIssueComment(comment, currentUserRedactionOptions.enabled); }, @@ -12896,7 +12909,7 @@ export function issueService(db: Db) { if (!issue) throw notFound("Issue not found"); const currentUserRedactionOptions = { - enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs, + enabled: (await instanceSettingsOn(dbOrTx)).general.censorUsernameInLogs, }; const redactedBody = redactCurrentUserText(body, currentUserRedactionOptions); const authorType = issueCommentAuthorTypeSchema.parse( diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 6b9c65999b6e..169ea6212e1f 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -2296,8 +2296,20 @@ export function recoveryService( return dbOrTx.select().from(agents).where(eq(agents.id, agentId)).then((rows) => rows[0] ?? null); } - async function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined) { - return (await evaluateAgentInvokabilityFromDb(db, agent)).invokable; + // Same rule as `getAgent` above: under `lockIssueParentMutationCompany` this + // must read on the caller's transaction. + async function isAgentInvokable( + agent: typeof agents.$inferSelect | null | undefined, + dbOrTx: Db | DbTransaction = db, + ) { + return (await evaluateAgentInvokabilityFromDb(dbOrTx, agent)).invokable; + } + + // Budget reads are pure reads, so the tx-scoped service is only ever used to + // keep them off a second pool connection while a lock is held. Same idiom as + // `issueRecoveryActionService(dbOrTx)` below. + function budgetsOn(dbOrTx: Db | DbTransaction) { + return dbOrTx === db ? budgets : budgetService(dbOrTx as Db); } // Column set behind `LatestIssueRun`. Shared by every helper that produces @@ -5277,20 +5289,21 @@ export function recoveryService( async function resolveStrandedIssueRecoveryOwnerAgentId( issue: typeof issues.$inferSelect, preferredOwnerAgentId?: string | null, + dbOrTx: Db | DbTransaction = db, ) { const candidateIds: string[] = []; if (preferredOwnerAgentId) candidateIds.push(preferredOwnerAgentId); if (issue.assigneeAgentId) { - const assignee = await getAgent(issue.assigneeAgentId); + const assignee = await getAgent(issue.assigneeAgentId, dbOrTx); if (assignee?.reportsTo) candidateIds.push(assignee.reportsTo); } if (issue.createdByAgentId) { - const creator = await getAgent(issue.createdByAgentId); + const creator = await getAgent(issue.createdByAgentId, dbOrTx); if (creator?.reportsTo) candidateIds.push(creator.reportsTo); candidateIds.push(issue.createdByAgentId); } - const roleCandidates = await db + const roleCandidates = await dbOrTx .select() .from(agents) .where(and(eq(agents.companyId, issue.companyId), inArray(agents.role, ["cto", "ceo"]))) @@ -5302,13 +5315,13 @@ export function recoveryService( for (const agentId of candidateIds) { if (seen.has(agentId)) continue; seen.add(agentId); - const candidate = await getAgent(agentId); + const candidate = await getAgent(agentId, dbOrTx); if (!candidate || candidate.companyId !== issue.companyId) continue; - const budgetBlock = await budgets.getInvocationBlock(issue.companyId, candidate.id, { + const budgetBlock = await budgetsOn(dbOrTx).getInvocationBlock(issue.companyId, candidate.id, { issueId: issue.id, projectId: issue.projectId, }); - if ((await isAgentInvokable(candidate)) && !budgetBlock) return candidate.id; + if ((await isAgentInvokable(candidate, dbOrTx)) && !budgetBlock) return candidate.id; } return null; @@ -5317,15 +5330,16 @@ export function recoveryService( async function resolveInvokableRecoveryAgentId( issue: typeof issues.$inferSelect, agentId: string | null | undefined, + dbOrTx: Db | DbTransaction = db, ) { if (!agentId) return null; - const candidate = await getAgent(agentId); + const candidate = await getAgent(agentId, dbOrTx); if (!candidate || candidate.companyId !== issue.companyId) return null; - const budgetBlock = await budgets.getInvocationBlock(issue.companyId, candidate.id, { + const budgetBlock = await budgetsOn(dbOrTx).getInvocationBlock(issue.companyId, candidate.id, { issueId: issue.id, projectId: issue.projectId, }); - return (await isAgentInvokable(candidate)) && !budgetBlock ? candidate.id : null; + return (await isAgentInvokable(candidate, dbOrTx)) && !budgetBlock ? candidate.id : null; } async function resolveStrandedRecoveryRouting(input: { @@ -5335,7 +5349,7 @@ export function recoveryService( preferredOwnerAgentId?: string | null; existingReturnOwnerAgentId?: string | null; existingOwnerAgentId?: string | null; - }) { + }, dbOrTx: Db | DbTransaction = db) { // `originalAgentId` intentionally keeps `latestRun.agentId` as the first candidate: // `provider_quota` retries need the agent who actually hit the quota, which can // diverge from `issue.assigneeAgentId` once THIS function has already escalated @@ -5398,10 +5412,10 @@ export function recoveryService( (ROUTE_TO_ORIGINAL_INFRA_ERROR_CODES.has(input.latestRun?.errorCode ?? "") || isInfraClassStrandedFailure(input.latestRun))); if (input.recoveryCause === "provider_quota") { - const retryAgentId = await resolveInvokableRecoveryAgentId(input.issue, originalAgentId); + const retryAgentId = await resolveInvokableRecoveryAgentId(input.issue, originalAgentId, dbOrTx); if (!retryAgentId) { return { - ownerAgentId: await resolveStrandedIssueRecoveryOwnerAgentId(input.issue), + ownerAgentId: await resolveStrandedIssueRecoveryOwnerAgentId(input.issue, null, dbOrTx), returnOwnerAgentId: originalAgentId, routingFallbackReason: "The original assignee is not invokable; quota recovery fell through to the manager ladder.", }; @@ -5413,12 +5427,12 @@ export function recoveryService( }; } if (routeToOriginal) { - const ownerAgentId = await resolveInvokableRecoveryAgentId(input.issue, returnOwnerAgentId); + const ownerAgentId = await resolveInvokableRecoveryAgentId(input.issue, returnOwnerAgentId, dbOrTx); if (ownerAgentId) { return { ownerAgentId, returnOwnerAgentId, routingFallbackReason: null }; } return { - ownerAgentId: await resolveStrandedIssueRecoveryOwnerAgentId(input.issue), + ownerAgentId: await resolveStrandedIssueRecoveryOwnerAgentId(input.issue, null, dbOrTx), returnOwnerAgentId, routingFallbackReason: "The original assignee is not invokable; recovery fell through to the manager ladder.", }; @@ -5427,6 +5441,7 @@ export function recoveryService( ownerAgentId: await resolveStrandedIssueRecoveryOwnerAgentId( input.issue, input.preferredOwnerAgentId, + dbOrTx, ), returnOwnerAgentId, routingFallbackReason: null, @@ -5815,7 +5830,7 @@ export function recoveryService( preferredOwnerAgentId: input.recoveryOwnerAgentId, existingReturnOwnerAgentId: existingAction?.returnOwnerAgentId, existingOwnerAgentId: existingAction?.ownerAgentId, - }); + }, dbOrTx); const ownerAgentId = routing.ownerAgentId; // BLO-18996: the single predicate for "will any sweep wake an owner for this action". // The wake budget and the wake path have to agree, and previously they were written as