From 412bec927995ccc1bab72601c35683d02a3401dd Mon Sep 17 00:00:00 2001 From: Omar Ramadan Date: Sat, 8 Aug 2026 13:00:58 +0000 Subject: [PATCH 1/3] fix(recovery): preserve lock handover evidence (BLO-19160) Co-Authored-By: Paperclip --- .../__tests__/issue-recovery-actions.test.ts | 233 +++++++++++- server/src/services/recovery/service.ts | 333 +++++++++++++++--- 2 files changed, 521 insertions(+), 45 deletions(-) diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index abba31ad9c20..97b21d5e1a94 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -43,6 +43,33 @@ import { strandedRecoveryWakeAttemptsExhausted, } from "../services/recovery/service.js"; +// BLO-19160: seam for the adoption-interleaving regression test. The stranded +// sweep reads its candidates as one bulk snapshot and only reaches the +// checkout-handover branch several awaits later, so an adoption committing in +// that window used to be followed with the snapshot's stale lock ids. +// `isAutomaticRecoverySuppressedByPauseHold` is the last await before +// `getLatestIssueRun`, which makes it the precise seam for "commit an adoption +// between the snapshot load and the handover branch". The mock delegates to the +// real implementation and the hook is null unless a test arms it, so every other +// test in this file is unaffected. +const pauseHoldSeam = vi.hoisted(() => ({ + onNextCheck: null as null | (() => Promise), +})); +vi.mock("../services/recovery/pause-hold-guard.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + isAutomaticRecoverySuppressedByPauseHold: async ( + ...args: Parameters + ) => { + const hook = pauseHoldSeam.onNextCheck; + pauseHoldSeam.onNextCheck = null; + if (hook) await hook(); + return actual.isAutomaticRecoverySuppressedByPauseHold(...args); + }, + }; +}); + const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; const defaultRecoveryActionMaxAttempts = loadConfig().recoveryActionMaxAttempts; @@ -358,6 +385,9 @@ describeEmbeddedPostgres("issue recovery actions", () => { }, 120_000); afterEach(async () => { + // Defensive: a test that arms the seam but never reaches it must not leak + // the hook into the next test. + pauseHoldSeam.onNextCheck = null; await db.delete(issueRecoveryActions); await db.delete(issueComments); await db.delete(environmentLeases); @@ -5100,7 +5130,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { }) .where(eq(issues.id, seeded.sourceIssueId)); - return { ...seeded, deadCheckoutRunId, queuedContextRunId, adoptingRunId }; + return { ...seeded, deadCheckoutRunId, queuedContextRunId, adoptingRunId, otherIssueId }; } function agentActor(companyId: string, agentId: string, runId: string) { @@ -5263,6 +5293,207 @@ describeEmbeddedPostgres("issue recovery actions", () => { }); }); + // BLO-19160 finding 1: once the adopter is terminal it is, by construction, + // scoped to a DIFFERENT issue — `getLatestIssueRun` could not see it + // otherwise. Substituting it as this issue's `latestRun` handed every + // downstream classifier (error code, workspace result, quota state, + // liveness, retry budget) evidence describing someone else's work. The + // sibling test above only covers a plain failure, which classifies as + // retryable and happens to land on the same re-dispatch either way; these + // two cover the outcomes where the foreign verdict actually changes what + // happens to the adopted issue. + it("does not block the adopted issue on a foreign adopter's non-retryable failure", async () => { + const { companyId, coderId, sourceIssueId, adoptingRunId, otherIssueId, queuedContextRunId } = + await seedAdoptedCheckout({ adoptingRunStatus: "running" }); + + const res = await request(createApp(agentActor(companyId, coderId, adoptingRunId))) + .patch(`/api/issues/${sourceIssueId}`) + .send({ title: "Annotated while stalled" }); + expect(res.status, JSON.stringify(res.body)).toBe(200); + + // The adopter dies on a workspace defect belonging to the OTHER issue it + // was dispatched for — `workspace_repo_mismatch` is non-retryable, so + // reading it as this issue's evidence blocks and reassigns this issue for + // a condition that was never true of it. + await db + .update(heartbeatRuns) + .set({ + status: "failed", + error: "workspace repo does not match the issue's project", + errorCode: "workspace_repo_mismatch", + contextSnapshot: { issueId: otherIssueId }, + finishedAt: new Date("2026-07-29T12:30:00.000Z"), + }) + .where(eq(heartbeatRuns.id, adoptingRunId)); + + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + const result = await recovery.reconcileStrandedAssignedIssues(); + + // Neither blocked nor reassigned, and no recovery action citing a foreign + // run as this issue's evidence. + expect(result.escalated).toBe(0); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + const [reconciled] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(reconciled).toMatchObject({ status: "in_progress", assigneeAgentId: coderId }); + // Instead: neutral continuation recovery. The retry parent is the handover + // marker (scoped to THIS issue), never the foreign adopter. + expect(enqueueWakeup).toHaveBeenCalledWith( + coderId, + expect.objectContaining({ + reason: "issue_continuation_needed", + payload: expect.objectContaining({ issueId: sourceIssueId }), + }), + ); + const [wakeCall] = enqueueWakeup.mock.calls as unknown as [ + [string, { payload: Record }], + ]; + // Assert the marker run positively, not just "not the foreign adopter": + // `!== adoptingRunId` also passes when provenance is dropped entirely, so + // it does not actually prove the promised marker provenance. + expect(wakeCall[1].payload.retryOfRunId).toBe(queuedContextRunId); + }); + + it("does not suppress recovery on a foreign adopter's quota exhaustion", async () => { + const { companyId, coderId, sourceIssueId, adoptingRunId, otherIssueId } = + await seedAdoptedCheckout({ adoptingRunStatus: "running" }); + + const res = await request(createApp(agentActor(companyId, coderId, adoptingRunId))) + .patch(`/api/issues/${sourceIssueId}`) + .send({ title: "Annotated while stalled" }); + expect(res.status, JSON.stringify(res.body)).toBe(200); + + await db + .update(heartbeatRuns) + .set({ + status: "failed", + error: "provider usage limit reached", + errorCode: "provider_quota_exhausted", + contextSnapshot: { issueId: otherIssueId }, + finishedAt: new Date("2026-07-29T12:30:00.000Z"), + }) + .where(eq(heartbeatRuns.id, adoptingRunId)); + // A provider-quota monitor armed against that same foreign run — the + // downstream consequence of the same substitution, and the second way the + // adopted issue's recovery got suppressed on another issue's quota state. + await db + .update(issues) + .set({ + monitorNextCheckAt: new Date(Date.now() + 60 * 60 * 1000), + executionPolicy: { + mode: "normal", + commentRequired: true, + stages: [], + monitor: { + nextCheckAt: new Date(Date.now() + 60 * 60 * 1000).toISOString(), + scheduledBy: "assignee", + kind: "external_service", + serviceName: "provider_quota_recovery", + externalRef: adoptingRunId, + }, + }, + }) + .where(eq(issues.id, sourceIssueId)); + + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + const result = await recovery.reconcileStrandedAssignedIssues(); + + // Recovery runs rather than being suppressed by a quota condition that + // belongs to another issue's run, and still does not escalate. + expect(result.escalated).toBe(0); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + expect(enqueueWakeup).toHaveBeenCalledWith( + coderId, + expect.objectContaining({ + reason: "issue_continuation_needed", + payload: expect.objectContaining({ issueId: sourceIssueId }), + }), + ); + const [reconciled] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(reconciled).toMatchObject({ status: "in_progress", assigneeAgentId: coderId }); + }); + + // BLO-19160 finding 2: the handover branch used to read `executionRunId` / + // `checkoutRunId` off the pre-loop candidate snapshot. An adoption that + // commits inside that window leaves the sweep observing the new handover + // marker while following the OLD lock ids — resolving the previous terminal + // owner instead of the live adopter, and escalating the issue out from under + // a run that holds both current locks. That is a narrow-window re-entry into + // the exact BLO-18860 failure mode. + it("resolves the live adopter when an adoption commits after the candidate snapshot", async () => { + const { companyId, coderId, sourceIssueId, adoptingRunId } = await seedAdoptedCheckout({ + adoptingRunStatus: "running", + }); + + // First adoption: run A takes the checkout and produces the handover + // marker, then goes terminal. Preserve the in-progress status marker so + // current master can clean up A's stale ownership without restoring the + // issue to `todo`; B can then adopt the unowned in-progress checkout in + // the seam below. + const first = await request(createApp(agentActor(companyId, coderId, adoptingRunId))) + .patch(`/api/issues/${sourceIssueId}`) + .send({ title: "Annotated while stalled" }); + expect(first.status, JSON.stringify(first.body)).toBe(200); + await db + .update(heartbeatRuns) + .set({ + status: "failed", + error: "adopter A died", + finishedAt: new Date("2026-07-29T12:30:00.000Z"), + }) + .where(eq(heartbeatRuns.id, adoptingRunId)); + await db + .update(issues) + .set({ checkoutRestoreStatus: "in_progress" }) + .where(eq(issues.id, sourceIssueId)); + + // Run B: the assignee's next live run, scoped to yet another issue. + const liveAdopterRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: liveAdopterRunId, + companyId, + agentId: coderId, + invocationSource: "timer", + status: "running", + contextSnapshot: { issueId: randomUUID() }, + createdAt: new Date("2026-07-29T13:00:00.000Z"), + startedAt: new Date("2026-07-29T13:00:00.000Z"), + }); + + // B adopts *during* the sweep — after the candidate snapshot is taken, + // before the handover branch runs (see `pauseHoldSeam`). + pauseHoldSeam.onNextCheck = async () => { + const second = await request(createApp(agentActor(companyId, coderId, liveAdopterRunId))) + .patch(`/api/issues/${sourceIssueId}`) + .send({ title: "Annotated again by the live run" }); + expect(second.status, JSON.stringify(second.body)).toBe(200); + const [afterSecond] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(afterSecond).toMatchObject({ + executionRunId: liveAdopterRunId, + checkoutRunId: liveAdopterRunId, + }); + }; + + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(pauseHoldSeam.onNextCheck).toBeNull(); + // The sweep followed the CURRENT lock to the live adopter B and read + // continuity — no escalation, no reassignment, no competing wake. + expect(result.escalated).toBe(0); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + expect(enqueueWakeup).not.toHaveBeenCalled(); + const [reconciled] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(reconciled).toMatchObject({ + status: "in_progress", + assigneeAgentId: coderId, + executionRunId: liveAdopterRunId, + checkoutRunId: liveAdopterRunId, + }); + }); + it("still escalates a genuinely stranded issue after a spent continuation retry", async () => { const { companyId, managerId, coderId, sourceIssueId } = await seedCompany(); const failedRunId = randomUUID(); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index cba0d78ecd4d..b62533512a46 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1000,6 +1000,32 @@ function isTerminalDispatchRaceRun( ); } +// BLO-19160: the three issue columns adoption rewrites — the execution lock pair +// plus the owner. Captured fresh when a handover marker is observed and used as +// a compare-and-set precondition on every recovery mutation that observation +// leads to. +type IssueLockOwnerState = { + executionRunId: string | null; + checkoutRunId: string | null; + assigneeAgentId: string | null; +}; + +function issueLockOwnerStateMatches(a: IssueLockOwnerState, b: IssueLockOwnerState) { + return a.executionRunId === b.executionRunId && + a.checkoutRunId === b.checkoutRunId && + a.assigneeAgentId === b.assigneeAgentId; +} + +// BLO-19160: the outcome of observing a checkout-handover marker when the +// adopter can no longer prove continuity. `markerRunId` is the handover run — +// the newest run genuinely scoped to this issue, so the honest retry parent. +// `lockOwnerState` is the CAS precondition for every mutation the observation +// leads to. +type AdoptionHandoverNeutralRecovery = { + markerRunId: string; + lockOwnerState: IssueLockOwnerState; +}; + function buildNonRetryableEscalationComment(input: { status: "todo" | "in_progress"; latestRun: LatestIssueRun; @@ -1765,6 +1791,94 @@ export function recoveryService( .then((rows) => rows[0] ?? null); } + // BLO-19160: the execution lock + owner as of a specific instant. The + // stranded-assigned sweep reads its candidates as one bulk snapshot and then + // performs many awaits per candidate, so by the time a per-candidate branch + // runs the snapshot's lock columns may be several seconds stale. Adoption + // rewrites exactly these three fields, so a handover observed against a stale + // snapshot resolves the *previous* terminal owner while a live adopter holds + // the current lock — and recovery then escalates or reassigns the issue out + // from under that live run. Re-read them at observation time, and CAS on them + // before any mutation the observation led to. + async function readIssueLockOwnerState( + companyId: string, + issueId: string, + ): Promise { + const [row] = await db + .select({ + executionRunId: issues.executionRunId, + checkoutRunId: issues.checkoutRunId, + assigneeAgentId: issues.assigneeAgentId, + }) + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.id, issueId))) + .limit(1); + return row ?? null; + } + + // True when `expected` no longer describes the issue's committed lock/owner — + // i.e. this sweep lost the race and must take no side effect. A vanished issue + // counts as changed. `null`/`undefined` expectation means "no handover was + // observed on this candidate", so nothing extra is enforced. + async function issueLockOwnerStateChanged( + issueId: string, + expected: IssueLockOwnerState | null | undefined, + ): Promise { + if (!expected) return false; + const [fresh] = await db + .select({ + executionRunId: issues.executionRunId, + checkoutRunId: issues.checkoutRunId, + assigneeAgentId: issues.assigneeAgentId, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .limit(1); + if (!fresh) return true; + return !issueLockOwnerStateMatches(expected, fresh); + } + + // Decide what a checkout-handover marker means for the issue it is scoped to. + // Returns null when there is nothing to recover — either a live same-assignee + // adopter holds the lock (continuity) or the issue vanished mid-sweep. + // Otherwise returns the neutral-recovery descriptor: the marker run for + // provenance, plus the lock/owner values to CAS every later mutation against. + async function resolveCheckoutAdoptionHandover( + issue: Pick, + handoverMarkerRun: NonNullable, + ): Promise { + // BLO-19160 finding 2: resolve the adopter from the lock as it is NOW, not + // as the candidate snapshot saw it before this loop began. Following stale + // lock ids resolves the *previous* terminal owner while a live adopter holds + // the current lock — a narrow-window re-entry into the exact BLO-18860 + // failure mode this whole path exists to prevent. + const lockOwnerState = await readIssueLockOwnerState(issue.companyId, issue.id); + if (!lockOwnerState) return null; + + const adoptingRun = await getCheckoutAdoptingRun( + { companyId: issue.companyId, ...lockOwnerState }, + handoverMarkerRun, + ); + // Compare against the assignee, not the sweep's `agentId`: adoption is only + // ever performed by the assignee's own run (`adoptStaleCheckoutRun` requires + // `assigneeAgentId = actor`), and on an `in_review` issue `agentId` is the + // review participant instead. + // + // The adopter proves continuity only while it is LIVE. A terminal adopter + // tells us nothing about this issue (see the caller), and so does the + // successor-less case: `clearCheckoutRunIfTerminal` (services/issues.ts) + // nulls BOTH lock columns once the adopter goes terminal, leaving no id to + // resolve here — the ordinary cleanup sequence, not an anomaly. + if ( + adoptingRun && + adoptingRun.agentId === lockOwnerState.assigneeAgentId && + !isTerminalIssueRun(adoptingRun) + ) { + return null; + } + return { markerRunId: handoverMarkerRun.id, lockOwnerState }; + } + // Count the number of consecutive (most-recent-first) succeeded runs for // this issue whose livenessState is non-productive (plan_only, // empty_response, failed, or null). Stops counting at the first @@ -2326,7 +2440,14 @@ export function recoveryService( source: string; retryOfRunId?: string | null; extraContext?: Record; + // BLO-19160: when this recovery follows a checkout-handover observation, + // the lock/owner values it was decided on. If an adoption committed in the + // meantime a live run owns this issue now, and queuing a wake would put + // competing work against it. + expectedLockOwnerState?: IssueLockOwnerState | null; }) { + if (await issueLockOwnerStateChanged(input.issueId, input.expectedLockOwnerState)) return null; + const queued = await deps.enqueueWakeup(input.agentId, { source: "automation", triggerDetail: "system", @@ -2356,7 +2477,14 @@ export function recoveryService( return queued; } - async function enqueueInitialAssignedTodoDispatch(issue: typeof issues.$inferSelect, agentId: string) { + async function enqueueInitialAssignedTodoDispatch( + issue: typeof issues.$inferSelect, + agentId: string, + expectedLockOwnerState?: IssueLockOwnerState | null, + ) { + // BLO-19160: see `enqueueStrandedIssueRecovery` — on the handover path this + // dispatch must not race a freshly committed adoption. + if (await issueLockOwnerStateChanged(issue.id, expectedLockOwnerState)) return null; return deps.enqueueWakeup(agentId, { source: "assignment", triggerDetail: "system", @@ -5185,7 +5313,11 @@ export function recoveryService( issue: typeof issues.$inferSelect; previousStatus: StrandedPreviousStatus; latestRun: LatestIssueRun; + expectedLockOwnerState?: IssueLockOwnerState | null; }) { + // BLO-19160: see `escalateStrandedAssignedIssue` — a handover-derived + // escalation must not commit if the lock/owner moved under it. + if (await issueLockOwnerStateChanged(input.issue.id, input.expectedLockOwnerState)) return null; const updated = await issuesSvc.update(input.issue.id, { status: "blocked" }); if (!updated) return null; @@ -5254,6 +5386,7 @@ export function recoveryService( issue: typeof issues.$inferSelect; previousStatus: "in_progress"; latestRun: LatestIssueRun; + expectedLockOwnerState?: IssueLockOwnerState | null; }) { return await db.transaction(async (tx) => { await tx.execute( @@ -5267,6 +5400,19 @@ export function recoveryService( .limit(1); if (!fresh || fresh.status !== "in_progress" || !hasActiveMonitorPath(fresh)) return null; + // BLO-19160: parking to `in_review` is a status mutation on the handover + // path just as much as an escalation is, so it takes the same CAS. + if ( + input.expectedLockOwnerState && + !issueLockOwnerStateMatches(input.expectedLockOwnerState, { + executionRunId: fresh.executionRunId, + checkoutRunId: fresh.checkoutRunId, + assigneeAgentId: fresh.assigneeAgentId, + }) + ) { + return null; + } + const updated = await issuesSvc.update(fresh.id, { status: "in_review" }, tx); if (!updated) return null; @@ -5335,6 +5481,18 @@ export function recoveryService( if (fresh.status === "in_review") return "already_parked"; if (fresh.status !== "in_progress") return "failed"; + // BLO-19160: deliberately NO lock-owner CAS here, unlike the sibling + // parks/escalations. Two reasons, in order: + // 1. `ReviewWaitingParkOutcome` has no "took no action" variant, and a + // lost race must not map to "failed" — the caller treats "failed" as + // a genuine park failure and falls through to `blocked` escalation, + // i.e. exactly the clobber a CAS is supposed to prevent. Guarding + // here without a new outcome variant is worse than not guarding. + // 2. It is unreachable on the handover path anyway: the only call site + // is gated on `isWaitingOnReviewContinuationRun(latestRun)`, which + // requires `latestRun?.status === "cancelled"`, and the handover + // path sets `latestRun` to null. + // The in_review transition runs an evidence gate (issues.ts) that throws // `unprocessable` when the issue has no reviewable evidence yet (analysis-only // work: no PR, branch, or commits). Catch it so a single un-reviewable issue @@ -5536,7 +5694,16 @@ export function recoveryService( ); } - async function resolveContinuationWaitingOnReview(issue: typeof issues.$inferSelect) { + // BLO-19160: `expectedLockOwnerState` is the handover-observation CAS. This + // helper mutates the issue to `blocked` and writes blocker relations, so it + // is an escalation side effect exactly like `escalateStrandedAssignedIssue` + // even though it is not one of the four enqueue/escalate helpers — which is + // precisely how the original "all N call sites guarded" audit missed it. Any + // audit of this path has to enumerate *mutations*, not helper names. + async function resolveContinuationWaitingOnReview( + issue: typeof issues.$inferSelect, + expectedLockOwnerState?: IssueLockOwnerState | null, + ) { const existingBlockers = await existingUnresolvedBlockerIssues(issue.companyId, issue.id); const openChildren = await db .select({ id: issues.id, identifier: issues.identifier }) @@ -5592,6 +5759,10 @@ export function recoveryService( // `in_review` park instead of aborting the whole periodic recovery sweep. let updated: Awaited>; try { + // BLO-19160: re-check the handover lock/owner state immediately before + // the mutation. A live adopter committing after the handover observation + // must not be blocked out by this path. + if (await issueLockOwnerStateChanged(issue.id, expectedLockOwnerState)) return null; updated = await issuesSvc.update(issue.id, { status: "blocked", blockedByIssueIds }); } catch (error) { if (!isBlockingRelationCycleError(error)) throw error; @@ -5736,6 +5907,12 @@ export function recoveryService( executionRunId: string | null; }; successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null; + // BLO-19160: when this escalation follows a checkout-handover observation, + // the lock/owner values it was decided on. Re-checked below against the + // in-transaction re-read, before any side effect — so a *detected* race + // takes no escalation or reassignment side effect. It is NOT a + // mutation-time CAS; see the limitation note at that check. + expectedLockOwnerState?: IssueLockOwnerState | null; }) { // `isRoutineExecutionDuplicateSuppressedRun` is a type predicate, so the // negative branch below narrows `input.latestRun` all the way to `null`. @@ -5752,6 +5929,7 @@ export function recoveryService( issue: input.issue, previousStatus: input.previousStatus, latestRun: input.latestRun, + expectedLockOwnerState: input.expectedLockOwnerState, }); } @@ -5899,6 +6077,41 @@ export function recoveryService( return null; } + // BLO-19160: same shape as the status CAS above, for the lock/owner + // columns. An adoption that commits between the handover observation and + // this transaction means a live run holds this issue's execution lock; + // escalating on the evidence read before it would reassign the issue away + // from that run and revoke the assignee's write access — the BLO-18860 + // failure mode through a narrower window. Bail rather than clobber. + // + // LIMITATION, measured — this NARROWS the window, it does not close it. + // `adoptStaleCheckoutRun` (services/issues.ts) takes no advisory lock; it + // serializes on a `select … for update` ROW lock of this row, so the two + // paths share no mutual-exclusion primitive and an adoption committing + // after this comparison is not excluded. The two obvious fixes both + // DEADLOCK here and were reverted after being measured: + // * `fresh` → `.for("update")`, and/or + // * routing the mutation below through `tx` + // Either one hangs `issue-recovery-actions.test.ts` at the 60s test + // timeout, because helpers between the read and the write touch this same + // issue row on the pooled `db` connection and block on the tx's lock. + // Closing it properly means threading `tx` through those helpers, which is + // BLO-18829's scope (`Stranded-escalation side effects escape when the + // expectedStatus CAS loses the race` — the identical defect class for the + // status CAS directly above). Until then: a *detected* race is side-effect + // free, because this check precedes the action upsert, quota monitor and + // wake enqueue. + if ( + input.expectedLockOwnerState && + !issueLockOwnerStateMatches(input.expectedLockOwnerState, { + executionRunId: fresh.executionRunId, + checkoutRunId: fresh.checkoutRunId, + assigneeAgentId: fresh.assigneeAgentId, + }) + ) { + return null; + } + const recoveryCause = resolveStrandedRecoveryCause(input.latestRun, input.recoveryCause); const mutationDb = input.expectedReviewStage ? tx : db; const { action, hasNewActivitySinceLastAttempt } = await ensureSourceScopedStrandedRecoveryAction({ @@ -6725,7 +6938,7 @@ export function recoveryService( continue; } - let latestRun = await getLatestIssueRun(issue.companyId, issue.id); + const newestIssueRun = await getLatestIssueRun(issue.companyId, issue.id); // `issue_terminal_status` means this queued dispatch was correctly // cancelled while the issue was terminal. The candidate query above has // already established that the issue is non-terminal now, so this is @@ -6733,50 +6946,58 @@ export function recoveryService( // to keep skipping the issue forever. Clear it before classification, // but retain the flag so an `in_progress` issue reaches the normal // continuation re-dispatch below instead of the generic no-run skip. + let latestRun: LatestIssueRun = newestIssueRun; const reopenedAfterTerminalDispatchRace = isTerminalDispatchRaceRun(latestRun); if (reopenedAfterTerminalDispatchRace) latestRun = null; - // Set when this issue's newest run is a handover marker whose successor - // can no longer be identified. Distinguishes "adopted, then the lock was - // cleaned up" from "this issue genuinely has no run history at all" — - // the no-run/no-lock guard below must skip only the latter. - let adoptionHandoverLostSuccessor = false; // BLO-18860: never judge an issue on a checkout-adoption cancellation. - // The adopting run is by construction the assignee's own *live* run, so - // this issue has continuity, not a lost execution path — but the + // The adopting run is by construction the assignee's own run, so this + // issue has continuity, not a lost execution path — but the // `hasActiveExecutionPath` check above cannot see that run (it matches on // `contextSnapshot ->> 'issueId'`, and the adopting run is scoped to // whichever issue its own dispatch was for). Left unhandled, the handover // marker is the newest run row for this issue, reads as // terminal-unsuccessful, and escalates the issue away from the agent that - // had just written to it. Resolve the evidence to the adopting run - // instead: live same-assignee run → continuity, otherwise judge the issue - // on that run's real outcome so no escalation ever cites - // `issue_checkout_adopted` as its cause. - if (isCheckoutAdoptionCancelledRun(latestRun)) { - const adoptingRun = await getCheckoutAdoptingRun(issue, latestRun); - // Compare against the assignee, not `agentId`: adoption is only ever - // performed by the assignee's own run (`adoptStaleCheckoutRun` requires - // `assigneeAgentId = actor`), and on an `in_review` issue `agentId` is - // the review participant instead. - if ( - adoptingRun && - adoptingRun.agentId === issue.assigneeAgentId && - !isTerminalIssueRun(adoptingRun) - ) { + // had just written to it. + let adoptionHandover: AdoptionHandoverNeutralRecovery | null = null; + if (isCheckoutAdoptionCancelledRun(newestIssueRun)) { + adoptionHandover = await resolveCheckoutAdoptionHandover(issue, newestIssueRun); + // Continuity (a live same-assignee adopter holds the lock) or the issue + // vanished mid-sweep. Either way there is nothing to recover. + if (!adoptionHandover) { result.skipped += 1; continue; } - // No successor run resolvable: the adopter went terminal and - // `clearCheckoutRunIfTerminal` (services/issues.ts) nulled BOTH lock - // columns, so `getCheckoutAdoptingRun` has no id left to look up. That - // is the ordinary cleanup sequence, not an anomaly. Record it — the - // handover marker stays the newest run scoped to this issue forever, so - // without this flag the no-run/no-lock guard below would skip the issue - // on this sweep and on every sweep after it, stranding for good an - // issue whose only crime was being adopted once. - adoptionHandoverLostSuccessor = !adoptingRun; - latestRun = adoptingRun; } + // BLO-19160 finding 1: on the handover path this issue is judged with NO + // run evidence. The handover marker itself is bookkeeping about the run + // that lost the checkout, and the adopter — once terminal — is by + // construction scoped to a *different* issue, so its error code, + // workspace result, quota state, liveness and retry budget all describe + // that other issue's work. Substituting the adopter as `latestRun` (as + // this branch used to) let a foreign nonretryable outcome block, + // reassign, or suppress recovery on an issue for which the condition was + // never true. Carrying no evidence instead lands the issue on the neutral + // continuation recovery at the end of the loop: it is known to need a + // live execution path, and nothing more than that is known. + // + // Annotated (rather than inferred) because `isCheckoutAdoptionCancelledRun` + // is a type predicate: without the annotation TS narrows its *negative* + // branch to `null` too, and reads every run check below as unreachable. + if (adoptionHandover) latestRun = null; + // The marker stays the newest run scoped to this issue forever, so the + // no-run/no-lock guard below must not read the resulting absence of + // evidence as "nothing to recover from" and skip the issue on this sweep + // and every sweep after it. + const adoptionHandoverNeedsNeutralRecovery = Boolean(adoptionHandover); + // Provenance for that neutral recovery. The handover marker is the newest + // run genuinely scoped to THIS issue, which makes it the honest retry + // parent — unlike the lock columns, which on this path may name a run + // dispatched for someone else's issue. + const adoptionHandoverMarkerRunId = adoptionHandover?.markerRunId ?? null; + // Lock/owner values re-read at the instant the handover was observed. + // Non-null only on the handover path; every recovery mutation below CASes + // against it so a lost race takes no side effect. + const adoptionHandoverLockGuard = adoptionHandover?.lockOwnerState ?? null; const agent = await getAgent(agentId); const agentInvokable = agent && agent.companyId === issue.companyId ? await isAgentInvokable(agent) @@ -6887,6 +7108,7 @@ export function recoveryService( continue; } const updated = await escalateStrandedRecoveryIssueInPlace({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: issue.status as StrandedPreviousStatus, latestRun, @@ -6926,6 +7148,7 @@ export function recoveryService( continue; } else { const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: issue.status as StrandedPreviousStatus, latestRun, @@ -7015,7 +7238,7 @@ export function recoveryService( acceptedInteractionResolvedAt, ); if (consecutive >= INTERACTION_CONTINUATION_REQUEUE_MAX_ATTEMPTS && latestPostResolutionRun) { - const resolved = await resolveContinuationWaitingOnReview(issue); + const resolved = await resolveContinuationWaitingOnReview(issue, adoptionHandoverLockGuard); if (resolved) { result.waitingOnReviewResolved += 1; result.issueIds.push(issue.id); @@ -7023,6 +7246,7 @@ export function recoveryService( } const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: issue.status as StrandedPreviousStatus, latestRun: latestPostResolutionRun, @@ -7041,6 +7265,7 @@ export function recoveryService( } const queued = await enqueueStrandedIssueRecovery({ + expectedLockOwnerState: adoptionHandoverLockGuard, issueId: issue.id, agentId, reason: "issue_continuation_needed", @@ -7076,6 +7301,7 @@ export function recoveryService( if (!participantLatestRun || !isTerminalIssueRun(participantLatestRun)) { if (!agentInvokable) { const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_review", latestRun: participantLatestRun, @@ -7170,6 +7396,7 @@ export function recoveryService( } if (participantAdapterFailureClassification?.kind === "configuration_incomplete") { const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_review", latestRun: participantLatestRun, @@ -7195,6 +7422,7 @@ export function recoveryService( if (!agentInvokable) { const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_review", latestRun: participantLatestRun, @@ -7223,6 +7451,7 @@ export function recoveryService( if (didAutomaticRecoveryFail(participantLatestRun, EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON)) { const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_review", latestRun: participantLatestRun, @@ -7255,6 +7484,7 @@ export function recoveryService( } const queued = await enqueueStrandedIssueRecovery({ + expectedLockOwnerState: adoptionHandoverLockGuard, issueId: issue.id, agentId: participantAgentId, reason: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON, @@ -7290,7 +7520,7 @@ export function recoveryService( } const queued = await enqueueWithAssignmentRecoveryCapacity(issue, agentId, () => - enqueueInitialAssignedTodoDispatch(issue, agentId) + enqueueInitialAssignedTodoDispatch(issue, agentId, adoptionHandoverLockGuard) ); if (queued) { result.assignmentDispatched += 1; @@ -7338,6 +7568,7 @@ export function recoveryService( continue; } const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "todo", latestRun, @@ -7432,6 +7663,7 @@ export function recoveryService( } const failureSummary = summarizeRunFailureForIssueComment(latestRun); const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "todo", latestRun, @@ -7456,6 +7688,7 @@ export function recoveryService( const queued = await enqueueWithAssignmentRecoveryCapacity(issue, agentId, () => enqueueStrandedIssueRecovery({ + expectedLockOwnerState: adoptionHandoverLockGuard, issueId: issue.id, agentId, reason: "issue_assignment_recovery", @@ -7473,15 +7706,16 @@ export function recoveryService( continue; } - // No run evidence and no lock: nothing to recover from. A lost-successor - // handover is the exception — there the absence of both is the *result* - // of normal adoption cleanup, and the issue still needs a live path, so - // let it fall through to the continuation re-dispatch at the end. + // No run evidence and no lock: nothing to recover from. A handover marker + // is the exception — there the absence of usable evidence is deliberate + // (BLO-19160) or the result of normal adoption cleanup, and the issue + // still needs a live path, so let it fall through to the continuation + // re-dispatch at the end. if ( !latestRun && !issue.checkoutRunId && !issue.executionRunId && - !adoptionHandoverLostSuccessor && + !adoptionHandoverNeedsNeutralRecovery && !reopenedAfterTerminalDispatchRace ) { result.skipped += 1; @@ -7495,6 +7729,7 @@ export function recoveryService( } const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_progress", latestRun, @@ -7539,6 +7774,7 @@ export function recoveryService( ); if (!exempted) { const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_progress", latestRun: successfulRun, @@ -7561,6 +7797,7 @@ export function recoveryService( continue; } const queued = await enqueueStrandedIssueRecovery({ + expectedLockOwnerState: adoptionHandoverLockGuard, issueId: issue.id, agentId, reason: "issue_continuation_needed", @@ -7596,6 +7833,7 @@ export function recoveryService( ); if (nonProductiveStreak >= NON_PRODUCTIVE_RUN_NOOP_THRESHOLD) { const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_progress", latestRun, @@ -7631,6 +7869,7 @@ export function recoveryService( continue; } const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_progress", latestRun, @@ -7725,7 +7964,7 @@ export function recoveryService( const classification = classifyContinuationFailure(latestRun); if (classification.errorCode === CONTINUATION_WAITING_ON_REVIEW_ERROR_CODE) { - const resolved = await resolveContinuationWaitingOnReview(issue); + const resolved = await resolveContinuationWaitingOnReview(issue, adoptionHandoverLockGuard); if (resolved) { result.waitingOnReviewResolved += 1; result.issueIds.push(issue.id); @@ -7816,6 +8055,7 @@ export function recoveryService( if (classification.kind === "non_retryable") { const failureSummary = summarizeRunFailureForIssueComment(latestRun); const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_progress", latestRun, @@ -7851,6 +8091,7 @@ export function recoveryService( ? ` Latest cause: \`${classification.errorCode}\`.` : ""; const updated = await escalateStrandedAssignedIssue({ + expectedLockOwnerState: adoptionHandoverLockGuard, issue, previousStatus: "in_progress", latestRun, @@ -7891,7 +8132,11 @@ export function recoveryService( reason: "issue_continuation_needed", retryReason: "issue_continuation_needed", source: "issue.continuation_recovery", - retryOfRunId: latestRun?.id ?? issue.checkoutRunId ?? null, + // BLO-19160: prefer the handover marker over the lock columns for + // provenance — on the handover path the lock may name a run dispatched + // for a different issue, while the marker is scoped to this one. + retryOfRunId: latestRun?.id ?? adoptionHandoverMarkerRunId ?? issue.checkoutRunId ?? null, + expectedLockOwnerState: adoptionHandoverLockGuard, }); if (queued) { result.continuationRequeued += 1; From df89ec68eac58608103bcbaa25495ef72e8b6345 Mon Sep 17 00:00:00 2001 From: kkroo Date: Tue, 25 Aug 2026 06:16:39 +0000 Subject: [PATCH 2/3] fix(recovery): serialize checkout handover mutations --- .../heartbeat-wake-dispatch-retry.test.ts | 58 + ...sue-checkout-routine-lock-conflict.test.ts | 44 +- .../__tests__/issue-recovery-actions.test.ts | 113 +- server/src/services/heartbeat.ts | 21 +- server/src/services/issue-checkout-status.ts | 43 + server/src/services/issue-recovery-actions.ts | 14 +- server/src/services/issues.ts | 519 ++++---- server/src/services/recovery/service.ts | 1056 ++++++++++------- 8 files changed, 1181 insertions(+), 687 deletions(-) diff --git a/server/src/__tests__/heartbeat-wake-dispatch-retry.test.ts b/server/src/__tests__/heartbeat-wake-dispatch-retry.test.ts index ae51ea7f5f8c..71a41d51468a 100644 --- a/server/src/__tests__/heartbeat-wake-dispatch-retry.test.ts +++ b/server/src/__tests__/heartbeat-wake-dispatch-retry.test.ts @@ -188,6 +188,64 @@ describeEmbeddedPostgres("heartbeat wake dispatch retry (BLO-14395)", () => { expect(dispatchFailedRows).toHaveLength(0); }); + it("durably skips a recovery wake when issue ownership changes before enqueueWakeup locks it", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const issueId = randomUUID(); + const expectedLockOwnerState = { + executionRunId: null, + checkoutRunId: null, + assigneeAgentId: agentId, + }; + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Ownership changes before recovery wake", + status: "todo", + priority: "high", + assigneeAgentId: agentId, + checkoutRunId: null, + executionRunId: null, + }); + + const racingHeartbeat = heartbeatService(db, { + skipQueuedRunDispatch: true, + beforeIssueWakeLockForTest: async ({ issueId: lockingIssueId }) => { + expect(lockingIssueId).toBe(issueId); + await db + .update(issues) + .set({ assigneeAgentId: null, updatedAt: new Date() }) + .where(eq(issues.id, issueId)); + }, + }); + + const run = await racingHeartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_assignment_recovery", + payload: { issueId }, + contextSnapshot: { issueId, taskId: issueId, wakeReason: "issue_assignment_recovery" }, + expectedLockOwnerState, + }); + + expect(run).toBeNull(); + const skippedRows = await db + .select({ reason: agentWakeupRequests.reason }) + .from(agentWakeupRequests) + .where(and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.agentId, agentId), + eq(agentWakeupRequests.status, "skipped"), + )); + expect(skippedRows).toEqual([{ reason: "issue_execution_ownership_changed" }]); + + const issueRuns = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId))); + expect(issueRuns).toHaveLength(0); + }); + it("passes a business-rule HttpError straight through with no retry delay and no durable dispatch_failed record", async () => { const { agentId } = await seedCompanyAndAgent({ agentStatus: "paused" }); diff --git a/server/src/__tests__/issue-checkout-routine-lock-conflict.test.ts b/server/src/__tests__/issue-checkout-routine-lock-conflict.test.ts index e5389f29920e..fc8baedb04ed 100644 --- a/server/src/__tests__/issue-checkout-routine-lock-conflict.test.ts +++ b/server/src/__tests__/issue-checkout-routine-lock-conflict.test.ts @@ -251,13 +251,12 @@ describeEmbeddedPostgres("checkout adoption vs open routine-execution lock (PEN- expect(victim?.executionRunId).toBe(actorRunId); }); - it("records what the inline-refresh path actually leaves behind when it loses the key", async () => { - // The other three cases all conflict inside a single transaction. This one - // does not, and that is the point: `clearStaleExecutionLock` COMMITS the - // reap of the victim's own dead lock, and only then does the separate - // refresh write raise 23505. So the victim's row does change even though the - // call fails — "not half-applied" is true of the unowned-adoption path and - // false here. + it("rolls back stale cleanup when the atomic adoption loses the key", async () => { + // Stale cleanup and adoption share one ownership transaction now. The + // trigger below attempts to let the sibling take the unique key at the + // exact point where the victim's dead lock would be cleared. The unique + // violation must abort that whole transaction: neither the cleanup nor the + // sibling's competing lock may survive the failed adoption. // // Both rows cannot hold the key at seed time (the partial index forbids it), // so the sibling's acquisition is injected with a trigger that fires exactly @@ -268,11 +267,11 @@ describeEmbeddedPostgres("checkout adoption vs open routine-execution lock (PEN- const routineId = randomUUID(); const fingerprint = "shared-dispatch-fingerprint"; - const deadRunId = await seedRun(companyId, agentId, "failed"); - await db - .update(heartbeatRuns) - .set({ finishedAt: new Date() }) - .where(eq(heartbeatRuns.id, deadRunId)); + // `queued` is reapable by stale-lock adoption but is not terminal, so the + // initial terminal-cleanup prepass leaves it in place. That puts the + // trigger inside the atomic stale-cleanup-plus-adoption transaction rather + // than in the earlier standalone cleanup transaction. + const deadRunId = await seedRun(companyId, agentId, "queued"); // The victim holds the key via its OWN dead run, so it is the row the // inline-refresh path reaps. @@ -349,18 +348,17 @@ describeEmbeddedPostgres("checkout adoption vs open routine-execution lock (PEN- details?: { ownerIssueId?: string | null }; }); - // Still a 409, not the 500 this PR removes — and specifically the - // routine-lock 409, so the assertions below cannot be satisfied by some - // other conflict path that never reached the inline refresh. + // Still a 409, not the 500 this PR removes. The sibling write happened + // inside the same transaction and therefore rolled back too, so there is + // no committed owner to name. expect(error?.status).toBe(409); - expect(error?.message).toBe("Routine execution already locked by another open issue"); - expect(error?.details?.ownerIssueId).toBe(ownerIssueId); + expect(error?.message).toBe("Routine execution dispatch lock is contended; retry the request"); + expect(error?.details?.ownerIssueId).toBeNull(); const victim = await readIssueLockState(victimIssueId); - // The committed reap survives the failed refresh. This is the residual - // state, asserted rather than assumed: the dead run's lock is gone and the - // victim now holds nothing. - expect(victim?.executionRunId, "the reap already committed").toBeNull(); + // The stale lock is still present because its cleanup rolled back with the + // failed adoption. A later ordinary cleanup can safely reap it. + expect(victim?.executionRunId, "stale cleanup must roll back").toBe(deadRunId); expect(victim?.checkoutRunId).toBeNull(); expect(victim?.status).toBe("in_progress"); @@ -372,10 +370,10 @@ describeEmbeddedPostgres("checkout adoption vs open routine-execution lock (PEN- .from(heartbeatRuns) .where(eq(heartbeatRuns.id, deadRunId)) .then((rows) => rows[0] ?? null); - expect(deadRun?.status).toBe("failed"); + expect(deadRun?.status).toBe("queued"); const owner = await readIssueLockState(ownerIssueId); - expect(owner?.executionRunId, "the sibling holds the key it took").toBe(ownerRunId); + expect(owner?.executionRunId, "the competing sibling write must roll back").toBeNull(); } finally { await db.execute(sql`drop trigger if exists pen2395_take_lock_on_release on issues`); await db.execute(sql`drop function if exists pen2395_take_lock_on_release()`); diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 97b21d5e1a94..48ecc05c68f4 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -566,6 +566,89 @@ describeEmbeddedPostgres("issue recovery actions", () => { expect(await svc.getActiveForIssue(randomUUID(), sourceIssueId)).toBeNull(); }); + it("does not refund a stale wake reservation after ownership or attempt changes", async () => { + const { companyId, managerId, coderId, sourceIssueId } = await seedCompany(); + const svc = issueRecoveryActionService(db); + const action = await svc.upsertSourceScoped({ + companyId, + sourceIssueId, + kind: "stranded_assigned_issue", + ownerType: "agent", + ownerAgentId: managerId, + cause: "stranded_assigned_issue", + fingerprint: "stale-refund:fingerprint", + nextAction: "Restore a live execution path.", + maxAttempts: 5, + }); + + // A replacement owner can start a fresh sequence at the same count. The old + // failed wake must not debit that replacement sequence. + await db + .update(issueRecoveryActions) + .set({ ownerAgentId: coderId, attemptCount: 1 }) + .where(eq(issueRecoveryActions.id, action.id)); + await svc.releaseWakeAttempt({ + companyId, + actionId: action.id, + expectedOwnerAgentId: managerId, + expectedAttemptCount: action.attemptCount, + }); + + let [current] = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, action.id)); + expect(current).toMatchObject({ ownerAgentId: coderId, attemptCount: 1 }); + + // The owner can also remain the same while a newer reservation advances + // the counter. Its refund must not decrement the newer reservation. + await db + .update(issueRecoveryActions) + .set({ ownerAgentId: managerId, attemptCount: action.attemptCount + 1 }) + .where(eq(issueRecoveryActions.id, action.id)); + await svc.releaseWakeAttempt({ + companyId, + actionId: action.id, + expectedOwnerAgentId: managerId, + expectedAttemptCount: action.attemptCount, + }); + + [current] = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, action.id)); + expect(current).toMatchObject({ ownerAgentId: managerId, attemptCount: action.attemptCount + 1 }); + }); + + it("refunds a wake reservation when owner and reserved attempt still match", async () => { + const { companyId, managerId, sourceIssueId } = await seedCompany(); + const svc = issueRecoveryActionService(db); + const action = await svc.upsertSourceScoped({ + companyId, + sourceIssueId, + kind: "stranded_assigned_issue", + ownerType: "agent", + ownerAgentId: managerId, + cause: "stranded_assigned_issue", + fingerprint: "matching-refund:fingerprint", + nextAction: "Restore a live execution path.", + maxAttempts: 5, + }); + + await svc.releaseWakeAttempt({ + companyId, + actionId: action.id, + expectedOwnerAgentId: managerId, + expectedAttemptCount: action.attemptCount, + }); + + const [current] = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, action.id)); + expect(current).toMatchObject({ ownerAgentId: managerId, attemptCount: 0 }); + }); + it.each([ ["job_missing", "in_progress"], ["job_missing", "todo"], @@ -1311,12 +1394,9 @@ describeEmbeddedPostgres("issue recovery actions", () => { expect(finalIssue?.assigneeAgentId).toBe(newOwnerId); }); - // BLO-20933 (review finding 2): `expectedCurrentAssigneeAgentId` is enforced by a THROWN - // 409, not a falsy return, and it fires after the action upsert and the owner wake have - // already committed. Unhandled, that throw escapes `reconcileStrandedAssignedIssues`'s - // per-issue loop (which has no try/catch) and leaves every remaining stranded issue in - // the batch unreconciled. The lost race must degrade to a skip for this one issue. - it("skips instead of throwing when the assignee changes mid-escalation", async () => { + // Recovery dispatch is post-commit. A reassignment performed by the injected + // wake therefore cannot race the transactional issue UPDATE or be overwritten. + it("keeps a post-commit reassignment from being overwritten by recovery", async () => { const { companyId, managerId, coderId, sourceIssue } = await seedCompany(); const raceWinnerId = randomUUID(); await db.insert(agents).values({ @@ -1331,8 +1411,8 @@ describeEmbeddedPostgres("issue recovery actions", () => { runtimeConfig: {}, permissions: {}, }); - // The injected wake runs after the lock-fresh read and before the source-issue UPDATE, - // so mutating the assignee here reproduces the lost race deterministically. + // The injected wake runs only after the transaction has committed, matching + // the production dispatch ordering. const enqueueWakeup = vi.fn< (agentId: string, opts?: { payload?: unknown }) => Promise<{ id: string }> >(async () => { @@ -1360,10 +1440,12 @@ describeEmbeddedPostgres("issue recovery actions", () => { comment: "Automatic continuation recovery failed.", })).resolves.not.toThrow(); + expect(enqueueWakeup).toHaveBeenCalledTimes(1); // The reassignment stands; recovery did not clobber it back to the manager. const [finalIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssue.id)); expect(finalIssue?.assigneeAgentId).toBe(raceWinnerId); - expect(finalIssue?.status).toBe("in_progress"); + expect(finalIssue?.status).toBe("blocked"); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(1); }); // BLO-20933: the regex used to also match the bare words `eviction`, `preempt(ion|ed)`, @@ -3915,13 +3997,14 @@ describeEmbeddedPostgres("issue recovery actions", () => { expect(enqueueWakeup).not.toHaveBeenCalled(); }); - it("keeps the source issue blocked when source-scoped wakeup is claimed synchronously", async () => { + it("lets a synchronously claimed source-scoped wake reopen the source issue", async () => { const { companyId, managerId, coderId, sourceIssue } = await seedCompany(); await db.update(agents).set({ status: "paused" }).where(eq(agents.id, managerId)); // The wake is CLAIMED here — the fixture picks the issue up synchronously — so it is a - // delivered wake and must return the queued run. Returning null would model a - // non-delivery, which is refunded and spends no budget (BLO-18996 follow-up), and the - // `attemptCount: 2` below would then read 0. + // delivered wake and must return the queued run. Recovery commits the blocked transition + // before dispatching this post-commit wake; the claimed wake then legitimately reopens the + // source issue for execution. Returning null would model a non-delivery, which is refunded + // and spends no budget (BLO-18996 follow-up), and the `attemptCount: 2` below would then read 0. const enqueueWakeup = vi.fn(async () => { await db .update(issues) @@ -3951,7 +4034,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { }); const [afterFirst] = await db.select().from(issues).where(eq(issues.id, sourceIssue.id)); - expect(afterFirst?.status).toBe("blocked"); + expect(afterFirst?.status).toBe("in_progress"); expect(afterFirst?.assigneeAgentId).toBe(coderId); const secondLatestRun = { @@ -3980,7 +4063,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { attemptCount: 2, }); const [afterSecond] = await db.select().from(issues).where(eq(issues.id, sourceIssue.id)); - expect(afterSecond?.status).toBe("blocked"); + expect(afterSecond?.status).toBe("in_progress"); const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, sourceIssue.id)); expect(comments).toHaveLength(1); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 43d65e32870a..7a8d8e608633 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -225,9 +225,12 @@ import { TERMINAL_HEARTBEAT_RUN_STATUSES, } from "./issue-execution-lock.js"; import { + issueLockOwnerStateMatches, + lockIssueOwnership, releaseIssueRunOwnership, restoreCheckoutPromotedStatus, restoreCheckoutPromotedStatuses, + type IssueLockOwnerState, } from "./issue-checkout-status.js"; import { resolveStaleDependabotAlertWakeIssue } from "./dependabot-alert-issues.js"; import { createToolGatewayService } from "./tool-gateway.js"; @@ -3677,6 +3680,7 @@ interface WakeupOptions { contextSnapshot?: Record; retryOfRunId?: string | null; scheduledRetryAttempt?: number; + expectedLockOwnerState?: IssueLockOwnerState | null; } type UsageTotals = { @@ -28278,8 +28282,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const writeSkippedRequest = async ( skipReason: string, patch: Partial = {}, + dbOrTx: Db | DbTransaction = db, ) => { - await db.insert(agentWakeupRequests).values({ + await dbOrTx.insert(agentWakeupRequests).values({ companyId: agent.companyId, agentId, source, @@ -28820,6 +28825,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) await options.beforeIssueWakeLockForTest?.({ issueId, agentId }); const outcome = await db.transaction(async (tx) => { + await lockIssueOwnership(tx, agent.companyId, issueId); await tx.execute( sql`select id from issues where id = ${issueId} and company_id = ${agent.companyId} for update`, ); @@ -28839,6 +28845,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) // BLO-19063: needed to resolve the same workspaceStrategy overlay the // scheduling path applies, so both agree on `per_run`. assigneeAdapterOverrides: issues.assigneeAdapterOverrides, + checkoutRunId: issues.checkoutRunId, executionRunId: issues.executionRunId, executionAgentNameKey: issues.executionAgentNameKey, createdAt: issues.createdAt, @@ -28864,6 +28871,18 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return { kind: "skipped" as const }; } + if ( + opts.expectedLockOwnerState && + !issueLockOwnerStateMatches(opts.expectedLockOwnerState, { + executionRunId: issue.executionRunId, + checkoutRunId: issue.checkoutRunId, + assigneeAgentId: issue.assigneeAgentId, + }) + ) { + await writeSkippedRequest("issue_execution_ownership_changed", {}, tx); + return { kind: "skipped" as const }; + } + if (worktreeExecutionCutoff && issue.createdAt < worktreeExecutionCutoff) { await tx.insert(agentWakeupRequests).values({ companyId: agent.companyId, diff --git a/server/src/services/issue-checkout-status.ts b/server/src/services/issue-checkout-status.ts index 1fd33b0e5834..4f348404507f 100644 --- a/server/src/services/issue-checkout-status.ts +++ b/server/src/services/issue-checkout-status.ts @@ -5,6 +5,49 @@ import { TERMINAL_HEARTBEAT_RUN_STATUS_VALUES } from "./issue-execution-lock.js" type DbTransaction = Parameters[0]>[0]; type DbOrTransaction = Db | DbTransaction; +/** + * The issue columns that jointly identify the current execution owner. + * + * Recovery observations carry this snapshot across a few awaits. Consumers + * must compare all three fields after taking the issue ownership lock; a + * partial comparison can accept an adopter that changed only one half of the + * execution lock pair. + */ +export type IssueLockOwnerState = { + executionRunId: string | null; + checkoutRunId: string | null; + assigneeAgentId: string | null; +}; + +export function issueLockOwnerStateMatches( + expected: IssueLockOwnerState, + actual: IssueLockOwnerState, +): boolean { + return expected.executionRunId === actual.executionRunId && + expected.checkoutRunId === actual.checkoutRunId && + expected.assigneeAgentId === actual.assigneeAgentId; +} + +/** + * Serialize every mutation that can transfer an issue's execution ownership. + * + * The recovery sweep and checkout adoption both perform several reads and + * writes on pooled connections. A row lock alone cannot coordinate those + * paths: recovery deliberately keeps its transaction open while its dependent + * side effects use other connections. Keep this key in one helper so both + * paths acquire the same transaction-scoped lock before taking the issue row + * lock or making an ownership decision. + */ +export async function lockIssueOwnership( + dbOrTx: Pick, + companyId: string, + issueId: string, +): Promise { + await dbOrTx.execute( + sql`select pg_advisory_xact_lock(hashtextextended(${companyId} || ':' || ${issueId}, 0))`, + ); +} + /** * Clear every issue-ownership column that still points at a terminalizing run. * diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index d2862f67a0f4..694280f5d389 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -610,8 +610,16 @@ export function issueRecoveryActionService(db: DbOrTransaction) { // reports its non-delivery paths (capacity deferral, tree hold, cooldown, disabled wake). // So only wakes that actually reached the queue count against the budget. Floors at 0 so a // refunded first attempt makes the next sweep's `existing.attemptCount + 1` land back on 1. - // Scoped to active statuses and matched on company so it cannot touch a resolved row. - async function releaseWakeAttempt(input: { companyId: string; actionId: string }): Promise { + // The owner and attempt count are the reservation token captured by the caller. Keep both + // in the UPDATE predicate so a refund from an older wake is an atomic no-op after ownership + // changes or a newer reservation. Scoped to active statuses and matched on company so it + // cannot touch a resolved row. + async function releaseWakeAttempt(input: { + companyId: string; + actionId: string; + expectedOwnerAgentId: string; + expectedAttemptCount: number; + }): Promise { await db .update(issueRecoveryActions) .set({ @@ -622,6 +630,8 @@ export function issueRecoveryActionService(db: DbOrTransaction) { and( eq(issueRecoveryActions.id, input.actionId), eq(issueRecoveryActions.companyId, input.companyId), + eq(issueRecoveryActions.ownerAgentId, input.expectedOwnerAgentId), + eq(issueRecoveryActions.attemptCount, input.expectedAttemptCount), inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]), ), ); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 5f4f430e89b9..335c3e248592 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -76,6 +76,7 @@ import { conflict, HttpError, notFound, unprocessable } from "../errors.js"; import { incrementBlockerResolvedWakeMetric } from "./blocker-resolved-wake-metrics.js"; import { checkoutRestoreStatusExpression, + lockIssueOwnership, restoreCheckoutPromotedStatus, } from "./issue-checkout-status.js"; import { logger } from "../middleware/logger.js"; @@ -950,6 +951,23 @@ type IssueUserContextInput = { type ProjectGoalReader = Pick; type DbReader = Pick; type DbTransaction = Parameters[0]>[0]; + +/** + * Serialize mutations to the issue parent/blocker graph for one company. + * + * Graph writers must take this advisory lock before locking any issue rows so + * parent, blocker, and combined updates share one lock order. Recovery uses + * the same boundary while it derives a blocker set that it will immediately + * persist. + */ +export async function lockIssueParentMutationCompany( + companyId: string, + dbOrTx: Pick, +): Promise { + await dbOrTx.execute( + sql`select pg_advisory_xact_lock(hashtextextended(${`paperclip:issue-parent:${companyId}`}, 0))`, + ); +} type IssueCreateInput = Omit & { labelIds?: string[]; blockedByIssueIds?: string[]; @@ -5172,6 +5190,10 @@ export function issueService(db: Db) { ) => Promise, ) { return db.transaction(async (tx) => { + // Ownership arbitration must precede every row/relationship lock. The + // recovery sweep holds this same transaction-scoped lock while it + // performs its dependent mutations on pooled connections. + await lockIssueOwnership(tx, companyId, issueId); await lockIssueBlockerRelations(tx, companyId, issueId); const currentBlockerIssueIds = await tx .select({ id: issueRelations.issueId }) @@ -5745,12 +5767,6 @@ export function issueService(db: Db) { } } - async function lockIssueParentMutationCompany(companyId: string, dbOrTx: any = db) { - await dbOrTx.execute( - sql`select pg_advisory_xact_lock(hashtextextended(${`paperclip:issue-parent:${companyId}`}, 0))`, - ); - } - async function assertValidIssueParent( companyId: string, issueId: string, @@ -6209,6 +6225,7 @@ export function issueService(db: Db) { }))) return null; return db.transaction(async (tx) => { + await lockIssueOwnership(tx, input.companyId, input.issueId); const lockedIssue = await tx .select({ id: issues.id, @@ -6281,11 +6298,13 @@ export function issueService(db: Db) { async function adoptStaleCheckoutRun(input: { issueId: string; + companyId: string; actorAgentId: string; actorRunId: string; expectedCheckoutRunId: string; }) { const result = await db.transaction(async (tx) => { + await lockIssueOwnership(tx, input.companyId, input.issueId); const lockedIssue = await tx .select({ id: issues.id, @@ -6441,10 +6460,12 @@ export function issueService(db: Db) { async function adoptUnownedCheckoutRun(input: { issueId: string; + companyId: string; actorAgentId: string; actorRunId: string; }) { const adopted = await db.transaction(async (tx) => { + await lockIssueOwnership(tx, input.companyId, input.issueId); const lockedIssue = await tx .select({ id: issues.id, @@ -6620,139 +6641,281 @@ export function issueService(db: Db) { }; } - async function clearExecutionRunIfTerminal(issueId: string): Promise { + type TerminalIssueRunCleanupMode = "execution" | "checkout" | "both"; + + async function withIssueOwnershipTransaction( + issueId: string, + onMissing: T, + operation: (tx: DbTransaction, companyId: string) => Promise, + ): Promise { return db.transaction(async (tx) => { - await tx.execute( - sql`select ${issues.id} from ${issues} where ${issues.id} = ${issueId} for update`, - ); - const issue = await tx - .select({ executionRunId: issues.executionRunId, companyId: issues.companyId }) + // Read the immutable company scope before taking the row lock. Every + // ownership writer acquires the advisory key before its issue row lock; + // doing the same here prevents the cleanup/adoption deadlock where one + // path holds the row while waiting for the key and the other holds the + // key while waiting for the row. + const [identity] = await tx + .select({ companyId: issues.companyId }) .from(issues) .where(eq(issues.id, issueId)) - .then((rows) => rows[0] ?? null); - if (!issue?.executionRunId) return false; + .limit(1); + if (!identity) return onMissing; + await lockIssueOwnership(tx, identity.companyId, issueId); + return operation(tx, identity.companyId); + }); + } + + /** + * Clear terminal ownership while the issue arbitration lock is held. + * + * `checkout` cleanup deliberately clears the bundled execution columns too, + * preserving the historical all-or-nothing behavior of + * clearCheckoutRunIfTerminal. `both` is the old execution-then-checkout pair + * evaluated in one transaction, so a competing adopter cannot commit in the + * gap between the two cleanup statements. + */ + async function clearTerminalIssueRunLocksInTx( + tx: DbTransaction, + issueId: string, + mode: TerminalIssueRunCleanupMode, + companyId: string, + ): Promise { + const [issue] = await tx + .select({ + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + companyId: issues.companyId, + }) + .from(issues) + .where(and(eq(issues.id, issueId), eq(issues.companyId, companyId))) + .limit(1) + .for("update"); + if (!issue) return false; + + const ownerRunIds = [...new Set([ + issue.checkoutRunId, + issue.executionRunId, + ].filter((runId): runId is string => Boolean(runId)))].sort(); + const ownerRuns = ownerRunIds.length > 0 + ? await tx + .select({ id: heartbeatRuns.id, status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(inArray(heartbeatRuns.id, ownerRunIds)) + .orderBy(asc(heartbeatRuns.id)) + .for("update") + : []; + const runById = new Map(ownerRuns.map((run) => [run.id, run])); + const executionRun = issue.executionRunId + ? runById.get(issue.executionRunId) ?? null + : null; + const checkoutRun = issue.checkoutRunId + ? runById.get(issue.checkoutRunId) ?? null + : null; + const executionTerminal = Boolean(issue.executionRunId) && ( + !executionRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(executionRun.status) + ); + const checkoutTerminal = Boolean(issue.checkoutRunId) && ( + !checkoutRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(checkoutRun.status) + ); - await tx.execute( - sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${issue.executionRunId} for update`, + const clearExecution = (mode === "execution" || mode === "both") && executionTerminal; + const clearCheckout = (mode === "checkout" || mode === "both") && + checkoutTerminal && + (!issue.executionRunId || issue.executionRunId === issue.checkoutRunId || executionTerminal); + // A checkout cleanup historically clears the execution columns as part of + // the same mutation, even when executionRunId is already NULL. + const clearExecutionColumns = clearExecution || clearCheckout; + if (!clearExecutionColumns && !clearCheckout) return false; + + const patch: Partial = { updatedAt: new Date() }; + if (clearCheckout) patch.checkoutRunId = null; + if (clearExecutionColumns) { + patch.executionRunId = null; + patch.executionAgentNameKey = null; + patch.executionLockedAt = null; + } + + const conditions = [eq(issues.id, issueId), eq(issues.companyId, companyId)]; + if (clearCheckout) { + conditions.push( + issue.checkoutRunId + ? eq(issues.checkoutRunId, issue.checkoutRunId) + : isNull(issues.checkoutRunId), ); - const run = await tx - .select({ status: heartbeatRuns.status }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, issue.executionRunId)) - .then((rows) => rows[0] ?? null); - if (run && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(run.status)) return false; + } + if (clearExecutionColumns) { + conditions.push( + issue.executionRunId + ? eq(issues.executionRunId, issue.executionRunId) + : isNull(issues.executionRunId), + ); + } - const updated = await tx - .update(issues) - .set({ - executionRunId: null, - executionAgentNameKey: null, - executionLockedAt: null, - updatedAt: new Date(), - }) - .where( - and( - eq(issues.id, issueId), - eq(issues.executionRunId, issue.executionRunId), - ), - ) - .returning({ id: issues.id }) - .then((rows) => rows[0] ?? null); + const updated = await tx + .update(issues) + .set(patch) + .where(and(...conditions)) + .returning({ id: issues.id }) + .then((rows) => rows[0] ?? null); + if (updated) { + await restoreCheckoutPromotedStatus(tx, { issueId, companyId }); + } + return Boolean(updated); + } - if (updated) { - await restoreCheckoutPromotedStatus(tx, { issueId, companyId: issue.companyId }); - } + async function clearTerminalIssueRunLocks( + issueId: string, + mode: TerminalIssueRunCleanupMode, + ): Promise { + return withIssueOwnershipTransaction(issueId, false, (tx, companyId) => + clearTerminalIssueRunLocksInTx(tx, issueId, mode, companyId)); + } - return Boolean(updated); - }); + async function clearExecutionRunIfTerminal(issueId: string): Promise { + return clearTerminalIssueRunLocks(issueId, "execution"); } - async function clearStaleExecutionLock(input: { + // Used by checkout/ownership assertions that historically called the two + // cleanup helpers back-to-back. Keep the pair in one transaction so an + // adopter cannot commit between the execution and checkout cleanup passes. + async function clearTerminalIssueRunLocksPair(issueId: string): Promise { + return clearTerminalIssueRunLocks(issueId, "both"); + } + + type StaleExecutionLockInput = { issueId: string; + companyId: string; expectedCheckoutRunId: string | null; expectedExecutionRunId: string; actorRunId: string | null; - }) { + }; + + async function clearStaleExecutionLockInTx( + tx: DbTransaction, + input: StaleExecutionLockInput, + ): Promise { // BLO-20321: reap never-started (`queued` / `scheduled_retry`) owners as well // as terminal ones. Callers re-acquire the lock and then run // cancelStaleIssueContextRuns(keepRunId: ), which cancels the // superseded run — so it cannot start later against a status the assignee has // since changed. - return db.transaction(async (tx) => { - const issue = await tx - .select({ - checkoutRunId: issues.checkoutRunId, - executionRunId: issues.executionRunId, - }) - .from(issues) - .where(eq(issues.id, input.issueId)) - .for("update") - .then((rows) => rows[0] ?? null); - if ( - issue?.executionRunId !== input.expectedExecutionRunId || - issue.checkoutRunId !== input.expectedCheckoutRunId - ) return false; - - const ownerRunIds = [...new Set([ - input.expectedExecutionRunId, - input.expectedCheckoutRunId, - ].filter((runId): runId is string => Boolean(runId)))].sort(); - const ownerRuns = await tx - .select({ - id: heartbeatRuns.id, - status: heartbeatRuns.status, - startedAt: heartbeatRuns.startedAt, - wakeupRequestId: heartbeatRuns.wakeupRequestId, - }) - .from(heartbeatRuns) - .where(inArray(heartbeatRuns.id, ownerRunIds)) - .orderBy(asc(heartbeatRuns.id)) - .for("update") - const ownerRunById = new Map(ownerRuns.map((run) => [run.id, run])); - const executionOwnerRun = ownerRunById.get(input.expectedExecutionRunId) ?? null; - const distinctCheckoutOwnerId = input.expectedCheckoutRunId !== null && - input.expectedCheckoutRunId !== input.actorRunId && - input.expectedCheckoutRunId !== input.expectedExecutionRunId - ? input.expectedCheckoutRunId - : null; - const distinctCheckoutOwnerRun = distinctCheckoutOwnerId - ? ownerRunById.get(distinctCheckoutOwnerId) ?? null + const [issue] = await tx + .select({ + companyId: issues.companyId, + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))) + .for("update") + .limit(1); + if ( + !issue || + issue.executionRunId !== input.expectedExecutionRunId || + issue.checkoutRunId !== input.expectedCheckoutRunId + ) return false; + + const ownerRunIds = [...new Set([ + input.expectedExecutionRunId, + input.expectedCheckoutRunId, + ].filter((runId): runId is string => Boolean(runId)))].sort(); + const ownerRuns = await tx + .select({ + id: heartbeatRuns.id, + status: heartbeatRuns.status, + startedAt: heartbeatRuns.startedAt, + wakeupRequestId: heartbeatRuns.wakeupRequestId, + }) + .from(heartbeatRuns) + .where(inArray(heartbeatRuns.id, ownerRunIds)) + .orderBy(asc(heartbeatRuns.id)) + .for("update"); + const ownerRunById = new Map(ownerRuns.map((run) => [run.id, run])); + const executionOwnerRun = ownerRunById.get(input.expectedExecutionRunId) ?? null; + const distinctCheckoutOwnerId = input.expectedCheckoutRunId !== null && + input.expectedCheckoutRunId !== input.actorRunId && + input.expectedCheckoutRunId !== input.expectedExecutionRunId + ? input.expectedCheckoutRunId : null; - if (distinctCheckoutOwnerId && !isReapableHeartbeatRunRow(distinctCheckoutOwnerRun)) { - return false; - } + const distinctCheckoutOwnerRun = distinctCheckoutOwnerId + ? ownerRunById.get(distinctCheckoutOwnerId) ?? null + : null; + if (distinctCheckoutOwnerId && !isReapableHeartbeatRunRow(distinctCheckoutOwnerRun)) { + return false; + } - const cancellation = { - reason: "Cancelled because the stale issue execution lock was released", - errorCode: "issue_execution_lock_reaped", - }; - if ( - !(await cancelNeverStartedOwnerRun(tx, executionOwnerRun, cancellation)) || - (distinctCheckoutOwnerId && - !(await cancelNeverStartedOwnerRun(tx, distinctCheckoutOwnerRun, cancellation))) - ) return false; + const cancellation = { + reason: "Cancelled because the stale issue execution lock was released", + errorCode: "issue_execution_lock_reaped", + }; + if ( + !(await cancelNeverStartedOwnerRun(tx, executionOwnerRun, cancellation)) || + (distinctCheckoutOwnerId && + !(await cancelNeverStartedOwnerRun(tx, distinctCheckoutOwnerRun, cancellation))) + ) return false; - const cleared = await tx + const cleared = await tx + .update(issues) + .set({ + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: new Date(), + }) + .where( + and( + eq(issues.id, input.issueId), + eq(issues.companyId, input.companyId), + eq(issues.executionRunId, input.expectedExecutionRunId), + input.expectedCheckoutRunId + ? eq(issues.checkoutRunId, input.expectedCheckoutRunId) + : isNull(issues.checkoutRunId), + ), + ) + .returning({ id: issues.id }) + .then((rows) => rows[0] ?? null); + + return cleared != null; + } + + async function clearStaleExecutionLock(input: StaleExecutionLockInput): Promise { + return withIssueOwnershipTransaction(input.issueId, false, (tx) => + clearStaleExecutionLockInTx(tx, input)); + } + + async function adoptStaleExecutionLock(input: StaleExecutionLockInput & { actorAgentId: string }) { + return withIssueOwnershipTransaction(input.issueId, null, async (tx) => { + const cleared = await clearStaleExecutionLockInTx(tx, input); + if (!cleared) return null; + + return tx .update(issues) .set({ - executionRunId: null, - executionAgentNameKey: null, - executionLockedAt: null, + checkoutRunId: input.actorRunId, + executionRunId: input.actorRunId, + executionLockedAt: new Date(), + checkoutRestoreStatus: checkoutRestoreStatusExpression, updatedAt: new Date(), }) - .where( - and( - eq(issues.id, input.issueId), - eq(issues.executionRunId, input.expectedExecutionRunId), - input.expectedCheckoutRunId - ? eq(issues.checkoutRunId, input.expectedCheckoutRunId) - : isNull(issues.checkoutRunId), - ), - ) - .returning({ id: issues.id }) + .where(and( + eq(issues.id, input.issueId), + eq(issues.companyId, input.companyId), + eq(issues.status, "in_progress"), + eq(issues.assigneeAgentId, input.actorAgentId), + isNull(issues.executionRunId), + input.expectedCheckoutRunId + ? eq(issues.checkoutRunId, input.expectedCheckoutRunId) + : isNull(issues.checkoutRunId), + )) + .returning({ + id: issues.id, + companyId: issues.companyId, + status: issues.status, + assigneeAgentId: issues.assigneeAgentId, + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) .then((rows) => rows[0] ?? null); - - return cleared != null; }); } @@ -6812,70 +6975,7 @@ export function issueService(db: Db) { // precondition: a terminal run holds no real claim regardless of who is // assigned or what status the issue is currently in. async function clearCheckoutRunIfTerminal(issueId: string): Promise { - return db.transaction(async (tx) => { - await tx.execute( - sql`select ${issues.id} from ${issues} where ${issues.id} = ${issueId} for update`, - ); - const issue = await tx - .select({ - checkoutRunId: issues.checkoutRunId, - executionRunId: issues.executionRunId, - companyId: issues.companyId, - }) - .from(issues) - .where(eq(issues.id, issueId)) - .then((rows) => rows[0] ?? null); - if (!issue?.checkoutRunId) return false; - - await tx.execute( - sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${issue.checkoutRunId} for update`, - ); - const run = await tx - .select({ status: heartbeatRuns.status }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, issue.checkoutRunId)) - .then((rows) => rows[0] ?? null); - if (run && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(run.status)) return false; - - if (issue.executionRunId && issue.executionRunId !== issue.checkoutRunId) { - await tx.execute( - sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${issue.executionRunId} for update`, - ); - const executionRun = await tx - .select({ status: heartbeatRuns.status }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, issue.executionRunId)) - .then((rows) => rows[0] ?? null); - if (executionRun && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(executionRun.status)) return false; - } - - const updated = await tx - .update(issues) - .set({ - checkoutRunId: null, - executionRunId: null, - executionAgentNameKey: null, - executionLockedAt: null, - updatedAt: new Date(), - }) - .where( - and( - eq(issues.id, issueId), - eq(issues.checkoutRunId, issue.checkoutRunId), - issue.executionRunId - ? eq(issues.executionRunId, issue.executionRunId) - : isNull(issues.executionRunId), - ), - ) - .returning({ id: issues.id }) - .then((rows) => rows[0] ?? null); - - if (updated) { - await restoreCheckoutPromotedStatus(tx, { issueId, companyId: issue.companyId }); - } - - return Boolean(updated); - }); + return clearTerminalIssueRunLocks(issueId, "checkout"); } // BLO-27572: `update` needs `addComment` to post the routine cancellation @@ -10624,8 +10724,7 @@ export function issueService(db: Db) { }); } - await clearExecutionRunIfTerminal(id); - await clearCheckoutRunIfTerminal(id); + await clearTerminalIssueRunLocksPair(id); if (checkoutRunId) { const routineLockOwner = await findOpenRoutineExecutionLockOwnerForIssue(db, issueCompany.companyId, id); @@ -10838,6 +10937,7 @@ export function issueService(db: Db) { ) { const staleAdoption = await adoptStaleCheckoutRun({ issueId: id, + companyId: issueCompany.companyId, actorAgentId: agentId, actorRunId: checkoutRunId, expectedCheckoutRunId: current.checkoutRunId, @@ -10952,6 +11052,7 @@ export function issueService(db: Db) { ) { const cleared = await clearStaleExecutionLock({ issueId: id, + companyId: issueCompany.companyId, expectedCheckoutRunId: current.checkoutRunId, expectedExecutionRunId: current.executionRunId, actorRunId: checkoutRunId, @@ -11019,8 +11120,7 @@ export function issueService(db: Db) { actorAgentId: string, actorRunId: string | null, ) => { - await clearExecutionRunIfTerminal(id); - await clearCheckoutRunIfTerminal(id); + await clearTerminalIssueRunLocksPair(id); const loadCurrent = () => db .select({ @@ -11086,6 +11186,7 @@ export function issueService(db: Db) { if (canAdoptUnownedCheckout(candidate)) { const adopted = await adoptUnownedCheckoutRun({ issueId: id, + companyId: candidate.companyId, actorAgentId, actorRunId: actorRunId!, }); @@ -11111,6 +11212,7 @@ export function issueService(db: Db) { const previousCheckoutRunId = candidate.checkoutRunId; const staleAdoption = await adoptStaleCheckoutRun({ issueId: id, + companyId: candidate.companyId, actorAgentId, actorRunId, expectedCheckoutRunId: previousCheckoutRunId, @@ -11215,6 +11317,7 @@ export function issueService(db: Db) { ) { const staleAdoption = await adoptStaleCheckoutRun({ issueId: id, + companyId: current.companyId, actorAgentId, actorRunId, expectedCheckoutRunId: current.checkoutRunId, @@ -11252,52 +11355,23 @@ export function issueService(db: Db) { current.executionRunId && current.executionRunId !== actorRunId ) { - const cleared = await clearStaleExecutionLock({ + const adopted = await adoptStaleExecutionLock({ issueId: id, + companyId: current.companyId, expectedCheckoutRunId: current.checkoutRunId, expectedExecutionRunId: current.executionRunId, actorRunId, + actorAgentId, }); - if (cleared) { - const refreshed = await db - .update(issues) - .set({ - checkoutRunId: actorRunId, - executionRunId: actorRunId, - executionLockedAt: new Date(), - checkoutRestoreStatus: checkoutRestoreStatusExpression, - updatedAt: new Date(), - }) - .where( - and( - eq(issues.id, id), - eq(issues.status, "in_progress"), - eq(issues.assigneeAgentId, actorAgentId), - isNull(issues.executionRunId), - current.checkoutRunId - ? eq(issues.checkoutRunId, current.checkoutRunId) - : isNull(issues.checkoutRunId), - ), - ) - .returning({ - id: issues.id, - companyId: issues.companyId, - status: issues.status, - assigneeAgentId: issues.assigneeAgentId, - checkoutRunId: issues.checkoutRunId, - executionRunId: issues.executionRunId, - }) - .then((rows) => rows[0] ?? null); - if (refreshed) { - await cancelStaleIssueContextRuns({ - companyId: refreshed.companyId, - issueId: refreshed.id, - keepRunId: actorRunId, - reason: "Cancelled because the stale issue execution lock was adopted by the current run", - errorCode: "issue_execution_lock_adopted", - }); - return { ...refreshed, adoptedFromRunId: current.executionRunId }; - } + if (adopted) { + await cancelStaleIssueContextRuns({ + companyId: adopted.companyId, + issueId: adopted.id, + keepRunId: actorRunId, + reason: "Cancelled because the stale issue execution lock was adopted by the current run", + errorCode: "issue_execution_lock_adopted", + }); + return { ...adopted, adoptedFromRunId: current.executionRunId }; } } @@ -11329,8 +11403,7 @@ export function issueService(db: Db) { actorAgentId: string, actorRunId: string | null, ) => { - await clearExecutionRunIfTerminal(id); - await clearCheckoutRunIfTerminal(id); + await clearTerminalIssueRunLocksPair(id); const current = await db .select({ id: issues.id, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index b62533512a46..6889327a0869 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -50,8 +50,19 @@ import { isVerifiedIssueTreeControlInteractionWake, issueTreeControlService, } from "../issue-tree-control.js"; -import { TERMINAL_HEARTBEAT_RUN_STATUSES, externalWaitFromDescription, issueService } from "../issues.js"; -import { releaseIssueRunOwnership, restoreCheckoutPromotedStatus } from "../issue-checkout-status.js"; +import { + TERMINAL_HEARTBEAT_RUN_STATUSES, + externalWaitFromDescription, + issueService, + lockIssueParentMutationCompany, +} from "../issues.js"; +import { + issueLockOwnerStateMatches, + lockIssueOwnership, + releaseIssueRunOwnership, + restoreCheckoutPromotedStatus, + type IssueLockOwnerState, +} from "../issue-checkout-status.js"; import { applyIssueMonitorPolicyTransition, derivePersistedMonitorState, @@ -102,6 +113,33 @@ import { type DbTransaction = Parameters[0]>[0]; +/** + * The recovery transaction has already upserted its source-scoped action when + * the issue UPDATE reports no match. That result is a lost write precondition, + * not a successful no-op: returning from the transaction callback would commit + * the action/comment/activity rows without the matching issue mutation. Throw a + * private sentinel so Drizzle rolls the transaction back, then translate it + * back to the normal "another writer won" null result at the boundary. + */ +class RecoveryEscalationRollback extends Error { + constructor() { + super("Recovery escalation rolled back because the issue update matched no row"); + this.name = "RecoveryEscalationRollback"; + } +} + +/** + * The provider-quota monitor is created after the escalation transaction commits, + * so a failed action CAS must abort the monitor transaction as well. Returning + * from the transaction callback would commit an orphaned wake/run pair. + */ +class ProviderQuotaMonitorRollback extends Error { + constructor() { + super("Provider-quota recovery monitor action changed before it could be updated"); + this.name = "ProviderQuotaMonitorRollback"; + } +} + const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["interrupted", "failed", "cancelled", "timed_out"] as const; export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000; @@ -471,6 +509,7 @@ type RecoveryWakeupOptions = { contextSnapshot?: Record; retryOfRunId?: string | null; scheduledRetryAttempt?: number; + expectedLockOwnerState?: IssueLockOwnerState | null; }; type RecoveryWakeup = ( @@ -571,19 +610,6 @@ export function isInfraClassStrandedFailure(latestRun: LatestIssueRun): boolean return /pod is gone|pod was removed/i.test(latestRun.error ?? ""); } -// BLO-20933: `issuesSvc.update` enforces `expectedCurrentAssigneeAgentId` by THROWING a 409 -// (both from the pre-read snapshot check and from the zero-matched-rows path), never by -// returning falsy. A recovery escalation that loses the assignee race must treat that as -// "another writer won, skip this issue" rather than letting it propagate: the caller's -// per-issue loop would otherwise abort the whole batch. Matched on the precondition detail -// we set ourselves so unrelated 409s still surface. -function isAssigneePreconditionConflict(error: unknown): boolean { - if (!(error instanceof HttpError) || error.status !== 409) return false; - const details = error.details; - if (!details || typeof details !== "object") return false; - return "expectedAssigneeAgentId" in details; -} - function resolveStrandedRecoveryCause( latestRun: LatestIssueRun, explicitCause?: StrandedRecoveryCause, @@ -1000,22 +1026,6 @@ function isTerminalDispatchRaceRun( ); } -// BLO-19160: the three issue columns adoption rewrites — the execution lock pair -// plus the owner. Captured fresh when a handover marker is observed and used as -// a compare-and-set precondition on every recovery mutation that observation -// leads to. -type IssueLockOwnerState = { - executionRunId: string | null; - checkoutRunId: string | null; - assigneeAgentId: string | null; -}; - -function issueLockOwnerStateMatches(a: IssueLockOwnerState, b: IssueLockOwnerState) { - return a.executionRunId === b.executionRunId && - a.checkoutRunId === b.checkoutRunId && - a.assigneeAgentId === b.assigneeAgentId; -} - // BLO-19160: the outcome of observing a checkout-handover marker when the // adopter can no longer prove continuity. `markerRunId` is the handover run — // the newest run genuinely scoped to this issue, so the honest retry parent. @@ -2469,6 +2479,7 @@ export function recoveryService( ...(input.extraContext ?? {}), }, "normal_model"), retryOfRunId: input.retryOfRunId, + expectedLockOwnerState: input.expectedLockOwnerState, scheduledRetryAttempt: typeof input.extraContext?.scheduledRetryAttempt === "number" ? input.extraContext.scheduledRetryAttempt @@ -2501,6 +2512,7 @@ export function recoveryService( wakeReason: "issue_assigned", source: "issue.assigned_todo_liveness_dispatch", }, "normal_model"), + expectedLockOwnerState, }); } @@ -5027,6 +5039,7 @@ export function recoveryService( latestRun: LatestIssueRun; recoveryCause: StrandedRecoveryCause; hasNewActivitySinceLastAttempt: boolean; + expectedLockOwnerState?: IssueLockOwnerState | null; }) { if (input.recoveryCause === "provider_quota" && !input.action.ownerAgentId) return; if (input.recoveryCause === "workspace_validation_failed" || input.recoveryCause === "configuration_incomplete") return; @@ -5053,11 +5066,15 @@ export function recoveryService( // horizon in `strandedRecoveryWakeAttemptsExhausted`, which no sweep rewrites and which // does not depend on `attemptCount` moving at all. Attempts bound delivered-but- // unproductive wakes; the horizon bounds wall-clock regardless of delivery. + const reservedOwnerAgentId = input.action.ownerAgentId; + const reservedAttemptCount = input.action.attemptCount; const refundUnspentWakeAttempt = async (cause: "enqueue_threw" | "enqueue_not_delivered", error?: unknown) => { const release = () => recoveryActionsSvc.releaseWakeAttempt({ companyId: input.issue.companyId, actionId: input.action.id, + expectedOwnerAgentId: reservedOwnerAgentId, + expectedAttemptCount: reservedAttemptCount, }); try { await release(); @@ -5137,6 +5154,7 @@ export function recoveryService( recoveryCause: input.recoveryCause, suppressedNonAssigneeWake: true, }, "status_only"), + expectedLockOwnerState: input.expectedLockOwnerState, }); return; } @@ -5171,6 +5189,7 @@ export function recoveryService( strandedRunId: input.latestRun?.id ?? null, recoveryCause: input.recoveryCause, }, "status_only"), + expectedLockOwnerState: input.expectedLockOwnerState, }); } @@ -5195,88 +5214,187 @@ export function recoveryService( actionId: string; agentId: string; }) { - const existing = await db - .select() - .from(heartbeatRuns) - .where(and( - eq(heartbeatRuns.companyId, input.issue.companyId), - eq(heartbeatRuns.agentId, input.agentId), - eq(heartbeatRuns.status, "scheduled_retry"), - sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${input.issue.id}`, - )) - .orderBy(desc(heartbeatRuns.scheduledRetryAt)) - .limit(1) - .then((rows) => rows[0] ?? null); - if (existing) return existing; - const now = new Date(); const retryAt = readProviderQuotaRetryAt(input.latestRun, now); - return db.transaction(async (tx) => { - const wakeup = await tx - .insert(agentWakeupRequests) - .values({ - companyId: input.issue.companyId, - agentId: input.agentId, - source: "automation", - triggerDetail: "system", - reason: "provider_quota_recovery", - payload: withRecoveryModelProfileHint({ - issueId: input.issue.id, - retryOfRunId: input.latestRun?.id ?? null, - retryReason: "provider_quota_recovery", - providerQuotaRetryNotBefore: retryAt.toISOString(), - }, "normal_model"), - status: "queued", - requestedByActorType: "system", - requestedByActorId: null, - idempotencyKey: `provider_quota_recovery:${input.issue.id}:${retryAt.toISOString()}`, - updatedAt: now, - }) - .returning() - .then((rows) => rows[0]!); - const scheduledRun = await tx - .insert(heartbeatRuns) - .values({ - companyId: input.issue.companyId, - agentId: input.agentId, - invocationSource: "automation", - triggerDetail: "system", - status: "scheduled_retry", - wakeupRequestId: wakeup.id, - retryOfRunId: input.latestRun?.id ?? null, - scheduledRetryAt: retryAt, - scheduledRetryAttempt: 1, - scheduledRetryReason: "provider_quota_recovery", - contextSnapshot: withRecoveryModelProfileHint({ - issueId: input.issue.id, - taskId: input.issue.id, - wakeReason: "provider_quota_recovery", - retryReason: "provider_quota_recovery", - providerQuotaRetryNotBefore: retryAt.toISOString(), - }, "normal_model"), - updatedAt: now, - }) - .returning() - .then((rows) => rows[0]!); - await tx - .update(agentWakeupRequests) - .set({ runId: scheduledRun.id, updatedAt: now }) - .where(eq(agentWakeupRequests.id, wakeup.id)); - await tx - .update(issueRecoveryActions) - .set({ - monitorPolicy: { - type: "wait_recovery", - retryAgentId: input.agentId, - scheduledRunId: scheduledRun.id, - retryAt: retryAt.toISOString(), + try { + return await db.transaction(async (tx) => { + // The escalation has committed, so this second transaction must prove that + // the issue is still the exact post-escalation state that authorized the + // monitor. Adoption, reassignment, resolution, or another recovery writer + // can otherwise leave a retry attached to stale recovery state. + await lockIssueOwnership(tx, input.issue.companyId, input.issue.id); + const [freshIssue] = await tx + .select() + .from(issues) + .where(and( + eq(issues.companyId, input.issue.companyId), + eq(issues.id, input.issue.id), + )) + .limit(1) + .for("update"); + if (!freshIssue || freshIssue.status !== input.issue.status || !issueLockOwnerStateMatches( + { + executionRunId: input.issue.executionRunId, + checkoutRunId: input.issue.checkoutRunId, + assigneeAgentId: input.issue.assigneeAgentId, }, - timeoutAt: retryAt, - updatedAt: now, - }) - .where(eq(issueRecoveryActions.id, input.actionId)); - return scheduledRun; - }); + { + executionRunId: freshIssue.executionRunId, + checkoutRunId: freshIssue.checkoutRunId, + assigneeAgentId: freshIssue.assigneeAgentId, + }, + )) { + return null; + } + + // Lock the action before checking its routing/evidence. The action is + // deliberately narrower than ACTIVE_RECOVERY_ACTION_STATUSES here: + // `escalated` is a terminal human-attention state and must not receive a + // scheduler monitor as a side effect of an old post-commit callback. + const [action] = await tx + .select() + .from(issueRecoveryActions) + .where(and( + eq(issueRecoveryActions.id, input.actionId), + eq(issueRecoveryActions.companyId, freshIssue.companyId), + eq(issueRecoveryActions.sourceIssueId, freshIssue.id), + )) + .limit(1) + .for("update"); + if (!action || action.status !== "active" || action.cause !== "provider_quota" || + action.ownerAgentId !== null || action.returnOwnerAgentId !== input.agentId) { + return null; + } + + // A recovery action records the run that supplied its evidence. If that + // evidence is present, the post-commit monitor callback must still be for + // the same run; an older callback must not arm a retry for a newer action + // generation. Older rows may lack either field, so absent evidence remains + // compatible with the pre-evidence schema. + const actionEvidence = parseObject(action.evidence); + const actionLatestRunId = readNonEmptyString(actionEvidence.latestRunId); + const actionLatestRunAgentId = readNonEmptyString(actionEvidence.latestRunAgentId); + if ( + (actionLatestRunId !== null && actionLatestRunId !== input.latestRun?.id) || + (actionLatestRunAgentId !== null && actionLatestRunAgentId !== input.latestRun?.agentId) + ) { + return null; + } + + const existing = await tx + .select() + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, freshIssue.companyId), + eq(heartbeatRuns.agentId, input.agentId), + eq(heartbeatRuns.status, "scheduled_retry"), + eq(heartbeatRuns.scheduledRetryReason, "provider_quota_recovery"), + or( + sql`${heartbeatRuns.contextSnapshot} ->> 'recoveryActionId' = ${action.id}`, + // Compatibility for monitors created before the stable action key + // was introduced. The reason and issue scope keep this fallback + // from matching unrelated scheduled retries for the same issue. + and( + sql`${heartbeatRuns.contextSnapshot} ->> 'recoveryActionId' is null`, + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${freshIssue.id}`, + ), + ), + )) + .orderBy(desc(heartbeatRuns.scheduledRetryAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + if (existing) return existing; + + const recoveryActionId = action.id; + const stableIdempotencyKey = `provider_quota_recovery:${recoveryActionId}`; + const wakeup = await tx + .insert(agentWakeupRequests) + .values({ + companyId: freshIssue.companyId, + agentId: input.agentId, + source: "automation", + triggerDetail: "system", + reason: "provider_quota_recovery", + payload: withRecoveryModelProfileHint({ + issueId: freshIssue.id, + recoveryActionId, + retryOfRunId: input.latestRun?.id ?? null, + retryReason: "provider_quota_recovery", + providerQuotaRetryNotBefore: retryAt.toISOString(), + }, "normal_model"), + status: "queued", + requestedByActorType: "system", + requestedByActorId: null, + idempotencyKey: stableIdempotencyKey, + updatedAt: now, + }) + .returning() + .then((rows) => rows[0]!); + const scheduledRun = await tx + .insert(heartbeatRuns) + .values({ + companyId: freshIssue.companyId, + agentId: input.agentId, + invocationSource: "automation", + triggerDetail: "system", + status: "scheduled_retry", + wakeupRequestId: wakeup.id, + retryOfRunId: input.latestRun?.id ?? null, + scheduledRetryAt: retryAt, + scheduledRetryAttempt: 1, + scheduledRetryReason: "provider_quota_recovery", + contextSnapshot: withRecoveryModelProfileHint({ + issueId: freshIssue.id, + taskId: freshIssue.id, + recoveryActionId, + wakeReason: "provider_quota_recovery", + retryReason: "provider_quota_recovery", + providerQuotaRetryNotBefore: retryAt.toISOString(), + }, "normal_model"), + updatedAt: now, + }) + .returning() + .then((rows) => rows[0]!); + await tx + .update(agentWakeupRequests) + .set({ runId: scheduledRun.id, updatedAt: now }) + .where(eq(agentWakeupRequests.id, wakeup.id)); + const [updatedAction] = await tx + .update(issueRecoveryActions) + .set({ + monitorPolicy: { + type: "wait_recovery", + retryAgentId: input.agentId, + recoveryActionId, + scheduledRunId: scheduledRun.id, + retryAt: retryAt.toISOString(), + }, + timeoutAt: retryAt, + updatedAt: now, + }) + .where(and( + eq(issueRecoveryActions.id, action.id), + eq(issueRecoveryActions.companyId, freshIssue.companyId), + eq(issueRecoveryActions.sourceIssueId, freshIssue.id), + eq(issueRecoveryActions.status, "active"), + eq(issueRecoveryActions.cause, "provider_quota"), + isNull(issueRecoveryActions.ownerAgentId), + eq(issueRecoveryActions.returnOwnerAgentId, input.agentId), + ...(actionLatestRunId === null + ? [] + : [sql`${issueRecoveryActions.evidence} ->> 'latestRunId' = ${actionLatestRunId}`]), + ...(actionLatestRunAgentId === null + ? [] + : [sql`${issueRecoveryActions.evidence} ->> 'latestRunAgentId' = ${actionLatestRunAgentId}`]), + )) + .returning(); + if (!updatedAction) throw new ProviderQuotaMonitorRollback(); + return scheduledRun; + }); + } catch (error) { + if (error instanceof ProviderQuotaMonitorRollback) return null; + throw error; + } } function buildRecoveryIssueInPlaceEscalationComment(input: { @@ -5315,47 +5433,85 @@ export function recoveryService( latestRun: LatestIssueRun; expectedLockOwnerState?: IssueLockOwnerState | null; }) { - // BLO-19160: see `escalateStrandedAssignedIssue` — a handover-derived - // escalation must not commit if the lock/owner moved under it. - if (await issueLockOwnerStateChanged(input.issue.id, input.expectedLockOwnerState)) return null; - const updated = await issuesSvc.update(input.issue.id, { status: "blocked" }); - if (!updated) return null; + const result = await db.transaction(async (tx) => { + await lockIssueOwnership(tx, input.issue.companyId, input.issue.id); + + // Acquire the graph lock before the issue row. `issuesSvc.update` takes + // this order for status mutations that touch the issue graph; retaining + // the same order here avoids an inversion with a concurrent blocker + // update while still keeping the handover decision and its side effects + // in one transaction. + await lockIssueParentMutationCompany(input.issue.companyId, tx); + const [fresh] = await tx + .select() + .from(issues) + .where(eq(issues.id, input.issue.id)) + .limit(1) + .for("update"); + if (!fresh || fresh.status !== input.previousStatus && fresh.status !== "blocked") return null; + if ( + input.expectedLockOwnerState && + !issueLockOwnerStateMatches(input.expectedLockOwnerState, { + executionRunId: fresh.executionRunId, + checkoutRunId: fresh.checkoutRunId, + assigneeAgentId: fresh.assigneeAgentId, + }) + ) return null; - const prefix = await getCompanyIssuePrefix(input.issue.companyId); - await issuesSvc.addComment( - input.issue.id, - buildRecoveryIssueInPlaceEscalationComment({ - issue: input.issue, - previousStatus: input.previousStatus, - latestRun: input.latestRun, - prefix, - }), - {}, - ); + const updated = await issuesSvc.update( + fresh.id, + { + status: "blocked", + expectedCurrentStatus: fresh.status, + expectedCurrentAssigneeAgentId: fresh.assigneeAgentId, + expectedCurrentCheckoutRunId: fresh.checkoutRunId, + expectedCurrentExecutionRunId: fresh.executionRunId, + }, + tx, + ); + if (!updated) return null; - await logActivity(db, { - companyId: input.issue.companyId, - actorType: "system", - actorId: "system", - agentId: null, - runId: null, - action: "issue.updated", - entityType: "issue", - entityId: input.issue.id, - details: { - identifier: input.issue.identifier, - status: "blocked", - previousStatus: input.previousStatus, - source: "recovery.reconcile_stranded_recovery_issue", - latestRunId: input.latestRun?.id ?? null, - latestRunStatus: input.latestRun?.status ?? null, - latestRunErrorCode: input.latestRun?.errorCode ?? null, - originKind: input.issue.originKind, - originId: input.issue.originId, - }, - }); + const prefix = await getCompanyIssuePrefix(fresh.companyId); + await issuesSvc.addComment( + fresh.id, + buildRecoveryIssueInPlaceEscalationComment({ + issue: fresh, + previousStatus: input.previousStatus, + latestRun: input.latestRun, + prefix, + }), + {}, + { authorType: "system" }, + tx, + ); - return updated; + const publish = await logActivity(tx as unknown as Db, { + companyId: fresh.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: null, + action: "issue.updated", + entityType: "issue", + entityId: fresh.id, + details: { + identifier: fresh.identifier, + status: "blocked", + previousStatus: input.previousStatus, + source: "recovery.reconcile_stranded_recovery_issue", + latestRunId: input.latestRun?.id ?? null, + latestRunStatus: input.latestRun?.status ?? null, + latestRunErrorCode: input.latestRun?.errorCode ?? null, + originKind: fresh.originKind, + originId: fresh.originId, + }, + }, { deferPublish: true }); + + return { updated, publish }; + }); + if (!result) return null; + result.publish(); + return result.updated; } function isWaitingOnReviewContinuationRun(latestRun: LatestIssueRun) { @@ -5382,23 +5538,28 @@ export function recoveryService( return monitor?.status === "triggered"; } + type ReviewWaitingParkOutcome = "parked" | "already_parked" | "lost_race" | "failed"; + async function parkReviewWaitingContinuationIssue(input: { issue: typeof issues.$inferSelect; previousStatus: "in_progress"; latestRun: LatestIssueRun; expectedLockOwnerState?: IssueLockOwnerState | null; - }) { - return await db.transaction(async (tx) => { - await tx.execute( - sql`select pg_advisory_xact_lock(hashtextextended(${input.issue.companyId} || ':' || ${input.issue.id}, 0))`, - ); + }): Promise { + const result = await db.transaction(async (tx) => { + await lockIssueOwnership(tx, input.issue.companyId, input.issue.id); const [fresh] = await tx .select() .from(issues) .where(eq(issues.id, input.issue.id)) - .limit(1); - if (!fresh || fresh.status !== "in_progress" || !hasActiveMonitorPath(fresh)) return null; + .limit(1) + .for("update"); + if (!fresh) return { outcome: "lost_race" as const }; + if (fresh.status === "in_review") return { outcome: "already_parked" as const }; + if (fresh.status !== input.previousStatus || !hasActiveMonitorPath(fresh)) { + return { outcome: "lost_race" as const }; + } // BLO-19160: parking to `in_review` is a status mutation on the handover // path just as much as an escalation is, so it takes the same CAS. @@ -5410,11 +5571,29 @@ export function recoveryService( assigneeAgentId: fresh.assigneeAgentId, }) ) { - return null; + return { outcome: "lost_race" as const }; } - const updated = await issuesSvc.update(fresh.id, { status: "in_review" }, tx); - if (!updated) return null; + let updated: Awaited>; + try { + updated = await issuesSvc.update( + fresh.id, + { + status: "in_review", + expectedCurrentStatus: input.previousStatus, + expectedCurrentAssigneeAgentId: fresh.assigneeAgentId, + expectedCurrentCheckoutRunId: fresh.checkoutRunId, + expectedCurrentExecutionRunId: fresh.executionRunId, + }, + tx, + ); + } catch (error) { + if (error instanceof HttpError && error.status === 409) { + return { outcome: "lost_race" as const }; + } + throw error; + } + if (!updated) return { outcome: "lost_race" as const }; const activeRecoveryAction = await recoveryActionsSvc.resolveActiveForIssue({ companyId: fresh.companyId, @@ -5425,7 +5604,7 @@ export function recoveryService( "Continuation retry was intentionally cancelled because the issue is waiting on review/CI, and the source issue has an active monitor path.", }, tx); - await logActivity(db, { + const publish = await logActivity(tx as unknown as Db, { companyId: fresh.companyId, actorType: "system", actorId: "system", @@ -5444,18 +5623,22 @@ export function recoveryService( latestRunErrorCode: input.latestRun?.errorCode ?? null, recoveryActionId: activeRecoveryAction?.id ?? null, }, - }); + }, { deferPublish: true }); - return updated; + return { outcome: "parked" as const, publish }; }); + if (result.outcome === "parked") { + result.publish(); + return "parked"; + } + return result.outcome; } - type ReviewWaitingParkOutcome = "parked" | "already_parked" | "failed"; - async function parkNoDependencyReviewWaitingIssue(input: { issue: typeof issues.$inferSelect; previousStatus: "in_progress"; latestRun: LatestIssueRun; + expectedLockOwnerState?: IssueLockOwnerState | null; }): Promise { // BLO-16146: a continuation that deliberately parked for review/approval, with no // dependency to convert into a `blocked` wait (resolveContinuationWaitingOnReview @@ -5467,31 +5650,26 @@ export function recoveryService( // parkReviewWaitingContinuationIssue minus the monitor-path requirement, plus a // plain-language comment (no monitor path guarantees a re-poke, so the wait must be // visible in the thread). - return await db.transaction(async (tx) => { - await tx.execute( - sql`select pg_advisory_xact_lock(hashtextextended(${input.issue.companyId} || ':' || ${input.issue.id}, 0))`, - ); + const result = await db.transaction(async (tx) => { + await lockIssueOwnership(tx, input.issue.companyId, input.issue.id); const [fresh] = await tx .select() .from(issues) .where(eq(issues.id, input.issue.id)) - .limit(1); - if (!fresh) return "failed"; + .limit(1) + .for("update"); + if (!fresh) return "lost_race"; if (fresh.status === "in_review") return "already_parked"; - if (fresh.status !== "in_progress") return "failed"; - - // BLO-19160: deliberately NO lock-owner CAS here, unlike the sibling - // parks/escalations. Two reasons, in order: - // 1. `ReviewWaitingParkOutcome` has no "took no action" variant, and a - // lost race must not map to "failed" — the caller treats "failed" as - // a genuine park failure and falls through to `blocked` escalation, - // i.e. exactly the clobber a CAS is supposed to prevent. Guarding - // here without a new outcome variant is worse than not guarding. - // 2. It is unreachable on the handover path anyway: the only call site - // is gated on `isWaitingOnReviewContinuationRun(latestRun)`, which - // requires `latestRun?.status === "cancelled"`, and the handover - // path sets `latestRun` to null. + if (fresh.status !== input.previousStatus) return "lost_race"; + + if (input.expectedLockOwnerState && !issueLockOwnerStateMatches(input.expectedLockOwnerState, { + executionRunId: fresh.executionRunId, + checkoutRunId: fresh.checkoutRunId, + assigneeAgentId: fresh.assigneeAgentId, + })) { + return "lost_race"; + } // The in_review transition runs an evidence gate (issues.ts) that throws // `unprocessable` when the issue has no reviewable evidence yet (analysis-only @@ -5501,15 +5679,26 @@ export function recoveryService( // Any other transient failure degrades the same safe way. let updated: Awaited>; try { - updated = await issuesSvc.update(fresh.id, { status: "in_review" }, tx); + updated = await issuesSvc.update( + fresh.id, + { + status: "in_review", + expectedCurrentStatus: input.previousStatus, + expectedCurrentAssigneeAgentId: fresh.assigneeAgentId, + expectedCurrentCheckoutRunId: fresh.checkoutRunId, + expectedCurrentExecutionRunId: fresh.executionRunId, + }, + tx, + ); } catch (err) { + if (err instanceof HttpError && err.status === 409) return "lost_race"; logger.warn( { err, issueId: fresh.id, identifier: fresh.identifier }, "parkNoDependencyReviewWaitingIssue: in_review park rejected; escalating instead", ); return "failed"; } - if (!updated) return "failed"; + if (!updated) return "lost_race"; const activeRecoveryAction = await recoveryActionsSvc.resolveActiveForIssue({ companyId: fresh.companyId, @@ -5532,7 +5721,7 @@ export function recoveryService( tx, ); - await logActivity(db, { + const publish = await logActivity(tx as unknown as Db, { companyId: fresh.companyId, actorType: "system", actorId: "system", @@ -5551,10 +5740,12 @@ export function recoveryService( latestRunErrorCode: input.latestRun?.errorCode ?? null, recoveryActionId: activeRecoveryAction?.id ?? null, }, - }); + }, { deferPublish: true }); - return "parked"; + return { outcome: "parked" as const, publish }; }); + if (typeof result === "object" && result.outcome === "parked") result.publish(); + return typeof result === "string" ? result : result.outcome; } async function existingBlockerIssueIds(companyId: string, issueId: string) { @@ -5575,12 +5766,13 @@ export function recoveryService( companyId: string, issueId: string, blockerIssueIds: string[], + dbOrTx: Db | DbTransaction = db, ) { const candidates = new Set(blockerIssueIds); const cycleForming = new Set(); if (candidates.size === 0) return cycleForming; - const rows = await db + const rows = await dbOrTx .select({ blockerIssueId: issueRelations.issueId, blockedIssueId: issueRelations.relatedIssueId, @@ -5673,8 +5865,12 @@ export function recoveryService( }); } - async function existingUnresolvedBlockerIssues(companyId: string, issueId: string) { - return db + async function existingUnresolvedBlockerIssues( + companyId: string, + issueId: string, + dbOrTx: Db | DbTransaction = db, + ) { + return dbOrTx .select({ id: issues.id, identifier: issues.identifier }) .from(issueRelations) .innerJoin( @@ -5704,113 +5900,143 @@ export function recoveryService( issue: typeof issues.$inferSelect, expectedLockOwnerState?: IssueLockOwnerState | null, ) { - const existingBlockers = await existingUnresolvedBlockerIssues(issue.companyId, issue.id); - const openChildren = await db - .select({ id: issues.id, identifier: issues.identifier }) - .from(issues) - .where( - and( - eq(issues.companyId, issue.companyId), - eq(issues.parentId, issue.id), - visibleIssueCondition(), - notInArray(issues.status, ["done", "cancelled"]), - ), - ); - const blockerRowsById = new Map(); - for (const row of existingBlockers) { - blockerRowsById.set(row.id, { ...row, source: "existing_unresolved_blocker" }); - } - for (const row of openChildren) { - if (!blockerRowsById.has(row.id)) { - blockerRowsById.set(row.id, { ...row, source: "open_child" }); + const result = await db.transaction(async (tx) => { + await lockIssueOwnership(tx, issue.companyId, issue.id); + // All issue-graph writers take this company lock before issue row locks. + // Holding it across blocker discovery, cycle filtering, and the update + // makes the dependency decision atomic with the relation mutation. + await lockIssueParentMutationCompany(issue.companyId, tx); + const [fresh] = await tx + .select() + .from(issues) + .where(and(eq(issues.companyId, issue.companyId), eq(issues.id, issue.id))) + .limit(1) + .for("update"); + if (!fresh || fresh.status !== issue.status) return null; + if ( + expectedLockOwnerState && + !issueLockOwnerStateMatches(expectedLockOwnerState, { + executionRunId: fresh.executionRunId, + checkoutRunId: fresh.checkoutRunId, + assigneeAgentId: fresh.assigneeAgentId, + }) + ) return null; + + const existingBlockers = await existingUnresolvedBlockerIssues(fresh.companyId, fresh.id, tx); + const openChildren = await tx + .select({ id: issues.id, identifier: issues.identifier }) + .from(issues) + .where( + and( + eq(issues.companyId, fresh.companyId), + eq(issues.parentId, fresh.id), + visibleIssueCondition(), + notInArray(issues.status, ["done", "cancelled"]), + ), + ); + const blockerRowsById = new Map(); + for (const row of existingBlockers) { + blockerRowsById.set(row.id, { ...row, source: "existing_unresolved_blocker" }); + } + for (const row of openChildren) { + if (!blockerRowsById.has(row.id)) { + blockerRowsById.set(row.id, { ...row, source: "open_child" }); + } } - } - let blockedByIssueIds = [...blockerRowsById.keys()]; - if (blockedByIssueIds.length === 0) return null; - const cycleFormingBlockerIds = await findCycleFormingBlockerIssueIds( - issue.companyId, - issue.id, - blockedByIssueIds, - ); - if (cycleFormingBlockerIds.size > 0) { - const skippedBlockers = [...cycleFormingBlockerIds] - .map((id) => blockerRowsById.get(id)) - .filter((row): row is { id: string; identifier: string | null; source: string } => Boolean(row)); - logger.warn( - { - companyId: issue.companyId, - issueId: issue.id, - identifier: issue.identifier, - skippedBlockerIssueIds: skippedBlockers.map((row) => row.id), - skippedBlockerIdentifiers: skippedBlockers.map((row) => row.identifier).filter(Boolean), - skippedBlockerSources: skippedBlockers.map((row) => ({ id: row.id, source: row.source })), - }, - "skipping cycle-forming review-wait blocker relations", - ); - blockedByIssueIds = blockedByIssueIds.filter((id) => !cycleFormingBlockerIds.has(id)); + let blockedByIssueIds = [...blockerRowsById.keys()]; if (blockedByIssueIds.length === 0) return null; - } + const cycleFormingBlockerIds = await findCycleFormingBlockerIssueIds( + fresh.companyId, + fresh.id, + blockedByIssueIds, + tx, + ); + if (cycleFormingBlockerIds.size > 0) { + const skippedBlockers = [...cycleFormingBlockerIds] + .map((id) => blockerRowsById.get(id)) + .filter((row): row is { id: string; identifier: string | null; source: string } => Boolean(row)); + logger.warn( + { + companyId: fresh.companyId, + issueId: fresh.id, + identifier: fresh.identifier, + skippedBlockerIssueIds: skippedBlockers.map((row) => row.id), + skippedBlockerIdentifiers: skippedBlockers.map((row) => row.identifier).filter(Boolean), + skippedBlockerSources: skippedBlockers.map((row) => ({ id: row.id, source: row.source })), + }, + "skipping cycle-forming review-wait blocker relations", + ); + blockedByIssueIds = blockedByIssueIds.filter((id) => !cycleFormingBlockerIds.has(id)); + if (blockedByIssueIds.length === 0) return null; + } - // The reachability check above and this write are not atomic: another relation - // update can add a path from `issue` to one of these blockers between the read and - // this write. The issue service re-validates on write and throws in that case - - // catch it here so one race falls through to the caller's existing no-dependency - // `in_review` park instead of aborting the whole periodic recovery sweep. - let updated: Awaited>; - try { - // BLO-19160: re-check the handover lock/owner state immediately before - // the mutation. A live adopter committing after the handover observation - // must not be blocked out by this path. - if (await issueLockOwnerStateChanged(issue.id, expectedLockOwnerState)) return null; - updated = await issuesSvc.update(issue.id, { status: "blocked", blockedByIssueIds }); - } catch (error) { - if (!isBlockingRelationCycleError(error)) throw error; - logger.warn( - { - companyId: issue.companyId, - issueId: issue.id, - identifier: issue.identifier, + let updated: Awaited>; + try { + updated = await issuesSvc.update( + fresh.id, + { + status: "blocked", + blockedByIssueIds, + expectedCurrentStatus: fresh.status, + expectedCurrentAssigneeAgentId: fresh.assigneeAgentId, + expectedCurrentCheckoutRunId: fresh.checkoutRunId, + expectedCurrentExecutionRunId: fresh.executionRunId, + }, + tx, + ); + } catch (error) { + if (!isBlockingRelationCycleError(error)) throw error; + logger.warn( + { + companyId: fresh.companyId, + issueId: fresh.id, + identifier: fresh.identifier, + blockedByIssueIds, + }, + "review-wait blocker write raced a concurrent relation update and formed a cycle; parking without dependency", + ); + return null; + } + if (!updated) return null; + + const waitingOn = formatIssueLinksForComment( + blockedByIssueIds + .map((id) => blockerRowsById.get(id)) + .filter((row): row is { id: string; identifier: string | null; source: string } => Boolean(row)), + ); + await issuesSvc.addComment( + fresh.id, + `This task is waiting on ${waitingOn} to finish. ` + + "It will continue automatically when that work is done - there's nothing you need to do. " + + "(It was paused because the latest run reported it was waiting for review/approval; " + + "Paperclip turned that into a normal dependency wait instead of flagging it as stuck.)", + {}, + { authorType: "system" }, + tx, + ); + const publish = await logActivity(tx as unknown as Db, { + companyId: fresh.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: null, + action: "issue.updated", + entityType: "issue", + entityId: fresh.id, + details: { + identifier: fresh.identifier, + status: "blocked", + previousStatus: issue.status, + source: "recovery.reconcile_continuation_waiting_on_review", blockedByIssueIds, }, - "review-wait blocker write raced a concurrent relation update and formed a cycle; parking without dependency", - ); - return null; - } - if (!updated) return null; - - const waitingOn = formatIssueLinksForComment( - blockedByIssueIds - .map((id) => blockerRowsById.get(id)) - .filter((row): row is { id: string; identifier: string | null; source: string } => Boolean(row)), - ); - await issuesSvc.addComment( - issue.id, - `This task is waiting on ${waitingOn} to finish. ` + - "It will continue automatically when that work is done - there's nothing you need to do. " + - "(It was paused because the latest run reported it was waiting for review/approval; " + - "Paperclip turned that into a normal dependency wait instead of flagging it as stuck.)", - {}, - { authorType: "system" }, - ); - await logActivity(db, { - companyId: issue.companyId, - actorType: "system", - actorId: "system", - agentId: null, - runId: null, - action: "issue.updated", - entityType: "issue", - entityId: issue.id, - details: { - identifier: issue.identifier, - status: "blocked", - previousStatus: issue.status, - source: "recovery.reconcile_continuation_waiting_on_review", - blockedByIssueIds, - }, + }, { deferPublish: true }); + return { updated, publish }; }); - return updated; + if (!result) return null; + result.publish(); + return result.updated; } // BLO-19954: cancel a routine-execution issue whose only run was suppressed @@ -5822,10 +6048,8 @@ export function recoveryService( issue: typeof issues.$inferSelect, latestRun: LatestIssueRun, ) { - return await db.transaction(async (tx) => { - await tx.execute( - sql`select pg_advisory_xact_lock(hashtextextended(${issue.companyId} || ':' || ${issue.id}, 0))`, - ); + const result = await db.transaction(async (tx) => { + await lockIssueOwnership(tx, issue.companyId, issue.id); const [fresh] = await tx .select() @@ -5846,7 +6070,7 @@ export function recoveryService( ); if (!updated) return null; - await logActivity(db, { + const publish = await logActivity(tx as unknown as Db, { companyId: fresh.companyId, actorType: "system", actorId: "system", @@ -5863,7 +6087,7 @@ export function recoveryService( latestRunId: latestRun?.id ?? null, latestRunErrorCode: latestRun?.errorCode ?? null, }, - }); + }, { deferPublish: true }); await issuesSvc.addComment( fresh.id, @@ -5877,8 +6101,11 @@ export function recoveryService( tx, ); - return updated; + return { updated, publish }; }); + if (!result) return null; + result.publish(); + return result.updated; } // BLO-27463: incremented by the dependency-wait gate inside @@ -5940,21 +6167,29 @@ export function recoveryService( // commit/return, waiting peers wake up and record their next attempt // against the same active source-scoped action. const escalation = await db.transaction(async (tx) => { - await tx.execute( - sql`select pg_advisory_xact_lock(hashtextextended(${input.issue.companyId} || ':' || ${input.issue.id}, 0))`, - ); + await lockIssueOwnership(tx, input.issue.companyId, input.issue.id); // Re-read source issue under the lock so the recovery action records // the latest owner/status evidence and repeated sweeps reuse the same // source-scoped action instead of creating issue-backed fallbacks. + // BLO-19160: the handover decision and every durable mutation that depends on it + // must share this transaction. The checkout/adoption paths take the same advisory + // lock, so an adoption either commits before this read (and is observed here) or + // waits until this transaction commits. The row lock then keeps non-advisory issue + // writers from changing the ownership snapshot while the recovery mutation runs. + // Keep all issue/action/comment/activity writes on `tx`; using the pooled `db` here + // would self-block on the row lock while this callback is awaiting that other query. + // Match `issuesSvc.update`'s graph-lock order: ownership arbitration, + // company graph lock, then the issue row. Taking the row first here and + // the graph lock inside `update` would invert against a concurrent + // blocker/parent mutation (which takes the graph lock before its row). + await lockIssueParentMutationCompany(input.issue.companyId, tx); const freshQuery = tx .select() .from(issues) .where(eq(issues.id, input.issue.id)) .limit(1); - const [fresh] = input.expectedReviewStage - ? await freshQuery.for("update") - : await freshQuery; + const [fresh] = await freshQuery.for("update"); if (!fresh) return null; // BLO-18643: mirror the park paths' own re-check (parkReviewWaitingContinuationIssue / // parkNoDependencyReviewWaitingIssue both bail if `fresh.status` has moved past the @@ -6077,30 +6312,10 @@ export function recoveryService( return null; } - // BLO-19160: same shape as the status CAS above, for the lock/owner - // columns. An adoption that commits between the handover observation and - // this transaction means a live run holds this issue's execution lock; - // escalating on the evidence read before it would reassign the issue away - // from that run and revoke the assignee's write access — the BLO-18860 - // failure mode through a narrower window. Bail rather than clobber. - // - // LIMITATION, measured — this NARROWS the window, it does not close it. - // `adoptStaleCheckoutRun` (services/issues.ts) takes no advisory lock; it - // serializes on a `select … for update` ROW lock of this row, so the two - // paths share no mutual-exclusion primitive and an adoption committing - // after this comparison is not excluded. The two obvious fixes both - // DEADLOCK here and were reverted after being measured: - // * `fresh` → `.for("update")`, and/or - // * routing the mutation below through `tx` - // Either one hangs `issue-recovery-actions.test.ts` at the 60s test - // timeout, because helpers between the read and the write touch this same - // issue row on the pooled `db` connection and block on the tx's lock. - // Closing it properly means threading `tx` through those helpers, which is - // BLO-18829's scope (`Stranded-escalation side effects escape when the - // expectedStatus CAS loses the race` — the identical defect class for the - // status CAS directly above). Until then: a *detected* race is side-effect - // free, because this check precedes the action upsert, quota monitor and - // wake enqueue. + // BLO-19160: re-check the handover evidence under the same lock before + // creating recovery state. This remains useful for callers whose candidate + // snapshot is already stale; the advisory/row lock closes the mutation-time + // race with checkout adoption. if ( input.expectedLockOwnerState && !issueLockOwnerStateMatches(input.expectedLockOwnerState, { @@ -6113,7 +6328,6 @@ export function recoveryService( } const recoveryCause = resolveStrandedRecoveryCause(input.latestRun, input.recoveryCause); - const mutationDb = input.expectedReviewStage ? tx : db; const { action, hasNewActivitySinceLastAttempt } = await ensureSourceScopedStrandedRecoveryAction({ issue: fresh, previousStatus: input.previousStatus, @@ -6121,71 +6335,30 @@ export function recoveryService( recoveryCause, recoveryOwnerAgentId: input.recoveryOwnerAgentId, successfulRunHandoffEvidence: input.successfulRunHandoffEvidence, - }, mutationDb); + }, tx); const isProviderQuotaWait = recoveryCause === "provider_quota" && !action.ownerAgentId && Boolean(action.returnOwnerAgentId); const { blockerIssueIds: blockerIds, needsHumanDecision, - } = await unresolvedBlockerHumanDecisionEscalationState(fresh.companyId, fresh.id, mutationDb); + } = await unresolvedBlockerHumanDecisionEscalationState(fresh.companyId, fresh.id, tx); - if (!input.expectedReviewStage) { - if (isProviderQuotaWait && action.returnOwnerAgentId) { - await ensureProviderQuotaWaitRecoveryMonitor({ - issue: fresh, - latestRun: input.latestRun, - actionId: action.id, - agentId: action.returnOwnerAgentId, - }); - } - await enqueueSourceScopedStrandedRecoveryWake({ - action, - issue: fresh, - latestRun: input.latestRun, - recoveryCause, - hasNewActivitySinceLastAttempt, - }); - } + // Wakes and quota monitors are intentionally deferred until after commit. + // Both can write issue/run state through another pooled connection and must + // never publish work for a transaction that later rolls back. const issueUpdate = { status: "blocked" as const, blockedByIssueIds: blockerIds, assigneeAgentId: action.ownerAgentId ?? fresh.assigneeAgentId, - // Do not overwrite an ordinary reassignment that commits after the - // lock-fresh read. The issue service repeats this snapshot in the SQL - // WHERE clause, so a blocked UPDATE is rejected against the latest - // assignee rather than applying the stale recovery decision. + expectedCurrentStatus: fresh.status, + // Keep the assignee snapshot as a belt-and-braces write precondition. expectedCurrentAssigneeAgentId: fresh.assigneeAgentId, }; - // That rejection arrives as a thrown 409, not a falsy return (see - // `isAssigneePreconditionConflict`), so it must be caught here. Letting it escape - // aborts the caller's whole reconcile batch and leaves every remaining stranded - // issue unreconciled — a far larger fault than the one issue we lost the race on. - let updated: Awaited>; - try { - updated = await issuesSvc.update(input.issue.id, issueUpdate, mutationDb); - } catch (error) { - if (!isAssigneePreconditionConflict(error)) throw error; - // Only the non-review path writes through `db`; the review-stage path writes through - // `tx`, where swallowing this would commit the action upsert without the matching - // issue UPDATE instead of rolling both back. That path also holds a `FOR UPDATE` row - // lock on the issue from its lock-fresh read, so a competing assignee write blocks - // until we commit and this conflict is unreachable there — but rethrow rather than - // trade an atomic rollback for a partial commit on an assumption. - if (input.expectedReviewStage) throw error; - logger.info( - { - issueId: fresh.id, - companyId: fresh.companyId, - expectedAssigneeAgentId: fresh.assigneeAgentId, - recoveryActionId: action.id, - recoveryCause, - }, - "skipping stranded recovery escalation: assignee changed before the update could be applied", - ); - return null; - } - if (!updated) return null; + // Any failed precondition rolls back the action and comments together with + // this issue update, so no partial recovery state escapes the transaction. + const updated = await issuesSvc.update(input.issue.id, issueUpdate, tx); + if (!updated) throw new RecoveryEscalationRollback(); if (isProviderQuotaWait) { return { updated, @@ -6273,7 +6446,7 @@ export function recoveryService( const escalationCommentMarker = announcesReassignment ? reassignmentMarker : `Recovery action: \`${action.id}\``; - const hasEscalationComment = await mutationDb + const hasEscalationComment = await tx .select({ id: issueComments.id, body: issueComments.body, metadata: issueComments.metadata }) .from(issueComments) .where(and(eq(issueComments.issueId, fresh.id), eq(issueComments.authorType, "system"))) @@ -6292,14 +6465,14 @@ export function recoveryService( authorType: "system", presentation: notice.presentation, metadata: notice.metadata, - }, mutationDb); + }, tx); } else { await issuesSvc.addComment( fresh.id, `${input.comment ?? "Automatic stranded-work recovery needs manual attention."}${recoveryLine}`, {}, { authorType: "system" }, - mutationDb, + tx, ); } } @@ -6334,7 +6507,7 @@ export function recoveryService( // is guaranteed to age the marker out and let a later sweep re-announce the same // exhaustion. Filtered by issue + author in SQL and capped at one row, so it costs // an index seek rather than the 50-row fetch it replaces. - const alreadyAnnounced = await mutationDb + const alreadyAnnounced = await tx .select({ id: issueComments.id }) .from(issueComments) .where(and( @@ -6375,12 +6548,12 @@ export function recoveryService( ].join("\n"), {}, { authorType: "system" }, - mutationDb, + tx, ); } } - const publishEscalationActivity = await logActivity(mutationDb as Db, { + const publishEscalationActivity = await logActivity(tx as unknown as Db, { companyId: fresh.companyId, actorType: "system", actorId: "system", @@ -6425,13 +6598,8 @@ export function recoveryService( blockerIssueIds: blockerIds, }, }, { - // Only the review-stage path passes a transaction as `mutationDb` (see the - // `input.expectedReviewStage ? tx : db` binding above). Deferring there hands the - // live/plugin publish back to the caller, which fires it after commit; on the - // non-review path `mutationDb` is the autocommit connection, so publishing inline - // is already correct and deferring would strand the event behind a caller that has - // nothing left to commit. - deferPublish: Boolean(input.expectedReviewStage), + // The activity row is transactional; publish only after its commit. + deferPublish: true, }); return { @@ -6449,13 +6617,14 @@ export function recoveryService( // marks this as a real strand, distinguishing it from the provider-quota park above. schedulerFailureHeartbeat: { prefix }, }; + }).catch((error) => { + if (error instanceof RecoveryEscalationRollback) return null; + throw error; }); if (!escalation) return null; - // `logActivity` deliberately deferred publication while the review-stage escalation was - // transactional. The state is committed now, so publish its live/plugin events before - // attempting the independent wake dispatch. A later dispatch failure leaves a durable - // recovery action, but must not make the successful escalation activity disappear. + // All durable state is committed now. Publish the activity before independent + // wake dispatch; a dispatch failure leaves the recovery action for the next sweep. escalation.publishEscalationActivity?.(); // The active recovery action committed above is the durable wake intent. @@ -6463,23 +6632,26 @@ export function recoveryService( // issue synchronously, and running it inside this transaction would either // self-block or publish work that an eventual rollback made inapplicable. // A failed dispatch leaves the action active for the next recovery sweep. - if (input.expectedReviewStage && escalation.recoveryCause === "provider_quota" && !escalation.action.ownerAgentId && escalation.action.returnOwnerAgentId) { + if (escalation.recoveryCause === "provider_quota" && !escalation.action.ownerAgentId && escalation.action.returnOwnerAgentId) { await ensureProviderQuotaWaitRecoveryMonitor({ - issue: escalation.fresh, + issue: escalation.updated, latestRun: input.latestRun, actionId: escalation.action.id, agentId: escalation.action.returnOwnerAgentId, }); } - if (input.expectedReviewStage) { - await enqueueSourceScopedStrandedRecoveryWake({ - action: escalation.action, - issue: escalation.fresh, - latestRun: input.latestRun, - recoveryCause: escalation.recoveryCause, - hasNewActivitySinceLastAttempt: escalation.hasNewActivitySinceLastAttempt, - }); - } + await enqueueSourceScopedStrandedRecoveryWake({ + action: escalation.action, + issue: escalation.fresh, + latestRun: input.latestRun, + recoveryCause: escalation.recoveryCause, + hasNewActivitySinceLastAttempt: escalation.hasNewActivitySinceLastAttempt, + expectedLockOwnerState: { + executionRunId: escalation.updated.executionRunId, + checkoutRunId: escalation.updated.checkoutRunId, + assigneeAgentId: escalation.updated.assigneeAgentId, + }, + }); if (escalation.needsHumanDecision) { const assigneeAgent = escalation.fresh.assigneeAgentId ? await db @@ -6549,47 +6721,74 @@ export function recoveryService( issue: typeof issues.$inferSelect; previousStatus: "todo" | "in_progress"; latestRun: LatestIssueRun; + expectedLockOwnerState?: IssueLockOwnerState | null; }) { - return await db.transaction(async (tx) => { + const result = await db.transaction(async (tx) => { // Serialize per (company, source-issue) so racing reconcile sweeps don't // double-escalate. Xact-scoped advisory lock, same key shape as // escalateStrandedAssignedIssue. - await tx.execute( - sql`select pg_advisory_xact_lock(hashtextextended(${input.issue.companyId} || ':' || ${input.issue.id}, 0))`, - ); + await lockIssueOwnership(tx, input.issue.companyId, input.issue.id); const [fresh] = await tx .select() .from(issues) .where(eq(issues.id, input.issue.id)) - .limit(1); + .limit(1) + .for("update"); if (!fresh) return null; // Peer sweep already escalated this source under the lock. if (fresh.status === "blocked") return fresh; + if (fresh.status !== input.previousStatus) return null; + if ( + input.expectedLockOwnerState && + !issueLockOwnerStateMatches(input.expectedLockOwnerState, { + executionRunId: fresh.executionRunId, + checkoutRunId: fresh.checkoutRunId, + assigneeAgentId: fresh.assigneeAgentId, + }) + ) return null; - const updated = await issuesSvc.update(input.issue.id, { status: "blocked" }); + let updated: Awaited>; + try { + updated = await issuesSvc.update( + fresh.id, + { + status: "blocked", + expectedCurrentStatus: fresh.status, + expectedCurrentAssigneeAgentId: fresh.assigneeAgentId, + expectedCurrentCheckoutRunId: fresh.checkoutRunId, + expectedCurrentExecutionRunId: fresh.executionRunId, + }, + tx, + ); + } catch (error) { + if (error instanceof HttpError && error.status === 409) return null; + throw error; + } if (!updated) return null; await issuesSvc.addComment( - input.issue.id, + fresh.id, buildZeroTokenStartupFailureComment({ previousStatus: input.previousStatus, latestRun: input.latestRun, }), {}, + { authorType: "system" }, + tx, ); - await logActivity(db, { - companyId: input.issue.companyId, + const publish = await logActivity(tx as unknown as Db, { + companyId: fresh.companyId, actorType: "system", actorId: "system", agentId: null, runId: null, action: "issue.updated", entityType: "issue", - entityId: input.issue.id, + entityId: fresh.id, details: { - identifier: input.issue.identifier, + identifier: fresh.identifier, status: "blocked", previousStatus: input.previousStatus, source: "recovery.reconcile_stranded_assigned_issue.zero_token_startup_failure", @@ -6598,10 +6797,13 @@ export function recoveryService( latestRunErrorCode: input.latestRun?.errorCode ?? null, recoveryWrapperSuppressed: true, }, - }); + }, { deferPublish: true }); - return updated; + return { updated, publish }; }); + if (!result) return null; + if ("publish" in result) result.publish(); + return "updated" in result ? result.updated : result; } // BLO-10889 (BLO-10866 WS2): defense-in-depth for the zero-token @@ -7631,6 +7833,7 @@ export function recoveryService( issue, previousStatus: "todo", latestRun, + expectedLockOwnerState: adoptionHandoverLockGuard, }); if (updated) { result.escalated += 1; @@ -7745,12 +7948,13 @@ export function recoveryService( continue; } if (isWaitingOnReviewContinuationRun(latestRun) && hasActiveMonitorPath(issue)) { - const updated = await parkReviewWaitingContinuationIssue({ + const parkOutcome = await parkReviewWaitingContinuationIssue({ issue, previousStatus: "in_progress", latestRun, + expectedLockOwnerState: adoptionHandoverLockGuard, }); - if (updated) { + if (parkOutcome === "parked") { result.reviewWaitingParked += 1; result.issueIds.push(issue.id); } else { @@ -7932,6 +8136,7 @@ export function recoveryService( issue, previousStatus: "in_progress", latestRun, + expectedLockOwnerState: adoptionHandoverLockGuard, }); if (updated) { result.escalated += 1; @@ -7981,6 +8186,7 @@ export function recoveryService( issue, previousStatus: "in_progress", latestRun, + expectedLockOwnerState: adoptionHandoverLockGuard, }); if (parkOutcome === "parked") { result.reviewWaitingParked += 1; @@ -7996,6 +8202,10 @@ export function recoveryService( result.skipped += 1; continue; } + if (parkOutcome === "lost_race") { + result.skipped += 1; + continue; + } // `failed` is a genuine park failure (evidence-gate rejection // because there's nothing reviewable yet, or a transient update // failure). Fall through to the normal blocked recovery path. From 24e8ab51bf40d2c43ded54639ed6af8d87f641f6 Mon Sep 17 00:00:00 2001 From: Omar Ramadan Date: Tue, 25 Aug 2026 13:49:43 +0000 Subject: [PATCH 3/3] fix(recovery): close handover wake races --- .../heartbeat-process-recovery.test.ts | 28 ++++----- .../__tests__/issue-recovery-actions.test.ts | 5 ++ server/src/services/heartbeat.ts | 10 ++++ server/src/services/recovery/service.ts | 57 ++++++++++++++++--- 4 files changed, 80 insertions(+), 20 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 0bbfe7cf8588..e9c8009916ff 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -6317,27 +6317,29 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { // No reverse `blocks` relation is seeded, so the recovery reachability check // (findCycleFormingBlockerIssueIds) will not flag openChildId as cycle-forming - it // only sees a cycle at write time, simulating a relation update that lands between - // the check and the write. The real write-time check (assertNoBlockingCycles) runs - // inside issuesSvc.update's db.transaction(), so the fault is injected on the first - // transaction the reconcile pass opens rather than on db.update directly (that - // transaction's tx.update/tx.select calls are a separate client, invisible to a - // db.update spy). - const transactionSpy = vi.spyOn(db, "transaction"); + // the check and the write. Inject the error at the review-wait helper's actual + // blocker-update boundary; mocking `db.transaction` would intercept the outer + // recovery transaction introduced for ownership serialization instead of the + // relation write that the helper catches. let cycleErrorThrown = false; - transactionSpy.mockImplementationOnce(async () => { + const beforeContinuationReviewBlockerUpdateForTest = vi.fn(async () => { cycleErrorThrown = true; throw new Error("Blocking relations cannot contain cycles"); }); - heartbeat = createHeartbeat({ penstockAvailabilityGate: allowPenstockGate }); + heartbeat = createHeartbeat({ + penstockAvailabilityGate: allowPenstockGate, + skipQueuedRunDispatch: true, + beforeContinuationReviewBlockerUpdateForTest, + }); let result: Awaited>; - try { - result = await heartbeat.reconcileStrandedAssignedIssues(); - } finally { - transactionSpy.mockRestore(); - } + result = await heartbeat.reconcileStrandedAssignedIssues(); expect(cycleErrorThrown).toBe(true); + expect(beforeContinuationReviewBlockerUpdateForTest).toHaveBeenCalledWith({ + issueId, + blockedByIssueIds: [openChildId], + }); expect(result.waitingOnReviewResolved).toBe(0); expect(result.reviewWaitingParked).toBe(1); expect(result.escalated).toBe(0); diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 48ecc05c68f4..c0eb3d92f3e6 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -856,6 +856,11 @@ describeEmbeddedPostgres("issue recovery actions", () => { recoveryActionId: action!.id, backstop: "stranded_recovery_wake_backstop", }, + expectedLockOwnerState: { + executionRunId: null, + checkoutRunId: null, + assigneeAgentId: managerId, + }, }); }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 173ab087d0b8..ad3f4ac4b1ca 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -9882,6 +9882,15 @@ export interface HeartbeatServiceOptions { skipQueuedRunDispatch?: boolean; /** Test-only hook for exercising the monitor select/claim race. */ issueMonitorClaimHook?: (issueId: string) => Promise; + /** + * Test-only seam for the review-wait blocker relation write. Production + * leaves this unset; tests use it to inject a write-time cycle error at the + * recovery helper's actual update boundary. + */ + beforeContinuationReviewBlockerUpdateForTest?: (input: { + issueId: string; + blockedByIssueIds: string[]; + }) => Promise | void; /** * Node role for this process (mirrors config.paperclipNodeRole; wired from * index.ts). On the "api" tier, run dispatch (claim + `executeRun`) is fenced @@ -10172,6 +10181,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const recovery = recoveryService(db, { enqueueWakeup, beforeStaleIssueLockSweepClearForTest: options.beforeStaleIssueLockSweepClearForTest, + beforeContinuationReviewBlockerUpdateForTest: options.beforeContinuationReviewBlockerUpdateForTest, }); function isPlanApprovalConfirmationPayload(payload: unknown) { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 6889327a0869..1cb6ae5f3ae7 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1722,6 +1722,16 @@ export function recoveryService( executionLockedAt: Date | null; }, ) => Promise | void; + /** + * Test-only seam for the review-wait blocker relation write. Production + * leaves this unset; tests use it to inject the same cycle error that the + * issue service can raise after a concurrent relation update, without + * intercepting the recovery transaction itself. + */ + beforeContinuationReviewBlockerUpdateForTest?: (input: { + issueId: string; + blockedByIssueIds: string[]; + }) => Promise | void; }, ) { const issuesSvc = issueService(db); @@ -5973,6 +5983,10 @@ export function recoveryService( let updated: Awaited>; try { + await deps.beforeContinuationReviewBlockerUpdateForTest?.({ + issueId: fresh.id, + blockedByIssueIds, + }); updated = await issuesSvc.update( fresh.id, { @@ -6054,8 +6068,9 @@ export function recoveryService( const [fresh] = await tx .select() .from(issues) - .where(eq(issues.id, issue.id)) - .limit(1); + .where(and(eq(issues.companyId, issue.companyId), eq(issues.id, issue.id))) + .limit(1) + .for("update"); if (!fresh || isTerminalIssueStatus(fresh.status)) return null; // BLO-27572: explicitly suppress the scheduler-side failure heartbeat that @@ -6063,11 +6078,27 @@ export function recoveryService( // cancellation that must stay silent: the window is NOT dark, because // another open execution issue already owns the dispatch lock and is doing // the work. A receipt here would manufacture a false dark-window alarm. - const updated = await issuesSvc.update( - fresh.id, - { status: "cancelled", suppressRoutineSchedulerFailureHeartbeat: true }, - tx, - ); + let updated: Awaited>; + try { + updated = await issuesSvc.update( + fresh.id, + { + status: "cancelled", + suppressRoutineSchedulerFailureHeartbeat: true, + expectedCurrentStatus: fresh.status, + expectedCurrentAssigneeAgentId: fresh.assigneeAgentId, + expectedCurrentCheckoutRunId: fresh.checkoutRunId, + expectedCurrentExecutionRunId: fresh.executionRunId, + }, + tx, + ); + } catch (error) { + // A concurrent owner/status writer won after the recovery snapshot. This + // duplicate is no longer ours to cancel; importantly, do not append the + // cancellation activity or comment for a mutation that did not happen. + if (error instanceof HttpError && error.status === 409) return null; + throw error; + } if (!updated) return null; const publish = await logActivity(tx as unknown as Db, { @@ -10212,6 +10243,13 @@ export function recoveryService( actionLastAttemptAt: issueRecoveryActions.lastAttemptAt, issueId: issues.id, issueStatus: issues.status, + // Preserve the ownership tuple observed with the candidate. The backstop + // claims the recovery action before enqueueing, so adoption can commit in + // the gap between those operations; enqueueWakeup performs the final + // ownership CAS under the issue lock. + executionRunId: issues.executionRunId, + checkoutRunId: issues.checkoutRunId, + assigneeAgentId: issues.assigneeAgentId, identifier: issues.identifier, totalCount: sql`count(*) over()::int`, }) @@ -10390,6 +10428,11 @@ export function recoveryService( recoveryCause: candidate.actionCause, backstop: "stranded_recovery_wake_backstop", }, "status_only"), + expectedLockOwnerState: { + executionRunId: candidate.executionRunId, + checkoutRunId: candidate.checkoutRunId, + assigneeAgentId: candidate.assigneeAgentId, + }, }); if (!wake) { result.deferredOrFailed += 1;