fix(queue): cap backlog-convergence re-review attempts per head SHA

src/queue/processors.ts

@@ -1052,6 +1052,49 @@ function regateRepairTargetKey(repoFullName: string, prNumber: number, headSha:
   return `${repoFullName}#${prNumber}#${headSha}`;
 }
 
+/**
+ * True when `pr`'s current head SHA has already exhausted REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA repair attempts
+ * within the lookback window. Records (at most once per SHA) the exhausted audit event + error-level log as a
+ * side effect the first time a SHA crosses the cap. Shared by both surfaceRepairPriorityPullNumbers and
+ * sweepRepoBacklogConvergence (#orb-retry-storm, backlog-convergence half): both sweeps re-select on the
+ * identical lastPublishedSurfaceSha-mismatch signal, so they share one attempt budget per SHA rather than each
+ * independently hammering the same stuck PR.
+ */
+async function isRegateRepairExhausted(env: Env, repoFullName: string, pr: Pick<PullRequestRecord, "number" | "headSha">): Promise<boolean> {
+  const headSha = pr.headSha!;
+  const sinceIso = new Date(Date.now() - REGATE_REPAIR_ATTEMPT_LOOKBACK_MS).toISOString();
+  const targetKey = regateRepairTargetKey(repoFullName, pr.number, headSha);
+  const attempts = await countRecentAuditEventsForActorAndTarget(env, "loopover", REGATE_REPAIR_ATTEMPT_EVENT_TYPE, targetKey, sinceIso);
+  if (attempts < REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA) return false;
+  const alreadyFlagged = await countRecentAuditEventsForActorAndTarget(env, "loopover", REGATE_REPAIR_EXHAUSTED_EVENT_TYPE, targetKey, sinceIso);
+  if (alreadyFlagged === 0) {
+    await recordAuditEvent(env, {
+      eventType: REGATE_REPAIR_EXHAUSTED_EVENT_TYPE,
+      actor: "loopover",
+      targetKey,
+      outcome: "denied",
+      detail: `re-gate repair exhausted after ${attempts} attempt(s) for the same head SHA; falling back to ordinary staleness cadence`,
+      metadata: { repoFullName, prNumber: pr.number, headSha, attempts },
+    });
+    // level:"error" is deliberate, not a code failure: this line only fires once the cap above already
+    // stopped the wasteful repair loop, so its OWN existence is the operator-visible signal (via the
+    // structured log → Sentry forwarder, forwardStructuredLogToSentry) that a PR kept failing repair for the
+    // same head SHA — the same "surface an anomaly at error level" convention selfhost_ai_provider_failed /
+    // selfhost_ai_providers_exhausted already use in src/selfhost/ai.ts.
+    console.error(
+      JSON.stringify({
+        level: "error",
+        event: "regate_repair_exhausted",
+        repo: repoFullName,
+        pullNumber: pr.number,
+        headSha,
+        attempts,
+      }),
+    );
+  }
+  return true;
+}
+
 async function surfaceRepairPriorityPullNumbers(
   env: Env,
   repoFullName: string,
@@ -1080,53 +1123,12 @@ async function surfaceRepairPriorityPullNumbers(
       }),
     );
   }
-  const sinceIso = new Date(Date.now() - REGATE_REPAIR_ATTEMPT_LOOKBACK_MS).toISOString();
   await Promise.all(
     [...priorityPullNumbers].map(async (prNumber) => {
       const pr = pulls.find((candidate) => candidate.number === prNumber);
       /* v8 ignore next -- priorityPullNumbers is only ever populated (both loops above) from a `pr` in `pulls` that already had a truthy headSha, so this lookup always succeeds with one; the guard only satisfies Array#find's `| undefined` return type. */
       if (!pr?.headSha) return;
-      const targetKey = regateRepairTargetKey(repoFullName, pr.number, pr.headSha);
-      const attempts = await countRecentAuditEventsForActorAndTarget(
-        env,
-        "loopover",
-        REGATE_REPAIR_ATTEMPT_EVENT_TYPE,
-        targetKey,
-        sinceIso,
-      );
-      if (attempts < REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA) return;
-      priorityPullNumbers.delete(prNumber);
-      const alreadyFlagged = await countRecentAuditEventsForActorAndTarget(
-        env,
-        "loopover",
-        REGATE_REPAIR_EXHAUSTED_EVENT_TYPE,
-        targetKey,
-        sinceIso,
-      );
-      if (alreadyFlagged > 0) return;
-      await recordAuditEvent(env, {
-        eventType: REGATE_REPAIR_EXHAUSTED_EVENT_TYPE,
-        actor: "loopover",
-        targetKey,
-        outcome: "denied",
-        detail: `re-gate repair exhausted after ${attempts} attempt(s) for the same head SHA; falling back to ordinary staleness cadence`,
-        metadata: { repoFullName, prNumber: pr.number, headSha: pr.headSha, attempts },
-      });
-      // level:"error" is deliberate, not a code failure: this line only fires once the cap above already
-      // stopped the wasteful repair loop, so its OWN existence is the operator-visible signal (via the
-      // structured log → Sentry forwarder, forwardStructuredLogToSentry) that a PR kept failing repair for the
-      // same head SHA — the same "surface an anomaly at error level" convention selfhost_ai_provider_failed /
-      // selfhost_ai_providers_exhausted already use in src/selfhost/ai.ts.
-      console.error(
-        JSON.stringify({
-          level: "error",
-          event: "regate_repair_exhausted",
-          repo: repoFullName,
-          pullNumber: pr.number,
-          headSha: pr.headSha,
-          attempts,
-        }),
-      );
+      if (await isRegateRepairExhausted(env, repoFullName, pr)) priorityPullNumbers.delete(prNumber);
     }),
   );
   return [...priorityPullNumbers];
@@ -1724,7 +1726,15 @@ export async function sweepRepoBacklogConvergence(
   const sweepInstallationId = repo?.installationId ?? null;
   if (sweepInstallationId == null) return;
   const openPullRequests = await listOpenPullRequests(env, repoFullName);
-  const candidates = selectBacklogConvergenceCandidates({ pulls: openPullRequests });
+  const allCandidates = selectBacklogConvergenceCandidates({ pulls: openPullRequests });
+  // #orb-retry-storm (backlog-convergence half): needsSurfaceConvergence re-fires on the exact same
+  // lastPublishedSurfaceSha-mismatch signal as the main sweep's outage-repair priority path, but this sweeper
+  // had no memory of prior attempts at all -- a PR whose gate-check finalize kept failing silently got a fresh
+  // full re-review dispatched every ~30 minutes indefinitely. Share the same per-SHA attempt budget as the main
+  // sweep (isRegateRepairExhausted) rather than adding an independent cap, since both sweeps competing for the
+  // same stuck PR would otherwise double the wasted spend the cap exists to prevent.
+  const exhaustedFlags = await Promise.all(allCandidates.map((pr) => isRegateRepairExhausted(env, repoFullName, pr)));
+  const candidates = allCandidates.filter((_pr, index) => !exhaustedFlags[index]);
   if (candidates.length === 0) return;
   // Stamp the backlog-convergence draining marker for EVERY candidate NOW, at dispatch — not in the downstream
   // per-PR job (#4502, mirrors #audit-sweep-dispatch-stamp). This makes getLatestBacklogConvergenceRegatedAt
@@ -1752,6 +1762,12 @@ export async function sweepRepoBacklogConvergence(
         repoFullName,
         prNumber: pr.number,
         installationId: sweepInstallationId,
+        // #orb-retry-storm: selectBacklogConvergenceCandidates only returns PRs needsSurfaceConvergence already
+        // confirmed have a truthy headSha, so this is unconditional (unlike the main sweep's priority-repair
+        // dispatch, which mixes repair and ordinary candidates). Passing it lets regatePullRequest record the
+        // attempt at execution time against the SAME shared per-SHA budget isRegateRepairExhausted checked
+        // above -- jobs deferred or dropped before they run still don't count against the cap.
+        repairHeadSha: pr.headSha!,
         ...(pr.createdAt ? { prCreatedAt: pr.createdAt } : {}),
       };
       const delaySeconds = Math.min(index * 10, 600);

test/unit/queue-2.test.ts

@@ -1933,6 +1933,84 @@ describe("queue processors", () => {
     }
   }, 60_000);
 
+  it("REGRESSION (#orb-retry-storm, backlog-convergence half): a PR whose head SHA already exhausted the shared repair-attempt budget is not re-dispatched by the backlog-convergence sweep", async () => {
+    const sent: import("../../src/types").JobMessage[] = [];
+    const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue });
+    await upsertInstallation(env, { action: "created", installation: { id: 9412, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } });
+    await upsertRepositoryFromGitHub(env, { name: "agent-repo4", full_name: "owner/agent-repo4", private: false, owner: { login: "owner" } }, 9412);
+    await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo4", autonomy: { merge: "auto" }, reviewCheckMode: "required", checkRunMode: "off", commentMode: "off", publicSurface: "off" });
+    // Never marked surface-published for this head -- needsSurfaceConvergence is true, a genuine backlog-convergence candidate.
+    await upsertPullRequestFromGitHub(env, "owner/agent-repo4", { number: 1, title: "Stuck convergence", state: "open", user: { login: "c" }, head: { sha: "stuck-sha" }, labels: [], body: "" });
+    const targetKey = "owner/agent-repo4#1#stuck-sha";
+    // Pre-seed the SAME shared budget the main sweep's repair path charges against -- this is the whole point of
+    // sharing isRegateRepairExhausted rather than giving backlog-convergence its own independent counter.
+    for (let i = 0; i < 5; i += 1) {
+      await repositoriesModule.recordAuditEvent(env, {
+        eventType: "agent.sweep.regate.repair_attempt",
+        actor: "loopover",
+        targetKey,
+        outcome: "queued",
+        detail: "prior attempt",
+        metadata: {},
+      });
+    }
+    vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z"));
+    const errors = vi.spyOn(console, "error").mockImplementation(() => undefined);
+
+    try {
+      await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "test", repoFullName: "owner/agent-repo4" });
+
+      const fanned = sent.filter((job): job is Extract<import("../../src/types").JobMessage, { type: "agent-regate-pr" }> => job.type === "agent-regate-pr");
+      expect(fanned.every((job) => job.deliveryId !== "backlog-convergence:owner/agent-repo4#1")).toBe(true);
+      const exhausted = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?")
+        .bind("agent.sweep.regate.repair_exhausted", targetKey)
+        .first<{ n: number }>();
+      expect(exhausted?.n).toBe(1);
+      const exhaustedLogs = errors.mock.calls.filter(([line]) => typeof line === "string" && line.includes("regate_repair_exhausted"));
+      expect(exhaustedLogs).toHaveLength(1);
+    } finally {
+      errors.mockRestore();
+    }
+  }, 60_000);
+
+  it("a fresh backlog-convergence candidate under the cap is dispatched with repairHeadSha set, and executing it charges the SAME shared attempt budget the main sweep's repair path reads", async () => {
+    const sent: import("../../src/types").JobMessage[] = [];
+    const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue });
+    await upsertInstallation(env, { action: "created", installation: { id: 9413, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } });
+    await upsertRepositoryFromGitHub(env, { name: "agent-repo5", full_name: "owner/agent-repo5", private: false, owner: { login: "owner" } }, 9413);
+    await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo5", autonomy: { merge: "auto" }, aiReviewMode: "off", checkRunMode: "off", commentMode: "off", publicSurface: "off" });
+    await upsertPullRequestFromGitHub(env, "owner/agent-repo5", { number: 2, title: "Fresh convergence", state: "open", user: { login: "c" }, head: { sha: "fresh-sha" }, base: { ref: "main" }, labels: [], body: "" });
+    const targetKey = "owner/agent-repo5#2#fresh-sha";
+    vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z"));
+    vi.stubGlobal("fetch", async (input: RequestInfo | URL) => {
+      const url = input.toString();
+      if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" });
+      if (/\/pulls\/2(?:\?|$)/.test(url)) return Response.json({ number: 2, title: "Fresh convergence", state: "open", user: { login: "c" }, head: { sha: "fresh-sha" }, mergeable_state: "clean", labels: [], body: "" });
+      if (url.includes("/pulls/2/files")) return Response.json([{ filename: "src/a.ts", status: "modified", additions: 1, deletions: 0, changes: 1, patch: "@@\n+export const ok = true;" }]);
+      return Response.json({});
+    });
+
+    await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "test", repoFullName: "owner/agent-repo5" });
+
+    const fanned = sent.filter((job): job is Extract<import("../../src/types").JobMessage, { type: "agent-regate-pr" }> => job.type === "agent-regate-pr");
+    const dispatched = fanned.find((job) => job.deliveryId === "backlog-convergence:owner/agent-repo5#2");
+    expect(dispatched).toMatchObject({ repoFullName: "owner/agent-repo5", prNumber: 2, repairHeadSha: "fresh-sha" });
+
+    const attemptsBefore = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?")
+      .bind("agent.sweep.regate.repair_attempt", targetKey)
+      .first<{ n: number }>();
+    expect(attemptsBefore?.n).toBe(0);
+
+    await processJob(env, dispatched!);
+
+    // The SAME event type/target key the main sweep's repair path reads -- proves the two sweeps share one
+    // budget for this SHA instead of each independently retrying it.
+    const attemptsAfter = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?")
+      .bind("agent.sweep.regate.repair_attempt", targetKey)
+      .first<{ n: number }>();
+    expect(attemptsAfter?.n).toBe(1);
+  }, 60_000);
+
   it("REGRESSION (#5385-sentry, GITTENSORY-1E gate-finding): a retryable GitHub rate-limit error surfacing from real post-readiness review work STILL records the repair attempt before propagating for the queue's own retry", async () => {
     // LoopOver review finding on PR #5482: the original fix recorded the attempt AFTER reReviewStoredPullRequest
     // returns, so a retryable error thrown from a genuinely-executed (post-readiness) pass never got charged --