From d8b1355c333e6ee13806d950d32e07212eebf29a Mon Sep 17 00:00:00 2001 From: Paperclip CTO Date: Wed, 5 Aug 2026 15:33:15 +0000 Subject: [PATCH 01/15] fix(recovery): stop pod-never-scheduled failures burning recovery budget (BLO-19889) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `k8s_pod_schedule_failed` means the agent pod never bound to a node, so the adapter was never invoked and no work product exists. It was not in TRANSIENT_INFRA_CONTINUATION_ERROR_CODES, so classifyContinuationFailure returned `default` (maxAttempts 1, no backoff) and the second consecutive failure hit escalateStrandedAssignedIssue — which both spends a stranded-recovery attempt and reassigns the issue up the org chain, for a cause no owner in that chain can act on. heartbeat.ts's shouldScheduleAutomaticRunRetry only re-queues this code for pr_review wakes, so an *issue* run had no retry engine at all. Same safety shape as `process_lost` (BLO-16182): bounded 3 attempts with 60s exponential backoff instead of instant escalate. `job_missing` is handled separately and evidence-gated: the external lifecycle Job can vanish *after* a non-idempotent side effect (BLO-18106), so it is treated as transient infra only when the reconciler durably proved `adapterInvocationStarted === false` — mirroring the existing `job_failed` gate. Missing or non-boolean evidence falls through to `default`. Work-class failures are unchanged and still escalate on the next attempt. Co-Authored-By: Claude --- .../service.infra-class-continuation.test.ts | 77 +++++++++++++++++++ server/src/services/recovery/service.ts | 30 ++++++++ 2 files changed, 107 insertions(+) create mode 100644 server/src/services/recovery/service.infra-class-continuation.test.ts diff --git a/server/src/services/recovery/service.infra-class-continuation.test.ts b/server/src/services/recovery/service.infra-class-continuation.test.ts new file mode 100644 index 000000000000..776e465ab35b --- /dev/null +++ b/server/src/services/recovery/service.infra-class-continuation.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, it } from "vitest"; +import { classifyContinuationFailure } from "./service.js"; + +// BLO-19889: infra-class run failures (the agent pod never ran) must not spend the +// stranded-recovery budget or bounce the issue up the org chain. +// +// Why classifyContinuationFailure is the right unit under test: its `maxAttempts` is +// the sole gate on the escalation in reconcileStrandedAssignedIssues +// (service.ts, the `consecutive >= classification.maxAttempts` branch). That branch +// is the ONLY caller of escalateStrandedAssignedIssue on the continuation path, and +// escalateStrandedAssignedIssue is what does BOTH things this ticket is about: +// - ensureSourceScopedStrandedRecoveryAction(...) -> attemptCount + 1 +// - issuesSvc.update(..., { status: "blocked", assigneeAgentId: action.ownerAgentId }) +// So `kind: "default"` (maxAttempts 1) means "escalate + reassign on the 2nd +// consecutive failure"; `kind: "transient_infra"` (maxAttempts 3) means "keep +// re-dispatching with backoff, spend nothing, reassign nobody". + +type Run = Parameters[0]; + +const run = (errorCode: string | null) => ({ errorCode } as unknown as Run); + +const jobMissingRun = (adapterInvocationStarted: unknown) => + ({ + errorCode: "job_missing", + resultJson: { externalLifecycleRecovery: { adapterInvocationStarted } }, + } as unknown as Run); + +describe("BLO-19889: infra-class continuation failures do not spend recovery budget", () => { + it("k8s_pod_schedule_failed is infra-class: bounded retry, not a 1-attempt escalate", () => { + // 0/17 nodes available (Unschedulable / Insufficient cpu / untolerated taint). + // The pod never bound to a node, so the adapter was never invoked and there is + // no work product -- identical safety shape to process_lost. + const c = classifyContinuationFailure(run("k8s_pod_schedule_failed")); + expect(c.kind).toBe("transient_infra"); + expect(c.maxAttempts).toBeGreaterThan(1); + }); + + it("work-class control: an ordinary run failure still escalates on the next attempt", () => { + // The paired assertion the AC asks for. A work-class failure keeps the old + // behavior -- one attempt, then escalateStrandedAssignedIssue (attempt burn + + // reassignment). If this ever flips to transient_infra, the change above has + // over-reached and genuinely stuck work would silently retry forever. + const workClass = classifyContinuationFailure(run("some_adapter_error")); + expect(workClass.kind).toBe("default"); + expect(workClass.maxAttempts).toBe(1); + + // ...and it is strictly cheaper to retry infra-class than work-class. + expect(classifyContinuationFailure(run("k8s_pod_schedule_failed")).maxAttempts) + .toBeGreaterThan(workClass.maxAttempts); + }); + + it("non-retryable codes are unaffected by the infra-class widening", () => { + expect(classifyContinuationFailure(run("agent_not_invokable")).kind).toBe("non_retryable"); + expect(classifyContinuationFailure(run("budget_blocked")).kind).toBe("non_retryable"); + }); + + describe("job_missing is evidence-gated, not unconditionally infra-class", () => { + it("re-dispatches only when the reconciler proved the adapter never started", () => { + const c = classifyContinuationFailure(jobMissingRun(false)); + expect(c.kind).toBe("transient_infra"); + expect(c.maxAttempts).toBeGreaterThan(1); + }); + + it("does NOT re-dispatch when the adapter had already been invoked", () => { + // BLO-18106: the external lifecycle Job can vanish *after* a non-idempotent + // side effect. Re-running would duplicate that work, so this must stay on the + // escalate path even though the error code looks infra-shaped. + expect(classifyContinuationFailure(jobMissingRun(true)).kind).toBe("default"); + }); + + it("fails safe when the invocation evidence is missing or not a boolean", () => { + expect(classifyContinuationFailure(jobMissingRun(undefined)).kind).toBe("default"); + expect(classifyContinuationFailure(jobMissingRun("false")).kind).toBe("default"); + expect(classifyContinuationFailure(run("job_missing")).kind).toBe("default"); + }); + }); +}); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 7a9c817f5e8c..249b6fff4e1a 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -812,8 +812,29 @@ const TRANSIENT_INFRA_CONTINUATION_ERROR_CODES = new Set([ // instead of the `default` single-attempt/instant-escalate path, so a lone // control-plane blip no longer strands the issue as `blocked`. "process_lost", + // BLO-19889: the agent pod was never scheduled onto a node (Unschedulable / + // image-pull / schedule-timeout), so the adapter was never invoked and no work + // product exists — the same "died before any model call" shape as process_lost + // above. Previously this fell through to `default` (1 attempt, no backoff), so a + // single node-pool saturation burst escalated the issue to `blocked` AND + // reassigned it up the org chain, for a cause no owner in that chain can act on. + // shouldScheduleAutomaticRunRetry only re-queues this code for pr_review wakes + // (heartbeat.ts), so before this entry an *issue* run had no retry engine at all. + "k8s_pod_schedule_failed", ]); +// BLO-19889: `job_missing` is NOT unconditionally infra-class. The external +// lifecycle Job can vanish *after* the adapter already performed non-idempotent +// work (that is the whole subject of BLO-18106), so blind re-dispatch is unsafe. +// Mirror the `job_failed` gate in shouldScheduleAutomaticRunRetry: treat it as +// transient infra only when the reconciler durably proved invocation never began. +// Absent/unknown evidence falls through to `default` — fail-safe by construction. +function isNeverInvokedJobMissingRun(latestRun: LatestIssueRun) { + if (readNonEmptyString(latestRun?.errorCode) !== "job_missing") return false; + const recovery = parseObject(parseObject(latestRun?.resultJson).externalLifecycleRecovery); + return recovery.adapterInvocationStarted === false; +} + const NON_RETRYABLE_CONTINUATION_ERROR_CODES = new Set([ "agent_not_invokable", "agent_not_found", @@ -1048,6 +1069,15 @@ export function classifyContinuationFailure(latestRun: LatestIssueRun): Continua errorCode, }; } + // BLO-19889: evidence-gated, so it cannot live in the flat code set above. + if (isNeverInvokedJobMissingRun(latestRun)) { + return { + kind: "transient_infra", + maxAttempts: CONTINUATION_RECOVERY_TRANSIENT_MAX_ATTEMPTS, + baseBackoffMs: CONTINUATION_RECOVERY_TRANSIENT_BASE_BACKOFF_MS, + errorCode, + }; + } return { kind: "default", maxAttempts: CONTINUATION_RECOVERY_DEFAULT_MAX_ATTEMPTS, From b5efe28cc682a80dd2b9d42e3c5eaf3fe6e0a4cb Mon Sep 17 00:00:00 2001 From: Paperclip Release Engineer Date: Wed, 5 Aug 2026 16:40:59 +0000 Subject: [PATCH 02/15] fix(recovery): persist job-missing invocation evidence Keep k8s_pod_schedule_failed on the fail-safe default path because adapters may emit it after main-container execution. Persist adapter.invoke evidence when the missing-Job reconciler finalizes a run, and cover the real reaper record plus classifier controls. Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 5 +++- server/src/services/heartbeat.ts | 3 ++- .../service.infra-class-continuation.test.ts | 24 +++++++------------ server/src/services/recovery/service.ts | 13 ++-------- 4 files changed, 17 insertions(+), 28 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 6f367f29c222..c97c62102f1b 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1749,7 +1749,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(run?.errorCode).toBe("process_lost"); }); - it("immediately reaps a fresh exact-missing Job after restart when no adapter owner remains", async () => { + it("immediately reaps a fresh exact-missing Job and records that adapter invocation started", async () => { const jobName = "agent-opencode-restart-missing"; const { companyId, agentId, runId } = await seedRunFixture({ adapterType: "opencode_k8s", @@ -1779,6 +1779,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(await heartbeat.getRun(runId)).toMatchObject({ status: "failed", errorCode: "job_missing", + resultJson: { + externalLifecycleRecovery: expect.objectContaining({ adapterInvocationStarted: true }), + }, }); const persistedReservation = await db .select() diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 59e4decd6dd2..4cf739a7eeaa 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -16599,7 +16599,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) : baseTerminalOutcome; if (!terminalOutcome) return false; - const adapterInvocationStarted = terminalOutcome.errorCode === "job_failed" + const adapterInvocationStarted = + terminalOutcome.errorCode === "job_failed" || terminalOutcome.errorCode === "job_missing" ? await hasAdapterInvocationEvent(input.run.id) : null; diff --git a/server/src/services/recovery/service.infra-class-continuation.test.ts b/server/src/services/recovery/service.infra-class-continuation.test.ts index 776e465ab35b..01f7853a33e0 100644 --- a/server/src/services/recovery/service.infra-class-continuation.test.ts +++ b/server/src/services/recovery/service.infra-class-continuation.test.ts @@ -1,8 +1,8 @@ import { describe, expect, it } from "vitest"; import { classifyContinuationFailure } from "./service.js"; -// BLO-19889: infra-class run failures (the agent pod never ran) must not spend the -// stranded-recovery budget or bounce the issue up the org chain. +// Infra-class run failures must not spend the stranded-recovery budget or bounce +// the issue up the org chain. // // Why classifyContinuationFailure is the right unit under test: its `maxAttempts` is // the sole gate on the escalation in reconcileStrandedAssignedIssues @@ -25,16 +25,7 @@ const jobMissingRun = (adapterInvocationStarted: unknown) => resultJson: { externalLifecycleRecovery: { adapterInvocationStarted } }, } as unknown as Run); -describe("BLO-19889: infra-class continuation failures do not spend recovery budget", () => { - it("k8s_pod_schedule_failed is infra-class: bounded retry, not a 1-attempt escalate", () => { - // 0/17 nodes available (Unschedulable / Insufficient cpu / untolerated taint). - // The pod never bound to a node, so the adapter was never invoked and there is - // no work product -- identical safety shape to process_lost. - const c = classifyContinuationFailure(run("k8s_pod_schedule_failed")); - expect(c.kind).toBe("transient_infra"); - expect(c.maxAttempts).toBeGreaterThan(1); - }); - +describe("BLO-18106: job_missing continuation recovery is evidence-gated", () => { it("work-class control: an ordinary run failure still escalates on the next attempt", () => { // The paired assertion the AC asks for. A work-class failure keeps the old // behavior -- one attempt, then escalateStrandedAssignedIssue (attempt burn + @@ -44,9 +35,12 @@ describe("BLO-19889: infra-class continuation failures do not spend recovery bud expect(workClass.kind).toBe("default"); expect(workClass.maxAttempts).toBe(1); - // ...and it is strictly cheaper to retry infra-class than work-class. - expect(classifyContinuationFailure(run("k8s_pod_schedule_failed")).maxAttempts) - .toBeGreaterThan(workClass.maxAttempts); + // k8s_pod_schedule_failed can be emitted after the main container starts, so + // it must not be replayed without stronger producer evidence. + expect(classifyContinuationFailure(run("k8s_pod_schedule_failed"))).toMatchObject({ + kind: "default", + maxAttempts: workClass.maxAttempts, + }); }); it("non-retryable codes are unaffected by the infra-class widening", () => { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 249b6fff4e1a..dcadbea2c074 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -812,18 +812,9 @@ const TRANSIENT_INFRA_CONTINUATION_ERROR_CODES = new Set([ // instead of the `default` single-attempt/instant-escalate path, so a lone // control-plane blip no longer strands the issue as `blocked`. "process_lost", - // BLO-19889: the agent pod was never scheduled onto a node (Unschedulable / - // image-pull / schedule-timeout), so the adapter was never invoked and no work - // product exists — the same "died before any model call" shape as process_lost - // above. Previously this fell through to `default` (1 attempt, no backoff), so a - // single node-pool saturation burst escalated the issue to `blocked` AND - // reassigned it up the org chain, for a cause no owner in that chain can act on. - // shouldScheduleAutomaticRunRetry only re-queues this code for pr_review wakes - // (heartbeat.ts), so before this entry an *issue* run had no retry engine at all. - "k8s_pod_schedule_failed", ]); -// BLO-19889: `job_missing` is NOT unconditionally infra-class. The external +// BLO-18106: `job_missing` is NOT unconditionally infra-class. The external // lifecycle Job can vanish *after* the adapter already performed non-idempotent // work (that is the whole subject of BLO-18106), so blind re-dispatch is unsafe. // Mirror the `job_failed` gate in shouldScheduleAutomaticRunRetry: treat it as @@ -1069,7 +1060,7 @@ export function classifyContinuationFailure(latestRun: LatestIssueRun): Continua errorCode, }; } - // BLO-19889: evidence-gated, so it cannot live in the flat code set above. + // BLO-18106: evidence-gated, so it cannot live in the flat code set above. if (isNeverInvokedJobMissingRun(latestRun)) { return { kind: "transient_infra", From c354431e7e9cfc73cc95fa9e924c694836dfde32 Mon Sep 17 00:00:00 2001 From: Paperclip Release Engineer Date: Wed, 5 Aug 2026 17:31:50 +0000 Subject: [PATCH 03/15] fix(recovery): align job missing retry evidence Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 5 +- .../heartbeat-retry-scheduling.test.ts | 96 +++++++------------ server/src/services/heartbeat.ts | 18 +--- .../service.infra-class-continuation.test.ts | 31 ++---- server/src/services/recovery/service.ts | 21 ---- 5 files changed, 53 insertions(+), 118 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index c97c62102f1b..589565fadf0b 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -258,6 +258,7 @@ import { INTERACTION_CONTINUATION_INFRA_WAKE_REASON, heartbeatService, redactDetectedSuccessfulRunProgressSummaryForBoard, + shouldScheduleAutomaticRunRetry, } from "../services/heartbeat.ts"; import { setPluginEventBus, setPluginEventOutboxDb } from "../services/activity-log.js"; import { pollOnce as drainPluginEventOutbox } from "../services/plugin-event-outbox.js"; @@ -1776,13 +1777,15 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const result = await heartbeat.reapOrphanedRuns({ suppressDispatchAfterReap: true }); expect(result.runIds).toContain(runId); - expect(await heartbeat.getRun(runId)).toMatchObject({ + const finalizedRun = await heartbeat.getRun(runId); + expect(finalizedRun).toMatchObject({ status: "failed", errorCode: "job_missing", resultJson: { externalLifecycleRecovery: expect.objectContaining({ adapterInvocationStarted: true }), }, }); + expect(finalizedRun && shouldScheduleAutomaticRunRetry(finalizedRun)).toBe(false); const persistedReservation = await db .select() .from(externalRuntimeReservations) diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 33ef3b8cf5ad..52b08818b3f9 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -1874,74 +1874,23 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { ).toBe(false); }); - it("retries job_failed only when durable evidence proves adapter invocation never began", () => { - expect( - shouldScheduleAutomaticRunRetry({ - errorCode: "job_failed", - resultJson: { externalLifecycleRecovery: { adapterInvocationStarted: false } }, - contextSnapshot: { issueId: randomUUID(), wakeReason: "issue_assigned" }, - }), - ).toBe(true); - expect( - shouldScheduleAutomaticRunRetry({ - errorCode: "job_failed", - resultJson: { externalLifecycleRecovery: { adapterInvocationStarted: true } }, - contextSnapshot: { issueId: randomUUID(), wakeReason: "issue_assigned" }, - }), - ).toBe(false); - expect( - shouldScheduleAutomaticRunRetry({ - errorCode: "job_failed", - resultJson: {}, - contextSnapshot: { issueId: randomUUID(), wakeReason: "issue_assigned" }, - }), - ).toBe(false); - expect( - shouldScheduleAutomaticRunRetry({ - errorCode: "job_failed", - resultJson: {}, - contextSnapshot: { wakeReason: "heartbeat_timer" }, - }), - ).toBe(false); - expect( - shouldScheduleAutomaticRunRetry({ - errorCode: "job_failed", - resultJson: {}, - contextSnapshot: null, - }), - ).toBe(false); - expect(JOB_FAILED_HEARTBEAT_RETRY_MAX_ATTEMPTS).toBe(4); - }); - - it("BLO-9147 AC2: CAPACITY_BLOCKED_HEARTBEAT_RETRY_MAX_ATTEMPTS exceeds rate-limit cap (12)", () => { - expect(CAPACITY_BLOCKED_HEARTBEAT_RETRY_MAX_ATTEMPTS).toBeGreaterThan(12); - }); - - // BLO-10448 — scheduler-level transient infra failures retry gate - it.each(["k8s_pod_schedule_failed", "job_missing"])( - "BLO-10448: retries %s on a pr_review wake (work never ran)", + it.each(["job_failed", "job_missing"])( + "retries %s only when durable evidence proves adapter invocation never began", (errorCode) => { expect( shouldScheduleAutomaticRunRetry({ errorCode, - resultJson: {}, - contextSnapshot: { wakeReason: "github_pr_opened", reviewKind: "pr_review", githubPrNumber: 408 }, + resultJson: { externalLifecycleRecovery: { adapterInvocationStarted: false } }, + contextSnapshot: { issueId: randomUUID(), wakeReason: "issue_assigned" }, }), ).toBe(true); - // thin snapshot (taskKey-only) — webhook-driven reviewer wakes get trimmed expect( shouldScheduleAutomaticRunRetry({ errorCode, - resultJson: {}, - contextSnapshot: { taskKey: "pr_review:Blockcast/Network-Operator-Portal:408" }, + resultJson: { externalLifecycleRecovery: { adapterInvocationStarted: true } }, + contextSnapshot: { issueId: randomUUID(), wakeReason: "issue_assigned" }, }), - ).toBe(true); - }, - ); - - it.each(["k8s_pod_schedule_failed", "job_missing"])( - "BLO-10448: does NOT retry %s on non-PR wakes (BLO-7913 leak guard)", - (errorCode) => { + ).toBe(false); expect( shouldScheduleAutomaticRunRetry({ errorCode, @@ -1953,12 +1902,41 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { shouldScheduleAutomaticRunRetry({ errorCode, resultJson: {}, - contextSnapshot: {}, + contextSnapshot: { wakeReason: "heartbeat_timer" }, + }), + ).toBe(false); + expect( + shouldScheduleAutomaticRunRetry({ + errorCode, + resultJson: {}, + contextSnapshot: null, }), ).toBe(false); + expect(JOB_FAILED_HEARTBEAT_RETRY_MAX_ATTEMPTS).toBe(4); }, ); + it("BLO-9147 AC2: CAPACITY_BLOCKED_HEARTBEAT_RETRY_MAX_ATTEMPTS exceeds rate-limit cap (12)", () => { + expect(CAPACITY_BLOCKED_HEARTBEAT_RETRY_MAX_ATTEMPTS).toBeGreaterThan(12); + }); + + it("does not retry ambiguous k8s_pod_schedule_failed outcomes", () => { + for (const contextSnapshot of [ + { wakeReason: "github_pr_opened", reviewKind: "pr_review", githubPrNumber: 408 }, + { taskKey: "pr_review:Blockcast/Network-Operator-Portal:408" }, + { issueId: randomUUID(), wakeReason: "issue_assigned" }, + {}, + ]) { + expect( + shouldScheduleAutomaticRunRetry({ + errorCode: "k8s_pod_schedule_failed", + resultJson: {}, + contextSnapshot, + }), + ).toBe(false); + } + }); + // BLO-17456: when a PR-review chain exhausts, the reviewer never posts its // required status, so the PR sits on "Expected — waiting for status" forever. // These drive the real exhaustion path (no mocks): loadConfig() reads diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 4cf739a7eeaa..9bde30d3f4ba 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1036,26 +1036,14 @@ export function shouldScheduleAutomaticRunRetry( return isIssueRun || isPrReviewRetryContext(contextSnapshot); } - // A failed external-lifecycle Job may have performed non-idempotent work. - // Retry only when the reconciler durably proved adapter invocation never + // A failed or missing external-lifecycle Job may have performed non-idempotent + // work. Retry only when the reconciler durably proved adapter invocation never // began; lock/status gates alone cannot make partial external writes safe. - if (run.errorCode === "job_failed") { + if (run.errorCode === "job_failed" || run.errorCode === "job_missing") { const recovery = parseObject(parseObject(run.resultJson).externalLifecycleRecovery); return isIssueRun && recovery.adapterInvocationStarted === false; } - // BLO-10448: scheduler-level transient infra failures where the agent pod - // never ran — the node pool was momentarily saturated (k8s_pod_schedule_failed: - // Unschedulable / image-pull / schedule-timeout) or the external-lifecycle Job - // vanished before completion (job_missing). The work never started, so re-queue - // pr_review wakes with bounded backoff to let the review land once capacity - // frees, instead of silently dropping it (observed on the Ally reviewer path: - // a single Unschedulable burst dropped a PR review with no retry). Non-PR wakes - // stay terminal, matching the k8s_concurrent_run_blocked leak guard above. - if (run.errorCode === "k8s_pod_schedule_failed" || run.errorCode === "job_missing") { - return isPrReviewRetryContext(parseObject(run.contextSnapshot)); - } - if (run.errorCode !== "adapter_failed" && run.errorCode !== "process_lost") return false; // BLO-9147 AC1: gate on wakeReason/reviewKind/taskKey from the persisted diff --git a/server/src/services/recovery/service.infra-class-continuation.test.ts b/server/src/services/recovery/service.infra-class-continuation.test.ts index 01f7853a33e0..7bd050641555 100644 --- a/server/src/services/recovery/service.infra-class-continuation.test.ts +++ b/server/src/services/recovery/service.infra-class-continuation.test.ts @@ -19,12 +19,6 @@ type Run = Parameters[0]; const run = (errorCode: string | null) => ({ errorCode } as unknown as Run); -const jobMissingRun = (adapterInvocationStarted: unknown) => - ({ - errorCode: "job_missing", - resultJson: { externalLifecycleRecovery: { adapterInvocationStarted } }, - } as unknown as Run); - describe("BLO-18106: job_missing continuation recovery is evidence-gated", () => { it("work-class control: an ordinary run failure still escalates on the next attempt", () => { // The paired assertion the AC asks for. A work-class failure keeps the old @@ -48,24 +42,17 @@ describe("BLO-18106: job_missing continuation recovery is evidence-gated", () => expect(classifyContinuationFailure(run("budget_blocked")).kind).toBe("non_retryable"); }); - describe("job_missing is evidence-gated, not unconditionally infra-class", () => { - it("re-dispatches only when the reconciler proved the adapter never started", () => { - const c = classifyContinuationFailure(jobMissingRun(false)); - expect(c.kind).toBe("transient_infra"); - expect(c.maxAttempts).toBeGreaterThan(1); - }); - - it("does NOT re-dispatch when the adapter had already been invoked", () => { - // BLO-18106: the external lifecycle Job can vanish *after* a non-idempotent - // side effect. Re-running would duplicate that work, so this must stay on the - // escalate path even though the error code looks infra-shaped. - expect(classifyContinuationFailure(jobMissingRun(true)).kind).toBe("default"); + it("keeps job_missing on the fail-safe path because production emits it only after invocation", () => { + expect(classifyContinuationFailure(run("job_missing"))).toMatchObject({ + kind: "default", + maxAttempts: 1, }); - it("fails safe when the invocation evidence is missing or not a boolean", () => { - expect(classifyContinuationFailure(jobMissingRun(undefined)).kind).toBe("default"); - expect(classifyContinuationFailure(jobMissingRun("false")).kind).toBe("default"); - expect(classifyContinuationFailure(run("job_missing")).kind).toBe("default"); + // Pre-invocation disappearance is persisted as process_lost, which remains + // the reachable bounded-retry path for work that provably never started. + expect(classifyContinuationFailure(run("process_lost"))).toMatchObject({ + kind: "transient_infra", + maxAttempts: 3, }); }); }); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index dcadbea2c074..7a9c817f5e8c 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -814,18 +814,6 @@ const TRANSIENT_INFRA_CONTINUATION_ERROR_CODES = new Set([ "process_lost", ]); -// BLO-18106: `job_missing` is NOT unconditionally infra-class. The external -// lifecycle Job can vanish *after* the adapter already performed non-idempotent -// work (that is the whole subject of BLO-18106), so blind re-dispatch is unsafe. -// Mirror the `job_failed` gate in shouldScheduleAutomaticRunRetry: treat it as -// transient infra only when the reconciler durably proved invocation never began. -// Absent/unknown evidence falls through to `default` — fail-safe by construction. -function isNeverInvokedJobMissingRun(latestRun: LatestIssueRun) { - if (readNonEmptyString(latestRun?.errorCode) !== "job_missing") return false; - const recovery = parseObject(parseObject(latestRun?.resultJson).externalLifecycleRecovery); - return recovery.adapterInvocationStarted === false; -} - const NON_RETRYABLE_CONTINUATION_ERROR_CODES = new Set([ "agent_not_invokable", "agent_not_found", @@ -1060,15 +1048,6 @@ export function classifyContinuationFailure(latestRun: LatestIssueRun): Continua errorCode, }; } - // BLO-18106: evidence-gated, so it cannot live in the flat code set above. - if (isNeverInvokedJobMissingRun(latestRun)) { - return { - kind: "transient_infra", - maxAttempts: CONTINUATION_RECOVERY_TRANSIENT_MAX_ATTEMPTS, - baseBackoffMs: CONTINUATION_RECOVERY_TRANSIENT_BASE_BACKOFF_MS, - errorCode, - }; - } return { kind: "default", maxAttempts: CONTINUATION_RECOVERY_DEFAULT_MAX_ATTEMPTS, From de516167608d8e472fcccfe172438828629797a9 Mon Sep 17 00:00:00 2001 From: Release Engineer Date: Wed, 5 Aug 2026 18:26:08 +0000 Subject: [PATCH 04/15] fix(recovery): prevent job-missing continuation replay Co-Authored-By: Paperclip --- .../heartbeat-retry-scheduling.test.ts | 12 +++++- .../__tests__/issue-recovery-actions.test.ts | 37 +++++++++++++++++++ server/src/services/heartbeat.ts | 10 +++-- .../service.infra-class-continuation.test.ts | 6 +-- server/src/services/recovery/service.ts | 3 ++ 5 files changed, 60 insertions(+), 8 deletions(-) diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 52b08818b3f9..8c9d61ae93e3 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -1874,7 +1874,7 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { ).toBe(false); }); - it.each(["job_failed", "job_missing"])( + it.each(["job_failed"])( "retries %s only when durable evidence proves adapter invocation never began", (errorCode) => { expect( @@ -1916,6 +1916,16 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { }, ); + it("does not retry job_missing even with synthetic never-invoked evidence", () => { + expect( + shouldScheduleAutomaticRunRetry({ + errorCode: "job_missing", + resultJson: { externalLifecycleRecovery: { adapterInvocationStarted: false } }, + contextSnapshot: { issueId: randomUUID(), wakeReason: "issue_assigned" }, + }), + ).toBe(false); + }); + it("BLO-9147 AC2: CAPACITY_BLOCKED_HEARTBEAT_RETRY_MAX_ATTEMPTS exceeds rate-limit cap (12)", () => { expect(CAPACITY_BLOCKED_HEARTBEAT_RETRY_MAX_ATTEMPTS).toBeGreaterThan(12); }); diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 9be0c9451eff..04e28861f834 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -483,6 +483,43 @@ describeEmbeddedPostgres("issue recovery actions", () => { expect(await svc.getActiveForIssue(randomUUID(), sourceIssueId)).toBeNull(); }); + it("does not enqueue continuation work after an invoked external Job disappears", async () => { + const { companyId, coderId, sourceIssueId } = await seedCompany(); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId: coderId, + invocationSource: "automation", + status: "failed", + error: "External lifecycle Job is missing while heartbeat run is still running", + errorCode: "job_missing", + resultJson: { + externalLifecycleRecovery: { adapterInvocationStarted: true }, + }, + contextSnapshot: { issueId: sourceIssueId }, + startedAt: new Date("2026-07-26T13:45:00.000Z"), + finishedAt: new Date("2026-07-26T13:52:00.000Z"), + }); + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + + const result = await recovery.reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ continuationRequeued: 0, escalated: 1 }); + expect(enqueueWakeup.mock.calls).not.toContainEqual([ + coderId, + expect.objectContaining({ reason: "issue_continuation_needed" }), + ]); + const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(updatedIssue).toMatchObject({ status: "blocked" }); + const comments = await db + .select({ body: issueComments.body }) + .from(issueComments) + .where(eq(issueComments.issueId, sourceIssueId)); + expect(comments.some(({ body }) => body.includes("non-retryable failure"))).toBe(true); + }); + it("escalates stranded assigned work into a source action instead of a recovery issue", async () => { const { companyId, managerId, coderId, sourceIssue } = await seedCompany(); // A DELIVERED wake returns the queued run. Null models one of `enqueueWakeup`'s diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 9bde30d3f4ba..ed7d969c6678 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1036,10 +1036,12 @@ export function shouldScheduleAutomaticRunRetry( return isIssueRun || isPrReviewRetryContext(contextSnapshot); } - // A failed or missing external-lifecycle Job may have performed non-idempotent - // work. Retry only when the reconciler durably proved adapter invocation never - // began; lock/status gates alone cannot make partial external writes safe. - if (run.errorCode === "job_failed" || run.errorCode === "job_missing") { + // A failed external-lifecycle Job may have performed non-idempotent work. Retry + // only when the reconciler durably proved adapter invocation never began; + // lock/status gates alone cannot make partial external writes safe. A missing + // Job is only produced after adapter.invoke and therefore never reaches this + // safe state; pre-invocation disappearance is process_lost instead. + if (run.errorCode === "job_failed") { const recovery = parseObject(parseObject(run.resultJson).externalLifecycleRecovery); return isIssueRun && recovery.adapterInvocationStarted === false; } diff --git a/server/src/services/recovery/service.infra-class-continuation.test.ts b/server/src/services/recovery/service.infra-class-continuation.test.ts index 7bd050641555..16ff72acfb8f 100644 --- a/server/src/services/recovery/service.infra-class-continuation.test.ts +++ b/server/src/services/recovery/service.infra-class-continuation.test.ts @@ -42,10 +42,10 @@ describe("BLO-18106: job_missing continuation recovery is evidence-gated", () => expect(classifyContinuationFailure(run("budget_blocked")).kind).toBe("non_retryable"); }); - it("keeps job_missing on the fail-safe path because production emits it only after invocation", () => { + it("makes job_missing non-retryable because production emits it only after invocation", () => { expect(classifyContinuationFailure(run("job_missing"))).toMatchObject({ - kind: "default", - maxAttempts: 1, + kind: "non_retryable", + maxAttempts: 0, }); // Pre-invocation disappearance is persisted as process_lost, which remains diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 7a9c817f5e8c..3a37108ab88f 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -821,6 +821,9 @@ const NON_RETRYABLE_CONTINUATION_ERROR_CODES = new Set([ "budget_exhausted", "issue_paused", "issue_dependencies_blocked", + // Production emits job_missing only after adapter.invoke, so continuation + // replay could duplicate a durable external side effect. + "job_missing", ]); // A continuation cancelled with this code is a *deliberate wait* (the latest run From 9a796fcc044c37b74e1676805c13e6b7f4f449da Mon Sep 17 00:00:00 2001 From: Release Engineer Date: Wed, 5 Aug 2026 18:58:42 +0000 Subject: [PATCH 05/15] fix(recovery): close ambiguous side-effect replay paths Co-Authored-By: Paperclip --- .../heartbeat-retry-scheduling.test.ts | 13 ++++ .../__tests__/issue-recovery-actions.test.ts | 60 ++++++++++++++++--- server/src/services/heartbeat.ts | 6 ++ .../service.infra-class-continuation.test.ts | 4 +- server/src/services/recovery/service.ts | 49 +++++++++++++++ 5 files changed, 123 insertions(+), 9 deletions(-) diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 8c9d61ae93e3..0a98ac658f40 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -1926,6 +1926,19 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { ).toBe(false); }); + it.each(["job_missing", "k8s_pod_schedule_failed"])( + "does not let stale transient metadata replay %s", + (errorCode) => { + expect( + shouldScheduleAutomaticRunRetry({ + errorCode, + resultJson: { errorFamily: "transient_upstream" }, + contextSnapshot: { issueId: randomUUID(), wakeReason: "issue_assigned" }, + }), + ).toBe(false); + }, + ); + it("BLO-9147 AC2: CAPACITY_BLOCKED_HEARTBEAT_RETRY_MAX_ATTEMPTS exceeds rate-limit cap (12)", () => { expect(CAPACITY_BLOCKED_HEARTBEAT_RETRY_MAX_ATTEMPTS).toBeGreaterThan(12); }); diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 04e28861f834..a0db4b83e962 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -483,8 +483,45 @@ describeEmbeddedPostgres("issue recovery actions", () => { expect(await svc.getActiveForIssue(randomUUID(), sourceIssueId)).toBeNull(); }); - it("does not enqueue continuation work after an invoked external Job disappears", async () => { + it.each([ + ["job_missing", "in_progress"], + ["job_missing", "todo"], + ["job_missing", "in_review"], + ["k8s_pod_schedule_failed", "in_progress"], + ["k8s_pod_schedule_failed", "todo"], + ["k8s_pod_schedule_failed", "in_review"], + ] as const)("does not enqueue recovery work after %s leaves an issue %s", async (errorCode, status) => { const { companyId, coderId, sourceIssueId } = await seedCompany(); + if (status === "in_review") { + const stageId = randomUUID(); + await db.update(issues).set({ + status, + executionPolicy: { + mode: "normal", + commentRequired: true, + stages: [{ + id: stageId, + type: "review", + approvalsNeeded: 1, + participants: [{ id: randomUUID(), type: "agent", agentId: coderId, userId: null }], + }], + }, + executionState: { + status: "pending", + currentStageId: stageId, + currentStageIndex: 0, + currentStageType: "review", + currentParticipant: { type: "agent", agentId: coderId, userId: null }, + returnAssignee: { type: "agent", agentId: coderId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }, + }).where(eq(issues.id, sourceIssueId)); + } else if (status === "todo") { + await db.update(issues).set({ status }).where(eq(issues.id, sourceIssueId)); + } const runId = randomUUID(); await db.insert(heartbeatRuns).values({ id: runId, @@ -493,7 +530,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { invocationSource: "automation", status: "failed", error: "External lifecycle Job is missing while heartbeat run is still running", - errorCode: "job_missing", + errorCode, resultJson: { externalLifecycleRecovery: { adapterInvocationStarted: true }, }, @@ -506,11 +543,20 @@ describeEmbeddedPostgres("issue recovery actions", () => { const result = await recovery.reconcileStrandedAssignedIssues(); - expect(result).toMatchObject({ continuationRequeued: 0, escalated: 1 }); - expect(enqueueWakeup.mock.calls).not.toContainEqual([ - coderId, - expect.objectContaining({ reason: "issue_continuation_needed" }), - ]); + expect(result).toMatchObject({ + continuationRequeued: 0, + dispatchRequeued: 0, + reviewParticipantRequeued: 0, + escalated: 1, + }); + expect(enqueueWakeup).toHaveBeenCalledTimes(1); + expect(enqueueWakeup.mock.calls[0]?.[1]).toMatchObject({ + reason: "source_scoped_recovery_action", + contextSnapshot: { + allowDeliverableWork: false, + recoveryIntent: "status_only", + }, + }); const [updatedIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); expect(updatedIssue).toMatchObject({ status: "blocked" }); const comments = await db diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index ed7d969c6678..1d2ce9ff90e5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -975,6 +975,12 @@ export async function probeStaleKillReviewEvidence( export function shouldScheduleAutomaticRunRetry( run: Pick, ) { + // These outcomes can follow non-idempotent external work. Reject them before + // reading merged result metadata, which may contain a stale transient family. + if (run.errorCode === "job_missing" || run.errorCode === "k8s_pod_schedule_failed") { + return false; + } + // BLO-18030: a hard-stale-kill force-terminates a Job that was claimed but // silent past EXTERNAL_LIFECYCLE_HARD_STALE_MS. That left pr_review wakes with // no recovery whatsoever: the run is terminal with no bounded retry, and the diff --git a/server/src/services/recovery/service.infra-class-continuation.test.ts b/server/src/services/recovery/service.infra-class-continuation.test.ts index 16ff72acfb8f..5df7b0c3376a 100644 --- a/server/src/services/recovery/service.infra-class-continuation.test.ts +++ b/server/src/services/recovery/service.infra-class-continuation.test.ts @@ -32,8 +32,8 @@ describe("BLO-18106: job_missing continuation recovery is evidence-gated", () => // k8s_pod_schedule_failed can be emitted after the main container starts, so // it must not be replayed without stronger producer evidence. expect(classifyContinuationFailure(run("k8s_pod_schedule_failed"))).toMatchObject({ - kind: "default", - maxAttempts: workClass.maxAttempts, + kind: "non_retryable", + maxAttempts: 0, }); }); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 3a37108ab88f..aefc3044a129 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -824,6 +824,9 @@ const NON_RETRYABLE_CONTINUATION_ERROR_CODES = new Set([ // Production emits job_missing only after adapter.invoke, so continuation // replay could duplicate a durable external side effect. "job_missing", + // Adapters also use this after main-container startup failures, so scheduling + // failure alone does not prove that external work never began. + "k8s_pod_schedule_failed", ]); // A continuation cancelled with this code is a *deliberate wait* (the latest run @@ -6045,6 +6048,31 @@ export function recoveryService( continue; } + const participantContinuationClassification = classifyContinuationFailure(participantLatestRun); + if ( + isUnsuccessfulTerminalIssueRun(participantLatestRun) && + participantContinuationClassification.kind === "non_retryable" + ) { + const failureSummary = summarizeRunFailureForIssueComment(participantLatestRun); + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "in_review", + latestRun: participantLatestRun, + recoveryOwnerAgentId: participantAgentId, + comment: + "Paperclip detected a non-retryable failure on the active review participant's run " + + `(\`${participantContinuationClassification.errorCode}\`). Skipping automatic retries and moving it to ` + + `\`blocked\` so it is visible for intervention.${failureSummary ?? ""}`, + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + const participantAdapterFailureClassification = isUnsuccessfulTerminalIssueRun(participantLatestRun) ? classifyAdapterFailureForRecovery(participantLatestRun, recoveryNow) : null; @@ -6189,6 +6217,27 @@ export function recoveryService( continue; } + const assignmentContinuationClassification = classifyContinuationFailure(latestRun); + if (assignmentContinuationClassification.kind === "non_retryable") { + const failureSummary = summarizeRunFailureForIssueComment(latestRun); + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "todo", + latestRun, + comment: + "Paperclip detected a non-retryable failure on this assigned issue's latest run " + + `(\`${assignmentContinuationClassification.errorCode}\`). Skipping automatic retries and moving it to ` + + `\`blocked\` so it is visible for intervention.${failureSummary ?? ""}`, + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + if (isNonRetryableTerminalRun(latestRun)) { if (await latestRunPredatesLatestUnblock(issue.companyId, issue.id, latestRun)) { // BLO-8050: operator just unblocked; skip re-escalation on stale evidence. From f372f6b6f07467a7b44ea1d2a6de160b07308077 Mon Sep 17 00:00:00 2001 From: Release Engineer Date: Wed, 5 Aug 2026 20:57:11 +0000 Subject: [PATCH 06/15] fix(recovery): close remaining side-effect replay paths Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 159 ++++++++++++++++++ server/src/services/recovery/service.ts | 59 ++++++- 2 files changed, 211 insertions(+), 7 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 589565fadf0b..10254bfad224 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -6504,6 +6504,67 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("records non-retryable review-participant recovery under the reviewer cause", async () => { + const { companyId, agentId, issueId, runId, wakeupRequestId, stageId } = + await seedInReviewParticipantRunFixture(); + const sourceAssigneeAgentId = randomUUID(); + const finishedAt = new Date("2026-03-19T00:05:00.000Z"); + + await db.insert(agents).values({ + id: sourceAssigneeAgentId, + companyId, + name: "CodexImplementor", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.update(issues).set({ + assigneeAgentId: sourceAssigneeAgentId, + executionRunId: null, + executionState: { + status: "pending", + currentStageId: stageId, + currentStageIndex: 0, + currentStageType: "review", + currentParticipant: { type: "agent", agentId, userId: null }, + returnAssignee: { type: "agent", agentId: sourceAssigneeAgentId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }, + }).where(eq(issues.id, issueId)); + await db.update(heartbeatRuns).set({ + status: "failed", + startedAt: new Date("2026-03-19T00:00:00.000Z"), + finishedAt, + updatedAt: finishedAt, + errorCode: "job_missing", + error: "External lifecycle Job disappeared after adapter invocation", + }).where(eq(heartbeatRuns.id, runId)); + await db.update(agentWakeupRequests).set({ + status: "failed", + claimedAt: new Date("2026-03-19T00:00:00.000Z"), + finishedAt, + updatedAt: finishedAt, + error: "External lifecycle Job disappeared after adapter invocation", + }).where(eq(agentWakeupRequests.id, wakeupRequestId)); + + const result = await createHeartbeat().reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ reviewParticipantRequeued: 0, escalated: 1 }); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(action).toMatchObject({ + cause: "execution_review_participant_recovery", + ownerAgentId: agentId, + previousOwnerAgentId: sourceAssigneeAgentId, + returnOwnerAgentId: sourceAssigneeAgentId, + }); + }); + it.each([ ["failed", "adapter_failed"], ["failed", "process_lost"], @@ -6674,6 +6735,55 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("blocks accepted interaction continuation after job_missing without replaying it", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + runErrorCode: "job_missing", + retryReason: "issue_continuation_needed", + }); + const interactionId = randomUUID(); + const resolvedAt = new Date("2026-03-18T23:59:00.000Z"); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt, + updatedAt: resolvedAt, + payload: { version: 1, prompt: "Approve the plan?" }, + result: { outcome: "accepted" }, + }); + await db.update(heartbeatRuns).set({ + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_continuation_needed", + retryReason: "issue_continuation_needed", + mutation: "interaction", + interactionId, + }, + }).where(eq(heartbeatRuns.id, runId)); + + const result = await createHeartbeat().reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ continuationRequeued: 0, escalated: 1 }); + const [issue, runs] = await Promise.all([ + db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null), + db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)), + ]); + expect(issue?.status).toBe("blocked"); + expect(runs.some((run) => run.id === runId)).toBe(true); + expect(runs.filter((run) => + (run.contextSnapshot as Record | null)?.source === + "issue.interaction_continuation_recovery" + )).toHaveLength(0); + }); + it("escalates accepted interaction continuation recovery after three review-park cancellations", async () => { const companyId = randomUUID(); const agentId = randomUUID(); @@ -8739,6 +8849,14 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { retryReason: null, runUsageJson: null, }, + { + label: "todo + non-retryable continuation failure", + issueStatus: "todo" as const, + runStatus: "failed" as const, + runErrorCode: "job_missing", + retryReason: null, + runUsageJson: null, + }, { label: "todo + zero-token startup failure run", issueStatus: "todo" as const, @@ -8827,6 +8945,47 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }, ); + it("skips non-retryable review-participant escalation when the failure predates an operator unblock", async () => { + const { companyId, agentId, issueId, runId, wakeupRequestId } = + await seedInReviewParticipantRunFixture(); + const failedAt = new Date("2026-03-19T00:05:00.000Z"); + await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, issueId)); + await db.update(heartbeatRuns).set({ + status: "failed", + createdAt: failedAt, + startedAt: failedAt, + finishedAt: failedAt, + updatedAt: failedAt, + errorCode: "job_missing", + error: "External lifecycle Job disappeared after adapter invocation", + }).where(eq(heartbeatRuns.id, runId)); + await db.update(agentWakeupRequests).set({ + status: "failed", + claimedAt: failedAt, + finishedAt: failedAt, + updatedAt: failedAt, + error: "External lifecycle Job disappeared after adapter invocation", + }).where(eq(agentWakeupRequests.id, wakeupRequestId)); + await db.insert(activityLog).values({ + id: randomUUID(), + companyId, + actorType: "user", + actorId: "operator", + action: "issue.updated", + entityType: "issue", + entityId: issueId, + details: { previousStatus: "blocked", status: "in_review" }, + createdAt: new Date("2026-03-19T01:00:00.000Z"), + }); + + const result = await createHeartbeat().reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ reviewParticipantRequeued: 0, escalated: 0 }); + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue).toMatchObject({ status: "in_review", assigneeAgentId: agentId }); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + }); + it("does not treat a productive terminal run as healthy when in-progress work has no live path", async () => { const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "in_progress", diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index aefc3044a129..c23321ffeef0 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1764,7 +1764,13 @@ export function recoveryService( .then((rows) => Boolean(rows[0])); } - async function getLatestIssueRunSince(companyId: string, issueId: string, agentId: string, since: Date): Promise { + async function getLatestIssueRunSince( + companyId: string, + issueId: string, + agentId: string, + since: Date, + interactionId: string, + ): Promise { return db .select({ id: heartbeatRuns.id, @@ -1785,6 +1791,7 @@ export function recoveryService( eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + sql`${heartbeatRuns.contextSnapshot} ->> 'interactionId' = ${interactionId}`, or(gte(heartbeatRuns.createdAt, since), gte(heartbeatRuns.finishedAt, since)), ), ) @@ -5939,6 +5946,41 @@ export function recoveryService( ); if (!successfulRunSinceResolution) { + const latestPostResolutionRun = await getLatestIssueRunSince( + issue.companyId, + issue.id, + agentId, + acceptedInteractionResolvedAt, + acceptedContinuationInteraction.id, + ); + const postResolutionClassification = latestPostResolutionRun && + isUnsuccessfulTerminalIssueRun(latestPostResolutionRun) + ? classifyContinuationFailure(latestPostResolutionRun) + : null; + if (postResolutionClassification?.kind === "non_retryable") { + if (await latestRunPredatesLatestUnblock(issue.companyId, issue.id, latestPostResolutionRun)) { + result.skipped += 1; + continue; + } + const failureSummary = summarizeRunFailureForIssueComment(latestPostResolutionRun); + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: issue.status as StrandedPreviousStatus, + latestRun: latestPostResolutionRun, + comment: + "Paperclip detected a non-retryable failure after an accepted interaction " + + `(\`${postResolutionClassification.errorCode}\`). Skipping continuation replay and moving it to ` + + `\`blocked\` so it is visible for intervention.${failureSummary ?? ""}`, + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + if (!agentInvokable) { result.skipped += 1; continue; @@ -5954,12 +5996,6 @@ export function recoveryService( continue; } - const latestPostResolutionRun = await getLatestIssueRunSince( - issue.companyId, - issue.id, - agentId, - acceptedInteractionResolvedAt, - ); const { consecutive } = await summarizeRecentContinuationRetries( issue.companyId, issue.id, @@ -6053,11 +6089,16 @@ export function recoveryService( isUnsuccessfulTerminalIssueRun(participantLatestRun) && participantContinuationClassification.kind === "non_retryable" ) { + if (await latestRunPredatesLatestUnblock(issue.companyId, issue.id, participantLatestRun)) { + result.skipped += 1; + continue; + } const failureSummary = summarizeRunFailureForIssueComment(participantLatestRun); const updated = await escalateStrandedAssignedIssue({ issue, previousStatus: "in_review", latestRun: participantLatestRun, + recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON, recoveryOwnerAgentId: participantAgentId, comment: "Paperclip detected a non-retryable failure on the active review participant's run " + @@ -6219,6 +6260,10 @@ export function recoveryService( const assignmentContinuationClassification = classifyContinuationFailure(latestRun); if (assignmentContinuationClassification.kind === "non_retryable") { + if (await latestRunPredatesLatestUnblock(issue.companyId, issue.id, latestRun)) { + result.skipped += 1; + continue; + } const failureSummary = summarizeRunFailureForIssueComment(latestRun); const updated = await escalateStrandedAssignedIssue({ issue, From 13ffa96d5ab11a29552bd2d13dbba6ce4ed060f8 Mon Sep 17 00:00:00 2001 From: Paperclip Release Engineer Date: Wed, 5 Aug 2026 23:36:55 +0000 Subject: [PATCH 07/15] fix(recovery): preserve side-effect replay fences Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 160 ++++++++++++++++-- server/src/services/heartbeat.ts | 16 +- server/src/services/recovery/service.ts | 23 ++- 3 files changed, 178 insertions(+), 21 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 10254bfad224..a450ccfeab91 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1008,6 +1008,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { reason: wakeReason, payload: { issueId, + executionStage: { stageId, stageType: "review" }, ...(input?.retryReason ? { retryReason: input.retryReason } : {}), }, status: "queued", @@ -1028,6 +1029,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { issueId, taskId: issueId, wakeReason, + executionStage: { stageId, stageType: "review" }, ...(input?.retryReason ? { retryReason: input.retryReason } : {}), }, updatedAt: now, @@ -1911,13 +1913,21 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); expect(await heartbeat.getRun(runId)).toMatchObject({ status: "failed", - errorCode: "pr_review_output_missing", + errorCode: "job_missing", + resultJson: { + externalLifecycleRecovery: expect.objectContaining({ adapterInvocationStarted: true }), + }, }); const retries = await db .select() .from(heartbeatRuns) .where(eq(heartbeatRuns.retryOfRunId, runId)); - expect(retries).toHaveLength(1); + expect(retries.some((retry) => + (retry.contextSnapshot as Record | null)?.source === "issue.continuation_recovery" + )).toBe(false); + expect(retries.every((retry) => + (retry.contextSnapshot as Record | null)?.allowDeliverableWork === false + )).toBe(true); }); async function recoverClaimedReviewWithUnavailableVerification(kind: "result" | "throw") { @@ -1960,20 +1970,32 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { it("keeps a missing-Job review claim fail-closed when GitHub returns an error", async () => { await expect(recoverClaimedReviewWithUnavailableVerification("result")).resolves.toMatchObject({ status: "failed", - errorCode: "pr_review_verification_unavailable", - error: expect.stringContaining("reviews_http_503"), + errorCode: "job_missing", + resultJson: { + externalLifecycleRecovery: expect.objectContaining({ + adapterInvocationStarted: true, + prReviewErrorCode: "pr_review_verification_unavailable", + prReviewErrorMessage: expect.stringContaining("reviews_http_503"), + }), + }, }); }); it("keeps a missing-Job review claim fail-closed when GitHub verification throws", async () => { await expect(recoverClaimedReviewWithUnavailableVerification("throw")).resolves.toMatchObject({ status: "failed", - errorCode: "pr_review_verification_unavailable", - error: expect.stringContaining("verification_threw"), + errorCode: "job_missing", + resultJson: { + externalLifecycleRecovery: expect.objectContaining({ + adapterInvocationStarted: true, + prReviewErrorCode: "pr_review_verification_unavailable", + prReviewErrorMessage: expect.stringContaining("verification_threw"), + }), + }, }); }); - it("fails and retries once when a PR-review request comment is not outcome evidence", async () => { + it("does not replay a missing-Job PR review when a request comment is not outcome evidence", async () => { const jobName = "agent-opencode-review-lost"; const headSha = "075a9aeff53a229199ab0583e916f33c22459983"; const { companyId, agentId, runId } = await seedRunFixture({ @@ -2013,14 +2035,24 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(mockGithubHasReviewerEvidenceForPr).toHaveBeenCalledTimes(1); expect(await heartbeat.getRun(runId)).toMatchObject({ status: "failed", - errorCode: "pr_review_output_missing", + errorCode: "job_missing", + resultJson: { + externalLifecycleRecovery: expect.objectContaining({ + adapterInvocationStarted: true, + prReviewErrorCode: "pr_review_output_missing", + }), + }, }); const retries = await db .select() .from(heartbeatRuns) .where(eq(heartbeatRuns.retryOfRunId, runId)); - expect(retries).toHaveLength(1); - expect(retries[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 1 }); + expect(retries.some((retry) => + (retry.contextSnapshot as Record | null)?.source === "issue.continuation_recovery" + )).toBe(false); + expect(retries.every((retry) => + (retry.contextSnapshot as Record | null)?.allowDeliverableWork === false + )).toBe(true); }); it("does not treat generic run artifacts as a completed missing-Job outcome", async () => { @@ -6102,6 +6134,55 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(retryRun?.contextSnapshot as Record).not.toHaveProperty("modelProfile"); }); + it("does not apply a prior-stage job_missing failure to a later stage with the same reviewer", async () => { + const { agentId, issueId, runId, wakeupRequestId } = await seedInReviewParticipantRunFixture(); + const nextStageId = randomUUID(); + const finishedAt = new Date("2026-03-19T00:05:00.000Z"); + await db.update(heartbeatRuns).set({ + status: "failed", + error: "External lifecycle Job is missing while heartbeat run is still running", + errorCode: "job_missing", + startedAt: new Date("2026-03-19T00:00:00.000Z"), + finishedAt, + updatedAt: finishedAt, + }).where(eq(heartbeatRuns.id, runId)); + await db.update(agentWakeupRequests).set({ + status: "failed", + finishedAt, + updatedAt: finishedAt, + }).where(eq(agentWakeupRequests.id, wakeupRequestId)); + await db.update(issues).set({ + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + executionState: { + status: "pending", + currentStageId: nextStageId, + currentStageIndex: 1, + currentStageType: "review", + currentParticipant: { type: "agent", agentId, userId: null }, + returnAssignee: { type: "agent", agentId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }, + }).where(eq(issues.id, issueId)); + + const result = await createHeartbeat().reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ reviewParticipantRequeued: 0, escalated: 0, skipped: 1 }); + const [issue, recoveryRuns] = await Promise.all([ + db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null), + db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.agentId, agentId), + sql`${heartbeatRuns.contextSnapshot} ->> 'currentStageId' = ${nextStageId}`, + )), + ]); + expect(issue?.status).toBe("in_review"); + expect(recoveryRuns).toHaveLength(0); + }); + it("re-enqueues a stranded execution-review participant when another agent has the latest issue run", async () => { const { companyId, agentId, issueId, runId, wakeupRequestId, stageId } = await seedInReviewParticipantRunFixture(); @@ -6784,6 +6865,65 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { )).toHaveLength(0); }); + it("blocks accepted interaction continuation after reassignment without replaying prior-owner work", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + runErrorCode: "job_missing", + retryReason: "issue_continuation_needed", + }); + const nextAgentId = randomUUID(); + const interactionId = randomUUID(); + const resolvedAt = new Date("2026-03-18T23:59:00.000Z"); + await db.insert(agents).values({ + id: nextAgentId, + companyId, + name: "NextOwner", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt, + updatedAt: resolvedAt, + payload: { version: 1, prompt: "Approve the plan?" }, + result: { outcome: "accepted" }, + }); + await db.update(heartbeatRuns).set({ + contextSnapshot: { issueId, interactionId }, + }).where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ assigneeAgentId: nextAgentId }).where(eq(issues.id, issueId)); + + const result = await createHeartbeat().reconcileStrandedAssignedIssues(); + + expect(result).toMatchObject({ continuationRequeued: 0, escalated: 1 }); + const [issue, nextOwnerRuns] = await Promise.all([ + db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null), + db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, nextAgentId)), + ]); + expect(issue?.status).toBe("blocked"); + expect(nextOwnerRuns.some((run) => + (run.contextSnapshot as Record | null)?.source === + "issue.interaction_continuation_recovery" + )).toBe(false); + expect(nextOwnerRuns).toHaveLength(1); + expect(nextOwnerRuns[0]?.contextSnapshot).toMatchObject({ + allowDeliverableWork: false, + recoveryIntent: "status_only", + }); + }); + it("escalates accepted interaction continuation recovery after three review-park cancellations", async () => { const companyId = randomUUID(); const agentId = randomUUID(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1d2ce9ff90e5..e6b2de5cb277 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1030,6 +1030,8 @@ export function shouldScheduleAutomaticRunRetry( run.errorCode === "pr_review_output_missing" || run.errorCode === "pr_review_verification_unavailable" ) { + const recovery = parseObject(parseObject(run.resultJson).externalLifecycleRecovery); + if (recovery.adapterInvocationStarted === true) return false; return isPrReviewRetryContext(parseObject(run.contextSnapshot)); } @@ -16586,7 +16588,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } : externalLifecycleTerminalOutcome(input.jobStatus, preserveRecordedOutcome); const terminalOutcome = - baseTerminalOutcome && prReviewIncompleteOverride && !input.staleKill + baseTerminalOutcome && prReviewIncompleteOverride && !input.staleKill && + baseTerminalOutcome.errorCode !== "job_missing" ? { ...baseTerminalOutcome, errorCode: prReviewIncompleteOverride.errorCode, @@ -16596,7 +16599,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (!terminalOutcome) return false; const adapterInvocationStarted = - terminalOutcome.errorCode === "job_failed" || terminalOutcome.errorCode === "job_missing" + baseTerminalOutcome?.errorCode === "job_failed" || baseTerminalOutcome?.errorCode === "job_missing" ? await hasAdapterInvocationEvent(input.run.id) : null; @@ -16617,6 +16620,12 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) jobPhase: terminalOutcome.jobPhase, jobReason: terminalOutcome.jobReason, jobMessage: terminalOutcome.jobMessage, + ...(prReviewIncompleteOverride + ? { + prReviewErrorCode: prReviewIncompleteOverride.errorCode, + prReviewErrorMessage: prReviewIncompleteOverride.errorMessage, + } + : {}), ...(adapterInvocationStarted !== null ? { adapterInvocationStarted } : {}), ...(containerDiagnostics ? { containerDiagnostics } : {}), }, @@ -16731,6 +16740,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const finalizationAgent = await getAgent(finalizedRun.agentId); if ( terminalOutcome.status === "failed" && + adapterInvocationStarted !== true && shouldScheduleAutomaticRunRetry(finalizedRun) && finalizationAgent ) { @@ -24537,6 +24547,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const shouldBlockImmediately = !recoveryAgentInvokable || !recoveryAgent || + run.errorCode === "job_missing" || + run.errorCode === "k8s_pod_schedule_failed" || isWorkspaceValidationFailedRun(run) || isConfigurationIncompleteFailedRun(run) || didAutomaticRecoveryFail(run, issue.status === "todo" ? "assignment_recovery" : "issue_continuation_needed"); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index c23321ffeef0..ecac4c81044f 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1538,10 +1538,11 @@ export function recoveryService( return count; } - async function getLatestIssueRunForAgent( + async function getLatestIssueRunForAgentStage( companyId: string, issueId: string, agentId: string, + stageId: string, ): Promise { return db .select({ @@ -1563,6 +1564,10 @@ export function recoveryService( eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + sql`coalesce( + ${heartbeatRuns.contextSnapshot} -> 'executionStage' ->> 'stageId', + ${heartbeatRuns.contextSnapshot} ->> 'currentStageId' + ) = ${stageId}`, ), ) .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) @@ -1741,7 +1746,6 @@ export function recoveryService( async function hasSuccessfulIssueRunSince( companyId: string, issueId: string, - agentId: string, since: Date, interactionId?: string | null, ) { @@ -1751,7 +1755,6 @@ export function recoveryService( .where( and( eq(heartbeatRuns.companyId, companyId), - eq(heartbeatRuns.agentId, agentId), eq(heartbeatRuns.status, "succeeded"), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, interactionId @@ -1767,7 +1770,6 @@ export function recoveryService( async function getLatestIssueRunSince( companyId: string, issueId: string, - agentId: string, since: Date, interactionId: string, ): Promise { @@ -1789,7 +1791,6 @@ export function recoveryService( .where( and( eq(heartbeatRuns.companyId, companyId), - eq(heartbeatRuns.agentId, agentId), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, sql`${heartbeatRuns.contextSnapshot} ->> 'interactionId' = ${interactionId}`, or(gte(heartbeatRuns.createdAt, since), gte(heartbeatRuns.finishedAt, since)), @@ -5854,8 +5855,14 @@ export function recoveryService( continue; } const recoveryNow = new Date(); - const participantLatestRunForRecovery = issue.status === "in_review" && participantAgentId - ? await getLatestIssueRunForAgent(issue.companyId, issue.id, participantAgentId) + const participantLatestRunForRecovery = issue.status === "in_review" && participantAgentId && + pendingExecutionState?.currentStageId + ? await getLatestIssueRunForAgentStage( + issue.companyId, + issue.id, + participantAgentId, + pendingExecutionState.currentStageId, + ) : null; const providerQuotaMonitorRun = issue.status === "in_review" ? participantLatestRunForRecovery @@ -5940,7 +5947,6 @@ export function recoveryService( const successfulRunSinceResolution = await hasSuccessfulIssueRunSince( issue.companyId, issue.id, - agentId, acceptedInteractionResolvedAt, acceptedContinuationInteraction.id, ); @@ -5949,7 +5955,6 @@ export function recoveryService( const latestPostResolutionRun = await getLatestIssueRunSince( issue.companyId, issue.id, - agentId, acceptedInteractionResolvedAt, acceptedContinuationInteraction.id, ); From c09d5e0e1b14f3779980fc0ff4ab45c9624a9b46 Mon Sep 17 00:00:00 2001 From: Release Engineer Date: Thu, 6 Aug 2026 02:36:09 +0000 Subject: [PATCH 08/15] fix(recovery): fence immediate review replay Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 72 +++++++++++++++++++ .../__tests__/issue-recovery-actions.test.ts | 16 +++-- server/src/services/heartbeat.ts | 32 +++++++++ 3 files changed, 114 insertions(+), 6 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index a450ccfeab91..e2a48f89d27a 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -6406,6 +6406,78 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(issue?.assigneeAgentId).toBe(agentId); }); + it.each(["job_missing", "k8s_pod_schedule_failed"])( + "blocks immediate review-participant recovery after %s without replaying deliverable work", + async (errorCode) => { + mockAdapterExecute.mockResolvedValueOnce({ + exitCode: 1, + signal: null, + timedOut: false, + errorCode, + errorMessage: "External lifecycle execution ended ambiguously", + provider: "test", + model: "test-model", + }); + const { agentId, issueId, runId, stageId } = await seedInReviewParticipantRunFixture(); + const heartbeat = createHeartbeat(); + + await heartbeat.resumeQueuedRuns(); + const settledRun = await waitForRunToSettle(heartbeat, runId, 8_000); + expect(settledRun).toMatchObject({ status: "failed", errorCode }); + expect(settledRun?.contextSnapshot).toMatchObject({ + executionStage: { stageId, stageType: "review" }, + }); + + const [issue, runs] = await Promise.all([ + db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null), + db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)), + ]); + expect(runs.some((row) => + (row.contextSnapshot as Record | null)?.retryReason === + "execution_review_participant_recovery" && + (row.contextSnapshot as Record | null)?.allowDeliverableWork !== false + )).toBe(false); + expect(issue?.status).toBe("blocked"); + }, + ); + + it("does not let a late prior-stage failure recover a later stage with the same reviewer", async () => { + const { agentId, issueId, runId } = await seedInReviewParticipantRunFixture(); + const nextStageId = randomUUID(); + mockAdapterExecute.mockImplementationOnce(async () => { + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + const executionState = issue?.executionState as Record; + await db.update(issues).set({ + executionState: { ...executionState, currentStageId: nextStageId, currentStageIndex: 1 }, + }).where(eq(issues.id, issueId)); + return { + exitCode: 1, + signal: null, + timedOut: false, + errorCode: "adapter_failed", + errorMessage: "The prior review stage failed after the next stage started", + provider: "test", + model: "test-model", + }; + }); + const heartbeat = createHeartbeat(); + + await heartbeat.resumeQueuedRuns(); + const settledRun = await waitForRunToSettle(heartbeat, runId, 8_000); + expect(settledRun).toMatchObject({ status: "failed", errorCode: "adapter_failed" }); + + const [issue, runs] = await Promise.all([ + db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null), + db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)), + ]); + expect(issue).toMatchObject({ status: "in_review", executionRunId: null }); + expect((issue?.executionState as Record)?.currentStageId).toBe(nextStageId); + expect(runs.filter((row) => + (row.contextSnapshot as Record | null)?.retryReason === + "execution_review_participant_recovery" + )).toHaveLength(0); + }); + it("retries a pending execution-review participant once before blocking with a recovery action", async () => { const { companyId, agentId, issueId, runId, stageId } = await seedInReviewParticipantRunFixture(); const heartbeat = createHeartbeat(); diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index a0db4b83e962..a2695f0ae321 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -492,8 +492,9 @@ describeEmbeddedPostgres("issue recovery actions", () => { ["k8s_pod_schedule_failed", "in_review"], ] as const)("does not enqueue recovery work after %s leaves an issue %s", async (errorCode, status) => { const { companyId, coderId, sourceIssueId } = await seedCompany(); + let stageId: string | null = null; if (status === "in_review") { - const stageId = randomUUID(); + stageId = randomUUID(); await db.update(issues).set({ status, executionPolicy: { @@ -534,7 +535,10 @@ describeEmbeddedPostgres("issue recovery actions", () => { resultJson: { externalLifecycleRecovery: { adapterInvocationStarted: true }, }, - contextSnapshot: { issueId: sourceIssueId }, + contextSnapshot: { + issueId: sourceIssueId, + ...(stageId ? { executionStage: { stageId, stageType: "review" } } : {}), + }, startedAt: new Date("2026-07-26T13:45:00.000Z"), finishedAt: new Date("2026-07-26T13:52:00.000Z"), }); @@ -979,7 +983,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { errorCode: "adapter_failed", startedAt: new Date("2026-07-15T20:00:00.000Z"), finishedAt: new Date("2026-07-15T20:01:00.000Z"), - contextSnapshot: { issueId: sourceIssueId }, + contextSnapshot: { issueId: sourceIssueId, executionStage: { stageId, stageType: "review" } }, }); const enqueueWakeup = vi.fn(async () => null); const recovery = recoveryService(db, { enqueueWakeup }); @@ -1040,7 +1044,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { errorCode: "adapter_failed", startedAt: new Date("2026-07-15T20:00:00.000Z"), finishedAt: new Date("2026-07-15T20:01:00.000Z"), - contextSnapshot: { issueId: sourceIssueId }, + contextSnapshot: { issueId: sourceIssueId, executionStage: { stageId, stageType: "review" } }, }); const enqueueWakeup = vi.fn(async () => null); const recovery = recoveryService(db, { enqueueWakeup }); @@ -1127,7 +1131,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { errorCode: "adapter_failed", startedAt: new Date("2026-07-15T20:00:00.000Z"), finishedAt: new Date("2026-07-15T20:01:00.000Z"), - contextSnapshot: { issueId: sourceIssueId }, + contextSnapshot: { issueId: sourceIssueId, executionStage: { stageId, stageType: "review" } }, }, { id: assigneeRunId, companyId, @@ -1199,7 +1203,7 @@ describeEmbeddedPostgres("issue recovery actions", () => { errorCode: "adapter_failed", startedAt: new Date("2026-07-15T20:00:00.000Z"), finishedAt: new Date("2026-07-15T20:01:00.000Z"), - contextSnapshot: { issueId: sourceIssueId }, + contextSnapshot: { issueId: sourceIssueId, executionStage: { stageId, stageType: "review" } }, }); const enqueueWakeup = vi.fn(async () => ({ id: randomUUID() } as never)); const recovery = recoveryService(db, { enqueueWakeup }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e6b2de5cb277..f96f5d418964 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -24089,6 +24089,32 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) : candidateIssues[0]) ?? null; if (!issue) return null; + const activeExecutionState = parseIssueExecutionState(issue.executionState); + const finalizedRunExecutionStage = parseObject(runContext.executionStage); + const finalizedRunStageId = + readNonEmptyString(finalizedRunExecutionStage.stageId) ?? readNonEmptyString(runContext.currentStageId); + const activeParticipant = activeExecutionState?.status === "pending" + ? activeExecutionState.currentParticipant + : null; + if ( + (run.errorCode === "job_missing" || run.errorCode === "k8s_pod_schedule_failed") && + issue.status === "in_review" && + !issue.assigneeUserId && + Boolean(activeExecutionState?.currentStageId) && + finalizedRunStageId === activeExecutionState?.currentStageId && + activeParticipant?.type === "agent" && + activeParticipant.agentId === run.agentId && + isExecutionReviewParticipantRecoveryEligibleRun(run) + ) { + return { + kind: "blocked" as const, + issue, + previousStatus: issue.status, + comment: buildExecutionReviewParticipantRecoveryComment({ latestRun: run }), + recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE, + recoveryOwnerAgentId: activeParticipant.agentId, + }; + } if (issue.executionRunId && issue.executionRunId !== run.id) return null; // Pre-dispatch validation recovery: if the finalizing run failed before @@ -24404,9 +24430,15 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const currentParticipant = executionState?.status === "pending" ? executionState.currentParticipant : null; + const reviewRunContext = parseObject(run.contextSnapshot); + const runExecutionStage = parseObject(reviewRunContext.executionStage); + const runStageId = + readNonEmptyString(runExecutionStage.stageId) ?? readNonEmptyString(reviewRunContext.currentStageId); const issueNeedsReviewParticipantRecovery = issue.status === "in_review" && !issue.assigneeUserId && + Boolean(executionState?.currentStageId) && + runStageId === executionState?.currentStageId && currentParticipant?.type === "agent" && currentParticipant.agentId === run.agentId && isExecutionReviewParticipantRecoveryEligibleRun(run) && From 8e7512a04a4e7b70c9b80d5d1c0047a3ba29a492 Mon Sep 17 00:00:00 2001 From: Release Engineer Date: Thu, 6 Aug 2026 05:24:20 +0000 Subject: [PATCH 09/15] fix(recovery): revalidate review stage before escalation Co-Authored-By: Paperclip --- .../__tests__/issue-recovery-actions.test.ts | 63 +++++++++++++++++++ server/src/services/heartbeat.ts | 27 +++++++- server/src/services/recovery/service.ts | 21 ++++++- 3 files changed, 107 insertions(+), 4 deletions(-) diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index a2695f0ae321..5baa8e498de9 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -570,6 +570,69 @@ describeEmbeddedPostgres("issue recovery actions", () => { expect(comments.some(({ body }) => body.includes("non-retryable failure"))).toBe(true); }); + it("does not escalate a stale review failure after the active stage advances", async () => { + const { companyId, managerId, coderId, sourceIssueId } = await seedCompany(); + const staleStageId = randomUUID(); + const activeStageId = randomUUID(); + const executionState = (stageId: string, participantAgentId: string) => ({ + status: "pending" as const, + currentStageId: stageId, + currentStageIndex: 0, + currentStageType: "review" as const, + currentParticipant: { type: "agent" as const, agentId: participantAgentId, userId: null }, + returnAssignee: { type: "agent" as const, agentId: coderId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }); + await db.update(issues).set({ + status: "in_review", + executionState: executionState(staleStageId, managerId), + }).where(eq(issues.id, sourceIssueId)); + const staleIssue = await db.select().from(issues).where(eq(issues.id, sourceIssueId)).then((rows) => rows[0]!); + const staleRun = { + id: randomUUID(), + companyId, + agentId: managerId, + status: "failed", + errorCode: "job_missing", + error: "External lifecycle Job disappeared after adapter invocation", + contextSnapshot: { issueId: sourceIssueId, executionStage: { stageId: staleStageId } }, + resultJson: null, + usageJson: null, + livenessState: null, + createdAt: new Date(), + } as const; + + await db.update(issues).set({ + executionState: executionState(activeStageId, coderId), + }).where(eq(issues.id, sourceIssueId)); + const enqueueWakeup = vi.fn(async () => null); + const recovery = recoveryService(db, { enqueueWakeup }); + + const updated = await recovery.escalateStrandedAssignedIssue({ + issue: staleIssue, + previousStatus: "in_review", + latestRun: staleRun, + recoveryCause: "execution_review_participant_recovery", + recoveryOwnerAgentId: managerId, + expectedReviewStage: { stageId: staleStageId, participantAgentId: managerId }, + }); + + expect(updated).toBeNull(); + expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); + expect(enqueueWakeup).not.toHaveBeenCalled(); + const [freshIssue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(freshIssue).toMatchObject({ + status: "in_review", + executionState: { + currentStageId: activeStageId, + currentParticipant: { type: "agent", agentId: coderId }, + }, + }); + }); + it("escalates stranded assigned work into a source action instead of a recovery issue", async () => { const { companyId, managerId, coderId, sourceIssue } = await seedCompany(); // A DELIVERED wake returns the queued run. Null models one of `enqueueWakeup`'s diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index f96f5d418964..015aa13caa17 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -23975,6 +23975,17 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ); } + function buildNonRetryableExecutionReviewParticipantComment(input: { + latestRun: Pick | null | undefined; + }) { + const failureSummary = summarizeRunFailureForIssueComment(input.latestRun); + return ( + "Paperclip skipped automatic recovery for the pending execution-review participant because the run may have " + + `performed non-idempotent work before its external lifecycle failed.${failureSummary ?? ""} ` + + "Moving it to `blocked` with a source-scoped recovery action instead of risking a duplicate review or artifact." + ); + } + async function releaseIssueExecutionAndPromote( run: typeof heartbeatRuns.$inferSelect, options: { suppressImmediateRecovery?: boolean } = {}, @@ -24100,7 +24111,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) (run.errorCode === "job_missing" || run.errorCode === "k8s_pod_schedule_failed") && issue.status === "in_review" && !issue.assigneeUserId && - Boolean(activeExecutionState?.currentStageId) && + finalizedRunStageId !== null && finalizedRunStageId === activeExecutionState?.currentStageId && activeParticipant?.type === "agent" && activeParticipant.agentId === run.agentId && @@ -24110,9 +24121,13 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) kind: "blocked" as const, issue, previousStatus: issue.status, - comment: buildExecutionReviewParticipantRecoveryComment({ latestRun: run }), + comment: buildNonRetryableExecutionReviewParticipantComment({ latestRun: run }), recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE, recoveryOwnerAgentId: activeParticipant.agentId, + expectedReviewStage: { + stageId: finalizedRunStageId, + participantAgentId: run.agentId, + }, }; } if (issue.executionRunId && issue.executionRunId !== run.id) return null; @@ -24437,7 +24452,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const issueNeedsReviewParticipantRecovery = issue.status === "in_review" && !issue.assigneeUserId && - Boolean(executionState?.currentStageId) && + runStageId !== null && runStageId === executionState?.currentStageId && currentParticipant?.type === "agent" && currentParticipant.agentId === run.agentId && @@ -24477,6 +24492,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) comment: buildExecutionReviewParticipantRecoveryComment({ latestRun: run }), recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE, recoveryOwnerAgentId: currentParticipant.agentId, + expectedReviewStage: { + stageId: runStageId, + participantAgentId: run.agentId, + }, }; } @@ -24709,6 +24728,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ? EXECUTION_REVIEW_PARTICIPANT_RECOVERY_CAUSE : undefined, recoveryOwnerAgentId: promotionResult.recoveryOwnerAgentId, + expectedReviewStage: + "expectedReviewStage" in promotionResult ? promotionResult.expectedReviewStage : undefined, }); return false; } diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index ecac4c81044f..03e31998643e 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -5124,6 +5124,7 @@ export function recoveryService( comment?: string; recoveryCause?: StrandedRecoveryCause; recoveryOwnerAgentId?: string | null; + expectedReviewStage?: { stageId: string; participantAgentId: string }; successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null; }) { if (isRoutineExecutionDuplicateSuppressedRun(input.latestRun)) { @@ -5172,6 +5173,20 @@ export function recoveryService( // function did NOT produce -- e.g. `in_review` from a park that raced it -- is a signal // that some other terminal action already claimed this issue for this cause. if (fresh.status !== input.previousStatus && fresh.status !== "blocked") return null; + if (input.expectedReviewStage) { + const executionState = parseIssueExecutionState(fresh.executionState); + const participant = executionState?.status === "pending" + ? executionState.currentParticipant + : null; + if ( + fresh.status !== "in_review" || + executionState?.currentStageId !== input.expectedReviewStage.stageId || + participant?.type !== "agent" || + participant.agentId !== input.expectedReviewStage.participantAgentId + ) { + return null; + } + } const recoveryCause = resolveStrandedRecoveryCause(input.latestRun, input.recoveryCause); const { action, hasNewActivitySinceLastAttempt } = await ensureSourceScopedStrandedRecoveryAction({ @@ -6061,7 +6076,7 @@ export function recoveryService( } if (issue.status === "in_review") { - if (!participantAgentId || !pendingExecutionState) { + if (!participantAgentId || !pendingExecutionState?.currentStageId) { result.skipped += 1; continue; } @@ -6105,6 +6120,10 @@ export function recoveryService( latestRun: participantLatestRun, recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON, recoveryOwnerAgentId: participantAgentId, + expectedReviewStage: { + stageId: pendingExecutionState.currentStageId, + participantAgentId, + }, comment: "Paperclip detected a non-retryable failure on the active review participant's run " + `(\`${participantContinuationClassification.errorCode}\`). Skipping automatic retries and moving it to ` + From c7b99cabf562fc4718fc35c958f641d4fa8aa75a Mon Sep 17 00:00:00 2001 From: Paperclip Date: Thu, 6 Aug 2026 06:33:09 +0000 Subject: [PATCH 10/15] fix(recovery): fence review finalization races Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 84 ++++++++++++++++++- .../__tests__/issue-recovery-actions.test.ts | 6 +- server/src/services/heartbeat.ts | 34 ++++++-- server/src/services/recovery/service.ts | 84 ++++++++++++++----- 4 files changed, 179 insertions(+), 29 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index e2a48f89d27a..88108165e5ea 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -32,6 +32,7 @@ import { environments, executionWorkspaces, externalRuntimeReservations, + githubCommitStatusDeliveries, heartbeatRunEvents, heartbeatRuns, issueComments, @@ -1902,8 +1903,19 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { reason: "NotFound", name: jobName, }); + const previousGateContext = process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT; + process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT = "review/ally-complete"; - const result = await heartbeat.reapOrphanedRuns({ suppressDispatchAfterReap: true }); + let result: Awaited>; + try { + result = await heartbeat.reapOrphanedRuns({ suppressDispatchAfterReap: true }); + } finally { + if (previousGateContext === undefined) { + delete process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT; + } else { + process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT = previousGateContext; + } + } expect(result.runIds).toContain(runId); expect(mockGithubHasReviewerEvidenceForPr).toHaveBeenCalledWith({ @@ -1928,6 +1940,18 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(retries.every((retry) => (retry.contextSnapshot as Record | null)?.allowDeliverableWork === false )).toBe(true); + const gateDeliveries = await db + .select() + .from(githubCommitStatusDeliveries) + .where(eq(githubCommitStatusDeliveries.sourceRunId, runId)); + expect(gateDeliveries).toHaveLength(1); + expect(gateDeliveries[0]).toMatchObject({ + context: "review/ally-complete", + repoFullName: "Blockcast/onprem-k8s", + sha: headSha, + state: "failure", + status: "queued", + }); }); async function recoverClaimedReviewWithUnavailableVerification(kind: "result" | "throw") { @@ -6407,7 +6431,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); it.each(["job_missing", "k8s_pod_schedule_failed"])( - "blocks immediate review-participant recovery after %s without replaying deliverable work", + "preserves a newer review execution after %s without replaying deliverable work", async (errorCode) => { mockAdapterExecute.mockResolvedValueOnce({ exitCode: 1, @@ -6428,19 +6452,71 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { executionStage: { stageId, stageType: "review" }, }); - const [issue, runs] = await Promise.all([ + const [issue, runs, recoveryActions] = await Promise.all([ db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null), db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)), + db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)), ]); expect(runs.some((row) => (row.contextSnapshot as Record | null)?.retryReason === "execution_review_participant_recovery" && (row.contextSnapshot as Record | null)?.allowDeliverableWork !== false )).toBe(false); - expect(issue?.status).toBe("blocked"); + expect(issue).toMatchObject({ + status: "in_review", + executionState: { + currentStageId: stageId, + currentParticipant: { type: "agent", agentId }, + }, + }); + expect(runs.some((row) => row.id !== runId)).toBe(true); + expect(recoveryActions).toHaveLength(0); }, ); + it("does not let an older terminal review run block a newer run in the same stage", async () => { + const { companyId, agentId, issueId, runId, stageId } = await seedInReviewParticipantRunFixture(); + const newerRunId = randomUUID(); + mockAdapterExecute.mockImplementationOnce(async () => { + await db.insert(heartbeatRuns).values({ + id: newerRunId, + companyId, + agentId, + invocationSource: "automation", + triggerDetail: "system", + status: "running", + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "execution_review_requested", + executionStage: { stageId, stageType: "review" }, + }, + }); + await db.update(issues).set({ executionRunId: newerRunId }).where(eq(issues.id, issueId)); + return { + exitCode: 1, + signal: null, + timedOut: false, + errorCode: "job_missing", + errorMessage: "The older external lifecycle Job disappeared", + provider: "test", + model: "test-model", + }; + }); + const heartbeat = createHeartbeat(); + + await heartbeat.resumeQueuedRuns(); + const settledRun = await waitForRunToSettle(heartbeat, runId, 8_000); + expect(settledRun).toMatchObject({ status: "failed", errorCode: "job_missing" }); + + const [issue, actions] = await Promise.all([ + db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null), + db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)), + ]); + expect(issue).toMatchObject({ status: "in_review", executionRunId: newerRunId }); + expect(actions).toHaveLength(0); + }); + it("does not let a late prior-stage failure recover a later stage with the same reviewer", async () => { const { agentId, issueId, runId } = await seedInReviewParticipantRunFixture(); const nextStageId = randomUUID(); diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 5baa8e498de9..4373219491d6 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -617,7 +617,11 @@ describeEmbeddedPostgres("issue recovery actions", () => { latestRun: staleRun, recoveryCause: "execution_review_participant_recovery", recoveryOwnerAgentId: managerId, - expectedReviewStage: { stageId: staleStageId, participantAgentId: managerId }, + expectedReviewStage: { + stageId: staleStageId, + participantAgentId: managerId, + executionRunId: null, + }, }); expect(updated).toBeNull(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 015aa13caa17..4034ce0b558a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -9293,9 +9293,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) * delivery logging; the heartbeat lifecycle only records that the work was * durably queued. (BLO-17456) */ - async function queueExhaustedPrReviewGateStatus( + async function queueFailedPrReviewGateStatus( run: typeof heartbeatRuns.$inferSelect, contextSnapshot: Record, + reason: "retry_exhausted" | "non_retryable_external_lifecycle", ) { const target = resolvePrReviewGateStatusTarget( contextSnapshot, @@ -9310,7 +9311,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) sha: target.sha, context: target.context, state: "failure", - description: "Paperclip reviewer run exhausted its automatic retries; no review was posted.", + description: reason === "retry_exhausted" + ? "Paperclip reviewer run exhausted its automatic retries; no review was posted." + : "Paperclip reviewer run ended ambiguously and was not replayed; no review was confirmed.", targetUrl: target.prUrl, prNumber: target.prNumber, prUrl: target.prUrl, @@ -9321,7 +9324,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) level: "info", message: delivery.status === "queued" || delivery.status === "processing" - ? `Queued PR-review gate status failure delivery for ${target.context} on ${target.repoFullName}@${target.sha.slice(0, 7)} after retry exhaustion` + ? `Queued PR-review gate status failure delivery for ${target.context} on ${target.repoFullName}@${target.sha.slice(0, 7)} after ${reason === "retry_exhausted" ? "retry exhaustion" : "a non-retryable external lifecycle failure"}` : `PR-review gate status failure delivery for ${target.context} on ${target.repoFullName}@${target.sha.slice(0, 7)} is already ${delivery.status}`, payload: { deliveryId: delivery.id, @@ -14057,7 +14060,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) // GitHub-evidence read and transient GitHub/token failures can retry. // Opt-in and swallowed so status delivery can never alter exhaustion // handling. - await queueExhaustedPrReviewGateStatus(run, contextSnapshot).catch((error) => { + await queueFailedPrReviewGateStatus(run, contextSnapshot, "retry_exhausted").catch((error) => { logger.warn( { err: error, runId: run.id }, "failed to queue exhausted PR-review gate status", @@ -23990,6 +23993,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) run: typeof heartbeatRuns.$inferSelect, options: { suppressImmediateRecovery?: boolean } = {}, ): Promise { + if (run.errorCode === "job_missing" || run.errorCode === "k8s_pod_schedule_failed") { + await queueFailedPrReviewGateStatus( + run, + parseObject(run.contextSnapshot), + "non_retryable_external_lifecycle", + ).catch((error) => { + logger.warn( + { err: error, runId: run.id }, + "failed to queue non-retryable PR-review gate status", + ); + }); + } + const runContext = parseObject(run.contextSnapshot); const contextIssueId = readNonEmptyString(runContext.issueId); const taskKey = deriveTaskKeyWithHeartbeatFallback(runContext, null); @@ -24107,6 +24123,13 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const activeParticipant = activeExecutionState?.status === "pending" ? activeExecutionState.currentParticipant : null; + if (issue.executionRunId && issue.executionRunId !== run.id) { + logger.info( + { issueId: issue.id, finalizingRunId: run.id, activeExecutionRunId: issue.executionRunId }, + "skipping terminal-run recovery because a newer issue execution is active", + ); + return null; + } if ( (run.errorCode === "job_missing" || run.errorCode === "k8s_pod_schedule_failed") && issue.status === "in_review" && @@ -24127,10 +24150,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) expectedReviewStage: { stageId: finalizedRunStageId, participantAgentId: run.agentId, + executionRunId: null, }, }; } - if (issue.executionRunId && issue.executionRunId !== run.id) return null; // Pre-dispatch validation recovery: if the finalizing run failed before // adapter launch, surface the primary issue for the blocked-recovery comment path. @@ -24495,6 +24518,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) expectedReviewStage: { stageId: runStageId, participantAgentId: run.agentId, + executionRunId: null, }, }; } diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 03e31998643e..f37ac88d3c0c 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -5124,7 +5124,12 @@ export function recoveryService( comment?: string; recoveryCause?: StrandedRecoveryCause; recoveryOwnerAgentId?: string | null; - expectedReviewStage?: { stageId: string; participantAgentId: string }; + expectedReviewStage?: { + stageId: string; + participantAgentId: string; + executionRunId: string | null; + }; + reviewStageClaimed?: boolean; successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null; }) { if (isRoutineExecutionDuplicateSuppressedRun(input.latestRun)) { @@ -5139,6 +5144,57 @@ export function recoveryService( }); } + if (input.expectedReviewStage) { + const expectedReviewStage = input.expectedReviewStage; + const claimed = await db.transaction(async (tx) => { + await tx.execute( + sql`select pg_advisory_xact_lock(hashtextextended(${input.issue.companyId} || ':' || ${input.issue.id}, 0))`, + ); + const [fresh] = await tx + .select() + .from(issues) + .where(eq(issues.id, input.issue.id)) + .limit(1) + .for("update"); + if (!fresh || fresh.status !== "in_review") return null; + const executionState = parseIssueExecutionState(fresh.executionState); + const participant = executionState?.status === "pending" + ? executionState.currentParticipant + : null; + if ( + executionState?.currentStageId !== expectedReviewStage.stageId || + participant?.type !== "agent" || + participant.agentId !== expectedReviewStage.participantAgentId || + fresh.executionRunId !== expectedReviewStage.executionRunId + ) { + logger.info( + { + issueId: fresh.id, + expectedReviewStage, + actualStageId: executionState?.currentStageId ?? null, + actualParticipantAgentId: participant?.type === "agent" ? participant.agentId : null, + actualExecutionRunId: fresh.executionRunId, + }, + "skipping stale review-stage recovery escalation", + ); + return null; + } + return tx + .update(issues) + .set({ status: "blocked", updatedAt: new Date() }) + .where(eq(issues.id, fresh.id)) + .returning() + .then((rows) => rows[0] ?? null); + }); + if (!claimed) return null; + return escalateStrandedAssignedIssue({ + ...input, + issue: claimed, + expectedReviewStage: undefined, + reviewStageClaimed: true, + }); + } + // Serialize escalation per (company, source-issue) so concurrent // reconcile sweeps don't fight over the same recovery-action upsert, // wakeup, and source-issue UPDATE. @@ -5172,21 +5228,9 @@ export function recoveryService( // (attempt-count bookkeeping, wake suppression) rather than no-op. Only a status this // function did NOT produce -- e.g. `in_review` from a park that raced it -- is a signal // that some other terminal action already claimed this issue for this cause. - if (fresh.status !== input.previousStatus && fresh.status !== "blocked") return null; - if (input.expectedReviewStage) { - const executionState = parseIssueExecutionState(fresh.executionState); - const participant = executionState?.status === "pending" - ? executionState.currentParticipant - : null; - if ( - fresh.status !== "in_review" || - executionState?.currentStageId !== input.expectedReviewStage.stageId || - participant?.type !== "agent" || - participant.agentId !== input.expectedReviewStage.participantAgentId - ) { - return null; - } - } + if (input.reviewStageClaimed) { + if (fresh.status !== "blocked") return null; + } else if (fresh.status !== input.previousStatus && fresh.status !== "blocked") return null; const recoveryCause = resolveStrandedRecoveryCause(input.latestRun, input.recoveryCause); const { action, hasNewActivitySinceLastAttempt } = await ensureSourceScopedStrandedRecoveryAction({ @@ -5220,11 +5264,12 @@ export function recoveryService( hasNewActivitySinceLastAttempt, }); - const updated = await issuesSvc.update(input.issue.id, { - status: "blocked", + const issueUpdate = { + status: "blocked" as const, blockedByIssueIds: blockerIds, assigneeAgentId: action.ownerAgentId ?? fresh.assigneeAgentId, - }); + }; + const updated = await issuesSvc.update(input.issue.id, issueUpdate); if (!updated) return null; if (isProviderQuotaWait) return updated; @@ -6123,6 +6168,7 @@ export function recoveryService( expectedReviewStage: { stageId: pendingExecutionState.currentStageId, participantAgentId, + executionRunId: issue.executionRunId, }, comment: "Paperclip detected a non-retryable failure on the active review participant's run " + From 0ebd9ffc1aa3858f06119646c8f99a1963372bee Mon Sep 17 00:00:00 2001 From: Paperclip Date: Thu, 6 Aug 2026 07:40:24 +0000 Subject: [PATCH 11/15] fix(recovery): close PR finalization race gaps Co-Authored-By: Paperclip --- .../heartbeat-process-recovery.test.ts | 95 ++++++++++++++++++- server/src/services/heartbeat.ts | 52 ++++++---- server/src/services/recovery/service.ts | 79 +++++---------- 3 files changed, 151 insertions(+), 75 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 88108165e5ea..0d9b34a85e03 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1954,6 +1954,66 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it.each(["pr_review_output_missing", "pr_review_verification_unavailable"])( + "terminalizes the PR gate for non-retryable %s after adapter invocation", + async (errorCode) => { + const headSha = "075a9aeff53a229199ab0583e916f33c22459983"; + mockAdapterExecute.mockResolvedValueOnce({ + exitCode: 1, + signal: null, + timedOut: false, + errorCode, + errorMessage: "Review evidence could not be confirmed", + resultJson: { externalLifecycleRecovery: { adapterInvocationStarted: true } }, + provider: "test", + model: "test-model", + }); + const { runId, issueId } = await seedQueuedIssueRunFixture(); + await db.update(heartbeatRuns).set({ + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "github_pr_review_requested", + reviewKind: "pr_review", + taskKey: `pr_review:Blockcast/paperclip:1048:${headSha}`, + githubRepoFullName: "Blockcast/paperclip", + githubPrNumber: 1048, + githubHeadSha: headSha, + }, + }).where(eq(heartbeatRuns.id, runId)); + const previousGateContext = process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT; + process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT = "review/ally-complete"; + + try { + await heartbeat.resumeQueuedRuns(); + expect(await waitForRunToSettle(heartbeat, runId, 8_000)).toMatchObject({ + status: "failed", + errorCode, + }); + } finally { + if (previousGateContext === undefined) { + delete process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT; + } else { + process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT = previousGateContext; + } + } + + const [gateDeliveries, issue] = await Promise.all([ + db.select().from(githubCommitStatusDeliveries).where( + eq(githubCommitStatusDeliveries.sourceRunId, runId), + ), + db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null), + ]); + expect(gateDeliveries).toHaveLength(1); + expect(gateDeliveries[0]).toMatchObject({ + context: "review/ally-complete", + sha: headSha, + state: "failure", + }); + expect(issue?.status).toBe("blocked"); + }, + ); + async function recoverClaimedReviewWithUnavailableVerification(kind: "result" | "throw") { const jobName = `agent-opencode-review-verification-${kind}`; const headSha = "075a9aeff53a229199ab0583e916f33c22459983"; @@ -6477,6 +6537,20 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { it("does not let an older terminal review run block a newer run in the same stage", async () => { const { companyId, agentId, issueId, runId, stageId } = await seedInReviewParticipantRunFixture(); const newerRunId = randomUUID(); + const headSha = "075a9aeff53a229199ab0583e916f33c22459983"; + await db.update(heartbeatRuns).set({ + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "execution_review_requested", + executionStage: { stageId, stageType: "review" }, + reviewKind: "pr_review", + taskKey: `pr_review:Blockcast/paperclip:1048:${headSha}`, + githubRepoFullName: "Blockcast/paperclip", + githubPrNumber: 1048, + githubHeadSha: headSha, + }, + }).where(eq(heartbeatRuns.id, runId)); mockAdapterExecute.mockImplementationOnce(async () => { await db.insert(heartbeatRuns).values({ id: newerRunId, @@ -6504,17 +6578,32 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }; }); const heartbeat = createHeartbeat(); + const previousGateContext = process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT; + process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT = "review/ally-complete"; - await heartbeat.resumeQueuedRuns(); - const settledRun = await waitForRunToSettle(heartbeat, runId, 8_000); + let settledRun: Awaited>; + try { + await heartbeat.resumeQueuedRuns(); + settledRun = await waitForRunToSettle(heartbeat, runId, 8_000); + } finally { + if (previousGateContext === undefined) { + delete process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT; + } else { + process.env.PAPERCLIP_PR_REVIEW_GATE_STATUS_CONTEXT = previousGateContext; + } + } expect(settledRun).toMatchObject({ status: "failed", errorCode: "job_missing" }); - const [issue, actions] = await Promise.all([ + const [issue, actions, gateDeliveries] = await Promise.all([ db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null), db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)), + db.select().from(githubCommitStatusDeliveries).where( + eq(githubCommitStatusDeliveries.sourceRunId, runId), + ), ]); expect(issue).toMatchObject({ status: "in_review", executionRunId: newerRunId }); expect(actions).toHaveLength(0); + expect(gateDeliveries).toHaveLength(0); }); it("does not let a late prior-stage failure recover a later stage with the same reviewer", async () => { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 4034ce0b558a..582e36be5f27 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -2441,6 +2441,20 @@ async function hasGitPushRemote(cwd: string | null | undefined) { return false; } +function isNonRetryablePrReviewTerminalOutcome( + run: Pick, +) { + if (run.errorCode === "job_missing" || run.errorCode === "k8s_pod_schedule_failed") return true; + if ( + run.errorCode !== "pr_review_output_missing" && + run.errorCode !== "pr_review_verification_unavailable" + ) { + return false; + } + const recovery = parseObject(parseObject(run.resultJson).externalLifecycleRecovery); + return recovery.adapterInvocationStarted === true; +} + export async function assertGitWorktreeBaseWorkspaceReady(input: { requestedExecutionWorkspaceMode: ReturnType; config: Record; @@ -23993,19 +24007,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) run: typeof heartbeatRuns.$inferSelect, options: { suppressImmediateRecovery?: boolean } = {}, ): Promise { - if (run.errorCode === "job_missing" || run.errorCode === "k8s_pod_schedule_failed") { - await queueFailedPrReviewGateStatus( - run, - parseObject(run.contextSnapshot), - "non_retryable_external_lifecycle", - ).catch((error) => { - logger.warn( - { err: error, runId: run.id }, - "failed to queue non-retryable PR-review gate status", - ); - }); - } - const runContext = parseObject(run.contextSnapshot); const contextIssueId = readNonEmptyString(runContext.issueId); const taskKey = deriveTaskKeyWithHeartbeatFallback(runContext, null); @@ -24128,10 +24129,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) { issueId: issue.id, finalizingRunId: run.id, activeExecutionRunId: issue.executionRunId }, "skipping terminal-run recovery because a newer issue execution is active", ); - return null; + return { kind: "superseded" as const }; } if ( - (run.errorCode === "job_missing" || run.errorCode === "k8s_pod_schedule_failed") && + isNonRetryablePrReviewTerminalOutcome(run) && issue.status === "in_review" && !issue.assigneeUserId && finalizedRunStageId !== null && @@ -24622,8 +24623,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const shouldBlockImmediately = !recoveryAgentInvokable || !recoveryAgent || - run.errorCode === "job_missing" || - run.errorCode === "k8s_pod_schedule_failed" || + isNonRetryablePrReviewTerminalOutcome(run) || isWorkspaceValidationFailedRun(run) || isConfigurationIncompleteFailedRun(run) || didAutomaticRecoveryFail(run, issue.status === "todo" ? "assignment_recovery" : "issue_continuation_needed"); @@ -24737,6 +24737,22 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; }); + if ( + promotionResult?.kind !== "superseded" && + isNonRetryablePrReviewTerminalOutcome(run) + ) { + await queueFailedPrReviewGateStatus( + run, + runContext, + "non_retryable_external_lifecycle", + ).catch((error) => { + logger.warn( + { err: error, runId: run.id }, + "failed to queue non-retryable PR-review gate status", + ); + }); + } + if (promotionResult?.kind === "blocked") { await recovery.escalateStrandedAssignedIssue({ issue: promotionResult.issue, @@ -24767,7 +24783,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return false; } - const promotedRun = promotionResult?.run ?? null; + const promotedRun = promotionResult && "run" in promotionResult ? promotionResult.run : null; if (!promotedRun) return false; if (promotionResult?.kind === "promoted" && promotionResult.reopenedActivity) { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index f37ac88d3c0c..d66ff6057dc6 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -5129,7 +5129,6 @@ export function recoveryService( participantAgentId: string; executionRunId: string | null; }; - reviewStageClaimed?: boolean; successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null; }) { if (isRoutineExecutionDuplicateSuppressedRun(input.latestRun)) { @@ -5144,57 +5143,6 @@ export function recoveryService( }); } - if (input.expectedReviewStage) { - const expectedReviewStage = input.expectedReviewStage; - const claimed = await db.transaction(async (tx) => { - await tx.execute( - sql`select pg_advisory_xact_lock(hashtextextended(${input.issue.companyId} || ':' || ${input.issue.id}, 0))`, - ); - const [fresh] = await tx - .select() - .from(issues) - .where(eq(issues.id, input.issue.id)) - .limit(1) - .for("update"); - if (!fresh || fresh.status !== "in_review") return null; - const executionState = parseIssueExecutionState(fresh.executionState); - const participant = executionState?.status === "pending" - ? executionState.currentParticipant - : null; - if ( - executionState?.currentStageId !== expectedReviewStage.stageId || - participant?.type !== "agent" || - participant.agentId !== expectedReviewStage.participantAgentId || - fresh.executionRunId !== expectedReviewStage.executionRunId - ) { - logger.info( - { - issueId: fresh.id, - expectedReviewStage, - actualStageId: executionState?.currentStageId ?? null, - actualParticipantAgentId: participant?.type === "agent" ? participant.agentId : null, - actualExecutionRunId: fresh.executionRunId, - }, - "skipping stale review-stage recovery escalation", - ); - return null; - } - return tx - .update(issues) - .set({ status: "blocked", updatedAt: new Date() }) - .where(eq(issues.id, fresh.id)) - .returning() - .then((rows) => rows[0] ?? null); - }); - if (!claimed) return null; - return escalateStrandedAssignedIssue({ - ...input, - issue: claimed, - expectedReviewStage: undefined, - reviewStageClaimed: true, - }); - } - // Serialize escalation per (company, source-issue) so concurrent // reconcile sweeps don't fight over the same recovery-action upsert, // wakeup, and source-issue UPDATE. @@ -5228,8 +5176,31 @@ export function recoveryService( // (attempt-count bookkeeping, wake suppression) rather than no-op. Only a status this // function did NOT produce -- e.g. `in_review` from a park that raced it -- is a signal // that some other terminal action already claimed this issue for this cause. - if (input.reviewStageClaimed) { - if (fresh.status !== "blocked") return null; + if (input.expectedReviewStage) { + const executionState = parseIssueExecutionState(fresh.executionState); + const participant = executionState?.status === "pending" + ? executionState.currentParticipant + : null; + if ( + fresh.status !== "in_review" || + executionState?.currentStageId !== input.expectedReviewStage.stageId || + participant?.type !== "agent" || + participant.agentId !== input.expectedReviewStage.participantAgentId || + fresh.executionRunId !== input.expectedReviewStage.executionRunId + ) { + logger.info( + { + issueId: fresh.id, + expectedReviewStage: input.expectedReviewStage, + actualStatus: fresh.status, + actualStageId: executionState?.currentStageId ?? null, + actualParticipantAgentId: participant?.type === "agent" ? participant.agentId : null, + actualExecutionRunId: fresh.executionRunId, + }, + "skipping stale review-stage recovery escalation", + ); + return null; + } } else if (fresh.status !== input.previousStatus && fresh.status !== "blocked") return null; const recoveryCause = resolveStrandedRecoveryCause(input.latestRun, input.recoveryCause); From 4e97d20d1a6befee3271e06226ff165a37ebd431 Mon Sep 17 00:00:00 2001 From: Paperclip Date: Thu, 6 Aug 2026 08:18:39 +0000 Subject: [PATCH 12/15] fix(recovery): make review finalization durable Co-Authored-By: Paperclip --- .../services/github-status-delivery-outbox.ts | 3 +- server/src/services/heartbeat.ts | 57 +++++-- server/src/services/issue-recovery-actions.ts | 2 +- server/src/services/recovery/service.ts | 149 ++++++++++++------ 4 files changed, 154 insertions(+), 57 deletions(-) diff --git a/server/src/services/github-status-delivery-outbox.ts b/server/src/services/github-status-delivery-outbox.ts index 9bc86fe20e62..c3ce947ff8d3 100644 --- a/server/src/services/github-status-delivery-outbox.ts +++ b/server/src/services/github-status-delivery-outbox.ts @@ -15,6 +15,7 @@ import { } from "./github-app-auth.js"; type DeliveryRow = typeof githubCommitStatusDeliveries.$inferSelect; +type DbTransaction = Parameters[0]>[0]; type DeliveryTerminalStatus = "delivered" | "skipped" | "failed" | "failed_permanent"; const POLL_INTERVAL_MS = 5_000; @@ -355,7 +356,7 @@ async function processDelivery(db: Db, row: DeliveryRow): Promise { } export async function enqueueGithubCommitStatusDelivery( - db: Db, + db: Db | DbTransaction, input: EnqueueGithubCommitStatusDeliveryInput, ): Promise { const now = new Date(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 582e36be5f27..5d73cdd64565 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -9311,6 +9311,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) run: typeof heartbeatRuns.$inferSelect, contextSnapshot: Record, reason: "retry_exhausted" | "non_retryable_external_lifecycle", + dbOrTx: Db | DbTransaction = db, + appendEvent = true, ) { const target = resolvePrReviewGateStatusTarget( contextSnapshot, @@ -9318,7 +9320,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ); if (!target) return; - const delivery = await enqueueGithubCommitStatusDelivery(db, { + const delivery = await enqueueGithubCommitStatusDelivery(dbOrTx, { companyId: run.companyId, sourceRunId: run.id, repoFullName: target.repoFullName, @@ -9332,6 +9334,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) prNumber: target.prNumber, prUrl: target.prUrl, }); + if (appendEvent) await appendFailedPrReviewGateStatusEvent(run, target, reason, delivery); + return delivery; + } + + async function appendFailedPrReviewGateStatusEvent( + run: typeof heartbeatRuns.$inferSelect, + target: NonNullable>, + reason: "retry_exhausted" | "non_retryable_external_lifecycle", + delivery: Awaited>, + ) { await appendRunEvent(run, await nextRunEventSeq(run.id), { eventType: "lifecycle", stream: "system", @@ -24019,6 +24031,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const recoverySessionBefore = recoveryAgentInvokable ? await resolveSessionBeforeForWakeup(recoveryAgent, taskKey) : null; + let gateDelivery: Awaited> | null = null; const promotionResult = await db.transaction(async (tx) => { // Lock the context issue (if any) AND every issue that still references this run. // @@ -24116,7 +24129,18 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ? candidateIssues.find((candidate) => candidate.id === contextIssueId) : candidateIssues[0]) ?? null; - if (!issue) return null; + if (!issue) { + if (isNonRetryablePrReviewTerminalOutcome(run)) { + gateDelivery = await queueFailedPrReviewGateStatus( + run, + runContext, + "non_retryable_external_lifecycle", + tx, + false, + ) ?? null; + } + return null; + } const activeExecutionState = parseIssueExecutionState(issue.executionState); const finalizedRunExecutionStage = parseObject(runContext.executionStage); const finalizedRunStageId = @@ -24131,6 +24155,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ); return { kind: "superseded" as const }; } + if (isNonRetryablePrReviewTerminalOutcome(run)) { + // The outbox row is part of the ownership decision: a replacement run + // cannot claim this issue until both the lock release and delivery + // intent commit. Publishing the informational event can remain best + // effort after commit because the outbox is the durable artifact. + gateDelivery = await queueFailedPrReviewGateStatus( + run, + runContext, + "non_retryable_external_lifecycle", + tx, + false, + ) ?? null; + } if ( isNonRetryablePrReviewTerminalOutcome(run) && issue.status === "in_review" && @@ -24737,18 +24774,20 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; }); - if ( - promotionResult?.kind !== "superseded" && - isNonRetryablePrReviewTerminalOutcome(run) - ) { - await queueFailedPrReviewGateStatus( - run, + if (gateDelivery) { + const target = resolvePrReviewGateStatusTarget( runContext, + loadConfig().prReviewGateStatusContext, + ); + if (target) await appendFailedPrReviewGateStatusEvent( + run, + target, "non_retryable_external_lifecycle", + gateDelivery, ).catch((error) => { logger.warn( { err: error, runId: run.id }, - "failed to queue non-retryable PR-review gate status", + "failed to append non-retryable PR-review gate status event", ); }); } diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 83c34d34aca5..34a14eb9a697 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -220,7 +220,7 @@ function isUniqueRecoveryActionConflict(error: unknown) { ); } -export function issueRecoveryActionService(db: Db) { +export function issueRecoveryActionService(db: DbOrTransaction) { const upsertQueues = new Map>(); async function runExclusiveUpsert( diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index d66ff6057dc6..34cdf6c8fe59 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -89,6 +89,8 @@ import { } from "./zero-token-startup-failure.js"; import { clearAgentTaskSessions } from "./session-reset.js"; +type DbTransaction = Parameters[0]>[0]; + 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; @@ -4193,7 +4195,8 @@ export function recoveryService( recoveryCause?: StrandedRecoveryCause; recoveryOwnerAgentId?: string | null; successfulRunHandoffEvidence?: SuccessfulRunHandoffRecoveryEvidence | null; - }) { + }, dbOrTx: Db | DbTransaction = db) { + const actionSvc = dbOrTx === db ? recoveryActionsSvc : issueRecoveryActionService(dbOrTx); const recoveryCause = resolveStrandedRecoveryCause(input.latestRun, input.recoveryCause); const routing = await resolveStrandedRecoveryRouting({ issue: input.issue, @@ -4221,14 +4224,14 @@ export function recoveryService( // Read existing action before upsert so we can compare lastAttemptAt against // issue.lastActivityAt and suppress duplicate non-assignee wakes when nothing changed. - const existingAction = await recoveryActionsSvc.getActiveForIssue(input.issue.companyId, input.issue.id); + const existingAction = await actionSvc.getActiveForIssue(input.issue.companyId, input.issue.id); const previousAttemptAt = existingAction?.lastAttemptAt ? new Date(existingAction.lastAttemptAt as Date | string) : null; const hasNewActivitySinceLastAttempt = !previousAttemptAt || input.issue.lastActivityAt > previousAttemptAt; - const action = await recoveryActionsSvc.upsertSourceScoped({ + const action = await actionSvc.upsertSourceScoped({ companyId: input.issue.companyId, sourceIssueId: input.issue.id, kind: strandedRecoveryActionKind(recoveryCause), @@ -4872,8 +4875,12 @@ export function recoveryService( return cycleForming; } - async function unresolvedBlockerHumanDecisionEscalationState(companyId: string, issueId: string) { - const blockerRows = await db + async function unresolvedBlockerHumanDecisionEscalationState( + companyId: string, + issueId: string, + dbOrTx: Db | DbTransaction = db, + ) { + const blockerRows = await dbOrTx .select({ blockerIssueId: issueRelations.issueId, assigneeAgentId: issues.assigneeAgentId, @@ -5149,7 +5156,7 @@ export function recoveryService( // The advisory lock is xact-scoped on this tx's connection; once we // commit/return, waiting peers wake up and record their next attempt // against the same active source-scoped action. - return await db.transaction(async (tx) => { + const escalation = await db.transaction(async (tx) => { await tx.execute( sql`select pg_advisory_xact_lock(hashtextextended(${input.issue.companyId} || ':' || ${input.issue.id}, 0))`, ); @@ -5157,11 +5164,14 @@ export function recoveryService( // 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. - const [fresh] = await 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; if (!fresh) return null; // BLO-18643: mirror the park paths' own re-check (parkReviewWaitingContinuationIssue / // parkNoDependencyReviewWaitingIssue both bail if `fresh.status` has moved past the @@ -5204,6 +5214,7 @@ export function recoveryService( } else if (fresh.status !== input.previousStatus && fresh.status !== "blocked") return null; const recoveryCause = resolveStrandedRecoveryCause(input.latestRun, input.recoveryCause); + const mutationDb = input.expectedReviewStage ? tx : db; const { action, hasNewActivitySinceLastAttempt } = await ensureSourceScopedStrandedRecoveryAction({ issue: fresh, previousStatus: input.previousStatus, @@ -5211,38 +5222,50 @@ export function recoveryService( recoveryCause, recoveryOwnerAgentId: input.recoveryOwnerAgentId, successfulRunHandoffEvidence: input.successfulRunHandoffEvidence, - }); + }, mutationDb); const isProviderQuotaWait = recoveryCause === "provider_quota" && !action.ownerAgentId && Boolean(action.returnOwnerAgentId); - if (isProviderQuotaWait && action.returnOwnerAgentId) { - await ensureProviderQuotaWaitRecoveryMonitor({ - issue: fresh, - latestRun: input.latestRun, - actionId: action.id, - agentId: action.returnOwnerAgentId, - }); - } const { blockerIssueIds: blockerIds, needsHumanDecision, - } = await unresolvedBlockerHumanDecisionEscalationState(fresh.companyId, fresh.id); + } = await unresolvedBlockerHumanDecisionEscalationState(fresh.companyId, fresh.id, mutationDb); - await enqueueSourceScopedStrandedRecoveryWake({ - action, - issue: fresh, - latestRun: input.latestRun, - recoveryCause, - hasNewActivitySinceLastAttempt, - }); + 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, + }); + } const issueUpdate = { status: "blocked" as const, blockedByIssueIds: blockerIds, assigneeAgentId: action.ownerAgentId ?? fresh.assigneeAgentId, }; - const updated = await issuesSvc.update(input.issue.id, issueUpdate); + const updated = await issuesSvc.update(input.issue.id, issueUpdate, mutationDb); if (!updated) return null; - if (isProviderQuotaWait) return updated; + if (isProviderQuotaWait) { + return { + updated, + action, + fresh, + recoveryCause, + hasNewActivitySinceLastAttempt, + needsHumanDecision, + blockerIds, + }; + } const prefix = await getCompanyIssuePrefix(fresh.companyId); const workspacePreflightHandoffCause = describeWorkspacePreflightRecoveryCause(input.latestRun); @@ -5314,7 +5337,7 @@ export function recoveryService( const escalationCommentMarker = announcesReassignment ? reassignmentMarker : `Recovery action: \`${action.id}\``; - const hasEscalationComment = await db + const hasEscalationComment = await mutationDb .select({ id: issueComments.id, body: issueComments.body, metadata: issueComments.metadata }) .from(issueComments) .where(and(eq(issueComments.issueId, fresh.id), eq(issueComments.authorType, "system"))) @@ -5333,13 +5356,14 @@ export function recoveryService( authorType: "system", presentation: notice.presentation, metadata: notice.metadata, - }); + }, mutationDb); } else { await issuesSvc.addComment( fresh.id, `${input.comment ?? "Automatic stranded-work recovery needs manual attention."}${recoveryLine}`, {}, { authorType: "system" }, + mutationDb, ); } } @@ -5374,7 +5398,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 db + const alreadyAnnounced = await mutationDb .select({ id: issueComments.id }) .from(issueComments) .where(and( @@ -5415,6 +5439,7 @@ export function recoveryService( ].join("\n"), {}, { authorType: "system" }, + mutationDb, ); } } @@ -5465,24 +5490,56 @@ export function recoveryService( }, }); - if (needsHumanDecision) { - const assigneeAgent = fresh.assigneeAgentId - ? await db - .select({ name: agents.name }) - .from(agents) - .where(and(eq(agents.companyId, fresh.companyId), eq(agents.id, fresh.assigneeAgentId))) - .limit(1) - .then((rows) => rows[0] ?? null) - : null; - await emitNeedsHumanDecisionEscalationEvent({ - issue: fresh, - assigneeAgentName: assigneeAgent?.name ?? null, - blockedByIssueIds: blockerIds, - }); - } - - return updated; + return { + updated, + action, + fresh, + recoveryCause, + hasNewActivitySinceLastAttempt, + needsHumanDecision, + blockerIds, + }; }); + if (!escalation) return null; + + // The active recovery action committed above is the durable wake intent. + // Dispatch after releasing the stage row lock: enqueueWakeup may claim the + // 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) { + await ensureProviderQuotaWaitRecoveryMonitor({ + issue: escalation.fresh, + 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, + }); + } + if (escalation.needsHumanDecision) { + const assigneeAgent = escalation.fresh.assigneeAgentId + ? await db + .select({ name: agents.name }) + .from(agents) + .where(and(eq(agents.companyId, escalation.fresh.companyId), eq(agents.id, escalation.fresh.assigneeAgentId))) + .limit(1) + .then((rows) => rows[0] ?? null) + : null; + await emitNeedsHumanDecisionEscalationEvent({ + issue: escalation.fresh, + assigneeAgentName: assigneeAgent?.name ?? null, + blockedByIssueIds: escalation.blockerIds, + }); + } + return escalation.updated; } function buildZeroTokenStartupFailureComment(input: { From 8da884cddf2b6116d5210f14140cf8bfecec0e7d Mon Sep 17 00:00:00 2001 From: "allyblockcast[bot]" Date: Sat, 8 Aug 2026 08:09:42 +0000 Subject: [PATCH 13/15] fix(recovery): defer the review-stage escalation publish Completes the phantom-activity fix from the preceding commit. That commit moved `logActivity` onto `mutationDb` so the activity row rolls back with the escalation, but omitted the `deferPublish` option -- and `logActivity` only returns the real publisher when that option is set (activity-log.ts:322). Without it the live/plugin events still fired inline, inside the transaction, so a rolled-back review-stage escalation could still emit a phantom `issue.updated`. The surrounding comment and the deferred `publishEscalationActivity?.()` call both described behaviour the code did not have. Passes `deferPublish: Boolean(input.expectedReviewStage)`, matching the `input.expectedReviewStage ? tx : db` binding that decides whether `mutationDb` is a transaction at all. The non-review path keeps publishing inline, where the connection is autocommit and there is no commit to wait for. Extracted from a2a5fcc38 -- deliberately WITHOUT that commit's `claimWakeAttempt` backstop-cooldown hunk, which is BLO-22795 scope and stays on #1131. Tests: two `logActivity` contract cases -- a deferred log whose transaction rolls back emits no live event and leaves no row; the committed counterpart does emit when the returned publisher is invoked. The second is a positive control, since the rollback assertion alone would pass vacuously against a mis-wired subscription. Refs BLO-18106 Co-Authored-By: Claude --- .../activity-log-responsible-user.test.ts | 118 ++++++++++++++++++ server/src/services/recovery/service.ts | 8 ++ 2 files changed, 126 insertions(+) diff --git a/server/src/__tests__/activity-log-responsible-user.test.ts b/server/src/__tests__/activity-log-responsible-user.test.ts index d179cc7c98a8..8eaf25dfde6f 100644 --- a/server/src/__tests__/activity-log-responsible-user.test.ts +++ b/server/src/__tests__/activity-log-responsible-user.test.ts @@ -20,6 +20,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { subscribeCompanyLiveEvents } from "../services/live-events.js"; type TableRows = Map>>; @@ -231,4 +232,121 @@ describeEmbeddedPostgres("logActivity responsible-user stamping", () => { expect(row?.responsibleUserId).toBe("key-user"); }); + + // The contract `deferPublish` exists to enforce: the live event must not escape a + // transaction that later rolls back. `publishLiveEvent` is in-memory and the plugin + // outbox writes on its own handle, so both bypass the enclosing transaction entirely -- + // publishing inline turns a rollback into a phantom event for an activity row that + // never existed. Callers logging inside a transaction (the review-stage recovery + // escalation is one) therefore have to pass the option and fire the returned publisher + // after commit. + it("does not publish a live event when a deferred activity's transaction rolls back", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + defaultResponsibleUserId: "default-user", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "running", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + const seen: unknown[] = []; + const unsubscribe = subscribeCompanyLiveEvents(companyId, (event) => { + seen.push(event); + }); + + try { + await expect(db.transaction(async (tx) => { + await logActivity(tx as unknown as Db, activityInput({ + companyId, + actorId: agentId, + agentId, + entityType: "agent", + entityId: agentId, + }), { deferPublish: true }); + // Stand-in for any post-log failure inside the same transaction. Because the + // publisher was deferred, nothing has been emitted yet at this point. + throw new Error("rollback"); + })).rejects.toThrow("rollback"); + } finally { + unsubscribe(); + } + + expect(seen).toHaveLength(0); + const rows = await db + .select({ id: activityLog.id }) + .from(activityLog) + .where(eq(activityLog.companyId, companyId)); + expect(rows).toHaveLength(0); + }); + + // Positive control for the test above. Asserting "no event was emitted" proves nothing + // on its own -- a subscription wired to the wrong channel would satisfy it vacuously. + // This pins the other half of the contract: the deferred publisher is the real one, and + // invoking it after commit does emit on the same channel the rollback test watches. + it("publishes the deferred live event when the transaction commits", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + defaultResponsibleUserId: "default-user", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "running", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + const seen: unknown[] = []; + const unsubscribe = subscribeCompanyLiveEvents(companyId, (event) => { + seen.push(event); + }); + + try { + const publish = await db.transaction(async (tx) => { + return await logActivity(tx as unknown as Db, activityInput({ + companyId, + actorId: agentId, + agentId, + entityType: "agent", + entityId: agentId, + }), { deferPublish: true }); + }); + // Still nothing: publication was handed back rather than fired inside the tx. + expect(seen).toHaveLength(0); + publish(); + expect(seen).toHaveLength(1); + } finally { + unsubscribe(); + } + + const rows = await db + .select({ id: activityLog.id }) + .from(activityLog) + .where(eq(activityLog.companyId, companyId)); + expect(rows).toHaveLength(1); + }); }); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 34cdf6c8fe59..a1395085de33 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -5488,6 +5488,14 @@ export function recoveryService( returnOwnerAgentId: action.returnOwnerAgentId, 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), }); return { From 8c882dea4be2b3da9dcd25b0c6e2d07ca72f937e Mon Sep 17 00:00:00 2001 From: "allyblockcast[bot]" Date: Sun, 9 Aug 2026 04:59:13 +0000 Subject: [PATCH 14/15] fix(recovery): pass expectedReviewStage on all review-participant escalations The three unguarded escalateStrandedAssignedIssue call sites carrying EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON were misclassified as non-review escalations, so they took the non-transactional branch of `input.expectedReviewStage ? tx : db` and wrote through the outer db -- phantom activity that survives a rollback. They also skipped the FOR UPDATE lock and the stale-stage rejection guard. All four review-participant sites now pass the flag. Both are after the line-6192 guard that proves participantAgentId and pendingExecutionState.currentStageId are non-null. Refs BLO-18106 --- server/src/services/recovery/service.ts | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index a1395085de33..49e9fbfd0870 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -6172,6 +6172,11 @@ export function recoveryService( comment: buildExecutionReviewParticipantUnavailableComment(participantLatestRun), recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON, recoveryOwnerAgentId: participantAgentId, + expectedReviewStage: { + stageId: pendingExecutionState.currentStageId, + participantAgentId, + executionRunId: issue.executionRunId, + }, }); if (updated) { result.escalated += 1; @@ -6274,6 +6279,11 @@ export function recoveryService( comment: buildExecutionReviewParticipantUnavailableComment(participantLatestRun), recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON, recoveryOwnerAgentId: participantAgentId, + expectedReviewStage: { + stageId: pendingExecutionState.currentStageId, + participantAgentId, + executionRunId: issue.executionRunId, + }, }); if (updated) { result.escalated += 1; @@ -6292,6 +6302,11 @@ export function recoveryService( comment: buildExecutionReviewParticipantRecoveryComment(participantLatestRun), recoveryCause: EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON, recoveryOwnerAgentId: participantAgentId, + expectedReviewStage: { + stageId: pendingExecutionState.currentStageId, + participantAgentId, + executionRunId: issue.executionRunId, + }, }); if (updated) { result.escalated += 1; From cc7d3f8e8fd9e3d2ea6d477f2870ccafbd039959 Mon Sep 17 00:00:00 2001 From: "allyblockcast[bot]" Date: Sun, 9 Aug 2026 05:12:01 +0000 Subject: [PATCH 15/15] fix(recovery): stage-scope the PR-review gate write and see queued replacements Two fixes in releaseIssueExecutionAndPromote: 1. Stage-scope the failed-PR-review gate write. It fired on every non-retryable PR-review terminal outcome, including after the review stage had already advanced -- marking the PR failed for a stage this run no longer owns. Suppressed only when the move is provable (both stage ids known and different); an unknown stage stays fail-open. 2. hasQueuedReplacementIssueWake. master's 8446c1011 moved the issues.executionRunId write from enqueue time to claim time, so the supersession guard could not see a replacement review run that was queued but not yet claimed, and the review-participant escalation moved the issue to blocked while its replacement was pending. Mirrors hasQueuedIssueWake; evaluated last so the query stays off the hot path. Refs BLO-18106 --- server/src/services/heartbeat.ts | 48 ++++++++++++++++++++++++++++++-- 1 file changed, 46 insertions(+), 2 deletions(-) diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 5d73cdd64565..cfa8138ca206 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -24015,6 +24015,37 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ); } + /** + * BLO-18106: master's `8446c1011` ("bind issue locks only for running runs") + * moved the `issues.executionRunId` write from *enqueue* time to *claim* + * time. The supersession guard above (`issue.executionRunId !== run.id`) was + * written against the enqueue-time binding, so a replacement review run that + * is queued but not yet claimed is invisible to it -- and the review-participant + * escalation then moves the issue to `blocked` while its replacement is + * already pending, which is exactly the strand this path exists to prevent. + * + * Mirrors `hasQueuedIssueWake` in recovery/service.ts. Callers MUST evaluate + * this last in their condition chain so the query stays off the hot path. + */ + async function hasQueuedReplacementIssueWake( + dbOrTx: typeof db | Parameters[0]>[0], + companyId: string, + issueId: string, + ) { + return dbOrTx + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.status, "queued"), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, + ), + ) + .limit(1) + .then((rows) => Boolean(rows[0])); + } + async function releaseIssueExecutionAndPromote( run: typeof heartbeatRuns.$inferSelect, options: { suppressImmediateRecovery?: boolean } = {}, @@ -24148,6 +24179,17 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const activeParticipant = activeExecutionState?.status === "pending" ? activeExecutionState.currentParticipant : null; + // BLO-18106: the failed-PR-review gate is a durable, PR-visible artifact. + // Writing it once the review stage has already advanced marks the PR + // failed for a stage this run no longer owns -- e.g. a replacement run + // has since taken the stage and may have completed the review. Only + // suppress when the move is *provable* (both stage ids known and + // different); an unknown stage stays fail-open so genuine failures still + // surface on the PR. + const finalizedRunStageSuperseded = + finalizedRunStageId !== null && + activeExecutionState?.currentStageId != null && + activeExecutionState.currentStageId !== finalizedRunStageId; if (issue.executionRunId && issue.executionRunId !== run.id) { logger.info( { issueId: issue.id, finalizingRunId: run.id, activeExecutionRunId: issue.executionRunId }, @@ -24155,7 +24197,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ); return { kind: "superseded" as const }; } - if (isNonRetryablePrReviewTerminalOutcome(run)) { + if (isNonRetryablePrReviewTerminalOutcome(run) && !finalizedRunStageSuperseded) { // The outbox row is part of the ownership decision: a replacement run // cannot claim this issue until both the lock release and delivery // intent commit. Publishing the informational event can remain best @@ -24176,7 +24218,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) finalizedRunStageId === activeExecutionState?.currentStageId && activeParticipant?.type === "agent" && activeParticipant.agentId === run.agentId && - isExecutionReviewParticipantRecoveryEligibleRun(run) + isExecutionReviewParticipantRecoveryEligibleRun(run) && + // Evaluated last on purpose: keeps the extra query off the hot path. + !(await hasQueuedReplacementIssueWake(tx, issue.companyId, issue.id)) ) { return { kind: "blocked" as const,