diff --git a/server/src/__tests__/heartbeat-dispatch-priority-sort.test.ts b/server/src/__tests__/heartbeat-dispatch-priority-sort.test.ts index 6239a3ebdb27..e8423ffe4d03 100644 --- a/server/src/__tests__/heartbeat-dispatch-priority-sort.test.ts +++ b/server/src/__tests__/heartbeat-dispatch-priority-sort.test.ts @@ -2992,6 +2992,167 @@ describeEmbeddedPostgres("heartbeat dispatch priority sort (BLO-12990)", () => { } }, 300_000); + it("does not let an aged isolation-retry-deferred critical row strand a fresh critical arrival via a false-ready absolute-floor rank (BLO-22091)", async () => { + // `dispatchRank`'s own `ready` computation previously called + // `isEffectivelyDependencyReadyForDispatch` alone, which screens only + // dependency-readiness. An isolation-retry-deferred row that IS + // dependency-ready therefore read `ready: true` there even though + // `claimQueuedRun` refuses it unconditionally at the top of the function + // (`isK8sIsolationRetryDeferred`) — the lanes above already knew to + // exclude such a row, but this one site never re-derived that screen. + // + // Once such a row ages past the absolute floor it ranked + // `ABSOLUTE_STARVATION_DISPATCH_RANK` — the very front of the queue. + // Because it is critical-priority, lane A had already marked it as + // emergency-lane work, so the claim loop's refusal handler, + // `scheduleEmergencyContinuationForStillQueuedRun`, aborted the whole pass + // on its refusal rather than falling through to `freshRunId` ranked right + // behind it, and scheduled a "resume_critical_lane" detached retry. + // + // That retry is self-healing — `dispatchDeferredRunIdsByAgent` excludes + // the refused row from the next scan, so `freshRunId` dispatches a moment + // later even pre-fix. Asserting only the EVENTUAL outcome would therefore + // pass on both sides of the fix and miss the regression entirely. What + // must not happen, and is what actually distinguishes pre-fix from + // post-fix, is the reschedule itself: post-fix `freshRunId` outranks the + // deferred row directly, so the claim loop claims it on the FIRST attempt + // of the FIRST pass and never even reaches the deferred row — no refusal, + // no reschedule. `onQueuedDispatchScheduledForTest` fires synchronously + // the instant a reschedule is requested (see `scheduleDetachedDispatchPass`), + // so this is observable without polling or racing detached execution. + const companyId = randomUUID(); + const agentId = randomUUID(); + const deferredIssueId = randomUUID(); + const freshIssueId = randomUUID(); + const issuePrefix = `D${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + const criticalLaneReschedules: Array<{ agentId: string; reason: string }> = []; + const boundedHeartbeat = heartbeatService(db, { + penstockGate: allowPenstockGate, + onQueuedDispatchScheduledForTest: (event) => { + if (event.reason === "resume_critical_lane") { + criticalLaneReschedules.push({ agentId: event.agentId, reason: event.reason }); + } + }, + }); + + await db.insert(companies).values({ + id: companyId, + name: "IsolationDeferredAbsoluteFloorCo", + issuePrefix, + requireBoardApprovalForNewAgents: false, + defaultResponsibleUserId: "responsible-user", + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "IsolationDeferredAbsoluteFloorAgent", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, + permissions: {}, + }); + + await db.insert(issues).values([ + { + id: deferredIssueId, + companyId, + title: "Critical work whose run is deferred for a k8s isolation retry", + status: "todo", + priority: "critical", + assigneeAgentId: agentId, + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }, + { + id: freshIssueId, + companyId, + title: "Fresh critical arrival that must not be stranded behind it", + status: "todo", + priority: "critical", + assigneeAgentId: agentId, + issueNumber: 2, + identifier: `${issuePrefix}-2`, + }, + ]); + + const deferredRunId = randomUUID(); + const freshRunId = randomUUID(); + // 7h > STARVATION_ABSOLUTE_ESCALATION_MS (6h): aged past the absolute + // floor, exactly the age at which the false-`ready` bug promoted this row + // to the very front of the queue. + const deferredCreatedAt = new Date(Date.now() - 7 * 60 * 60 * 1000); + const freshCreatedAt = new Date(); + + await db.insert(heartbeatRuns).values([ + { + id: deferredRunId, + companyId, + agentId, + invocationSource: "automation", + triggerDetail: "system", + status: "queued", + contextSnapshot: { + issueId: deferredIssueId, + wakeReason: "issue_assigned", + // An hour out - genuinely unclaimable for the whole test, by design. + paperclipK8sIsolationRetryAt: new Date(Date.now() + 60 * 60_000).toISOString(), + }, + createdAt: deferredCreatedAt, + updatedAt: deferredCreatedAt, + }, + { + id: freshRunId, + companyId, + agentId, + invocationSource: "heartbeat", + triggerDetail: "timer", + status: "queued", + contextSnapshot: { issueId: freshIssueId, wakeReason: "issue_assigned" }, + createdAt: freshCreatedAt, + updatedAt: freshCreatedAt, + }, + ]); + + const dispatchedRunIds: string[] = []; + mockAdapterExecute.mockImplementation(async (args: { runId: string }) => { + dispatchedRunIds.push(args.runId); + return { + exitCode: 0, + signal: null as string | null, + timedOut: false, + errorMessage: null as string | null, + resultJson: { exitCode: 0 }, + provider: "test", + model: "test-model", + }; + }); + + try { + await boundedHeartbeat.resumeQueuedRuns(); + + // The regression, checked at the point it actually happens: the first + // pass must not have needed to abort and reschedule at all. + expect(criticalLaneReschedules).toHaveLength(0); + + await waitForRunToSettle(boundedHeartbeat, freshRunId); + + // Sanity check on the harness, not the regression itself (see above): + // the row does need to actually dispatch, just without the detour. + expect(dispatchedRunIds).toContain(freshRunId); + // The deferred row must still not dispatch - ranking it correctly must + // not make it disappear or dispatch early; it stays queued until its + // own retry timestamp arrives. + expect(dispatchedRunIds).not.toContain(deferredRunId); + expect((await boundedHeartbeat.getRun(deferredRunId))?.status).toBe("queued"); + } finally { + await boundedHeartbeat.drainInFlightExecutions(60_000); + } + }); + it("dispatches critical issue work before an aged issue-less run without starving routine issue work (BLO-19337)", async () => { // Regression for BLO-18995: dispatchRank returned a flat `10` for any run // without an issueId *above* the STARVATION_* aging escalation, so that diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e68038973fbf..4e3dbce90d22 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -5601,6 +5601,75 @@ function isEffectivelyDependencyReadyForDispatch( return allowsIssueInteractionWake(contextSnapshot); } +/** + * The single definition of "is this queued run's issue actually claimable + * right now" (BLO-22091). Every dispatch lane's eligibility test + * (`foundReadyCritical`, `foundReadyRecovery`, `foundReadyAbsolute`) calls + * this and nothing else, so a future change to any one screen cannot + * silently desynchronize the lanes from what the claim path actually + * accepts — the exact shape of leaks 2-5 (BLO-21792), which were each one + * site re-deriving this verdict slightly differently from the others. + * + * Three screens, matching what `claimQueuedRun` applies to a routine + * (non-reopen) queued run: + * 1. The issue must exist and be non-terminal (`TERMINAL_ISSUE_STATUSES`). + * 2. It must not be a k8s-isolation retry still waiting on its timestamp. + * 3. It must be dependency-ready, or exempted via a verified issue- + * interaction wake (`isEffectivelyDependencyReadyForDispatch` above). + * + * `claimQueuedRun`'s own dependency-blocker gate calls + * `isEffectivelyDependencyReadyForDispatch` directly rather than this whole + * predicate, and the `dispatchRank` invocation below calls the narrower + * sibling `isDispatchRankReady` instead of this — see that function's + * comment for why terminal-status can't be folded in there without + * reintroducing an unbounded wait for a different exemption. + */ +function isRunClaimable( + contextSnapshot: Record | null | undefined, + issue: { status: string } | null | undefined, + readiness: { isDependencyReady: boolean } | null | undefined, + now: Date, +): boolean { + if (!issue || TERMINAL_ISSUE_STATUSES.has(issue.status)) return false; + if (isK8sIsolationRetryDeferred(contextSnapshot, now)) return false; + return isEffectivelyDependencyReadyForDispatch(contextSnapshot, readiness); +} + +/** + * `dispatchRank`'s `ready` screen (BLO-22091). Deliberately narrower than + * `isRunClaimable` above: it omits the terminal-issue check on purpose. + * + * By the time a row reaches the ranking step, `queuedRuns` has already been + * pruned for terminal issues by the main scan via `evaluateQueuedRunStaleness` + * — the same function `claimQueuedRun` itself uses — which lets a + * `resumeIntent`/wake-comment-carrying row survive against a done/cancelled + * issue so a comment can reopen it. That pruning is authoritative and + * resume-intent-aware; this function must not re-exclude what it already + * decided to keep. If it did, `dispatchRank`'s `!ready` branch never ages + * (a row that "cannot be claimed" must not escalate to the front, however + * long it has waited — see that function) so a legitimately-claimable + * reopened row would rank `12+` forever and starve exactly the way this + * ticket exists to prevent, just via a different exemption than leak 5's. + * + * The lanes above don't have this problem: they already excluded terminal + * issues from their own candidate batches unconditionally before this fix + * (via `criticalIssueById`'s filter, lane B's `candidates` filter, and lane + * C's inline check), so a resume-intent-exempted terminal row was already + * never surfaced as "found ready" by any lane — `isRunClaimable` preserves + * that pre-existing behavior rather than changing it. Only `dispatchRank`'s + * `ready` previously had *no* terminal screen at all (trusting the upstream + * prune), so adding the full `isRunClaimable` there — rather than this + * narrower sibling — would have been a net new regression, not a fix. + */ +function isDispatchRankReady( + contextSnapshot: Record | null | undefined, + readiness: { isDependencyReady: boolean } | null | undefined, + now: Date, +): boolean { + if (isK8sIsolationRetryDeferred(contextSnapshot, now)) return false; + return isEffectivelyDependencyReadyForDispatch(contextSnapshot, readiness); +} + async function listUnresolvedBlockerSummaries( dbOrTx: Pick, companyId: string, @@ -15658,11 +15727,18 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const dependencyReadiness = await issuesSvc.listDependencyReadiness(run.companyId, [issueId]); const readiness = dependencyReadiness.get(issueId); - const unresolvedBlockerCount = readiness?.unresolvedBlockerCount ?? 0; - issueDependencyReadyForAutoCheckout = unresolvedBlockerCount === 0; - if (unresolvedBlockerCount > 0 && !allowsIssueInteractionWake(context)) { + issueDependencyReadyForAutoCheckout = readiness?.isDependencyReady ?? true; + // Consumes the exact same predicate every dispatch lane and dispatchRank + // call for this screen (`isRunClaimable`, BLO-22091) instead of + // re-deriving the blocker-count/interaction-wake boolean locally — that + // local re-derivation is how leak 5 (BLO-21792) happened: this gate and + // the lanes agreed by convention, not by construction, until they didn't. + if (!isEffectivelyDependencyReadyForDispatch(context, readiness)) { await cancelQueuedRunForBlockedDependencies(run, issueId, readiness?.unresolvedBlockerIssueIds ?? []); - logger.info({ runId: run.id, issueId, unresolvedBlockerCount }, "claimQueuedRun: cancelled blocked queued run"); + logger.info( + { runId: run.id, issueId, unresolvedBlockerCount: readiness?.unresolvedBlockerCount ?? 0 }, + "claimQueuedRun: cancelled blocked queued run", + ); return null; } @@ -20088,15 +20164,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) for (const { issue } of criticalBatch) issueById.set(issue.id, issue); // Dependency-ready is not necessarily claimable: isolation retries // remain queued until their retry timestamp. Keep paging past them so - // a deferred head row cannot hide runnable emergency work. + // a deferred head row cannot hide runnable emergency work. Calls the + // single `isRunClaimable` predicate (BLO-22091) rather than + // re-deriving its screens here. foundReadyCritical = criticalRuns.some((run) => { const snapshot = parseObject(run.contextSnapshot); const issueId = readNonEmptyString(snapshot.issueId); - return Boolean( - issueId - && isEffectivelyDependencyReadyForDispatch(snapshot, batchReadiness.get(issueId)) - && !isK8sIsolationRetryDeferred(snapshot, dispatchNow) - ); + const issue = issueId ? issueById.get(issueId) : null; + return isRunClaimable(snapshot, issue, issueId ? batchReadiness.get(issueId) : null, dispatchNow); }); if (foundReadyCritical || criticalLaneExhausted) break; } @@ -20204,14 +20279,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } const batchReadiness = await listQueuedRunDependencyReadiness(agent.companyId, candidates); for (const [issueId, readiness] of batchReadiness) dependencyReadiness.set(issueId, readiness); + // Calls the single `isRunClaimable` predicate (BLO-22091) rather than + // re-deriving its screens here; `candidates` is already terminal- + // filtered above, so the issue lookup below is never null in practice. foundReadyRecovery = candidates.some((run) => { const snapshot = parseObject(run.contextSnapshot); const issueId = readNonEmptyString(snapshot.issueId); - return Boolean( - issueId - && isEffectivelyDependencyReadyForDispatch(snapshot, batchReadiness.get(issueId)) - && !isK8sIsolationRetryDeferred(snapshot, dispatchNow) - ); + const issue = issueId ? issueById.get(issueId) : null; + return isRunClaimable(snapshot, issue, issueId ? batchReadiness.get(issueId) : null, dispatchNow); }); if (foundReadyRecovery || recoveryLaneExhausted) break; } @@ -20369,22 +20444,15 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) for (const [issueId, readiness] of batchReadiness) dependencyReadiness.set(issueId, readiness); // "Eligible" must mean the same thing the claim loop means, or the // lane stops paging on a row that will not in fact be claimed and the - // masking survives in a narrower form. A row counts only if it has a - // live issue, is *effectively* dependency-ready (see - // `isEffectivelyDependencyReadyForDispatch` — a blocked issue whose - // wake is an interaction wake IS claimable), and is not an isolation - // retry still waiting on its timestamp — the same screens the claim - // path applies, matching how lane A defines `foundReadyCritical`. + // masking survives in a narrower form. Calls the single + // `isRunClaimable` predicate (BLO-22091) — the same one lane A, lane + // B, and the dispatchRank invocation below call — so this can no + // longer independently drift from what `claimQueuedRun` decides. foundReadyAbsolute = batch.some((run) => { const snapshot = parseObject(run.contextSnapshot); const issueId = readNonEmptyString(snapshot.issueId); - if (!issueId) return false; - const issue = issueById.get(issueId); - if (!issue || TERMINAL_ISSUE_STATUSES.has(issue.status)) return false; - return Boolean( - isEffectivelyDependencyReadyForDispatch(snapshot, batchReadiness.get(issueId)) - && !isK8sIsolationRetryDeferred(snapshot, dispatchNow) - ); + const issue = issueId ? issueById.get(issueId) : null; + return isRunClaimable(snapshot, issue, issueId ? batchReadiness.get(issueId) : null, dispatchNow); }); if (foundReadyAbsolute || absoluteLaneExhausted) break; } @@ -20672,10 +20740,20 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) for (const queuedRun of queuedRuns) { const snapshot = parseObject(queuedRun.contextSnapshot); const issueId = readNonEmptyString(snapshot.issueId); - const ready = issueId - ? isEffectivelyDependencyReadyForDispatch(snapshot, dependencyReadiness.get(issueId)) - : true; const issue = issueId ? issueById.get(issueId) : null; + // Calls the single `isRunClaimable` predicate (BLO-22091) so a row's + // rank can never disagree with the lanes above about what "ready" + // means for the screens that stay identical at every stage. This + // previously called `isEffectivelyDependencyReadyForDispatch` alone, + // which only screens dependency-readiness: an isolation-retry- + // deferred row that was otherwise dependency-ready read `ready: true` + // here, so once it aged past the absolute floor it ranked + // `ABSOLUTE_STARVATION_DISPATCH_RANK` — the very front of the queue — + // despite `claimQueuedRun` refusing it unconditionally at claim time. + // Deliberately `isDispatchRankReady`, not the lanes' `isRunClaimable` + // — see that function's comment for why terminal-status must stay + // out of this specific screen. + const ready = issueId ? isDispatchRankReady(snapshot, dependencyReadiness.get(issueId), dispatchNow) : true; // Require both recoveryActionId AND source:"issue_recovery_action" (every // enqueueWakeup call in recovery/service.ts stamps both together) so this // stays an explicit, narrowly-scoped coupling rather than any future wake