fix(queue): dedupe concurrent AI reviews for the same PR head (#2429)

* fix(queue): dedupe concurrent AI reviews for the same PR head

A webhook pass and an agent-regate-pr sweep pass can independently miss
the AI review cache for the same (repo, PR, head SHA, mode) and each
fire a real LLM call, potentially landing different verdicts at the
same head. Add a per-key advisory lock (mirroring the existing
per-PR maintenance lock) around the review call; a pass that loses the
race returns the existing ai_review_inconclusive shape so the gate
holds for human review instead of racing an independent verdict.

* fix(queue): close the concurrent-claim race in the transient-lock get/set fallback

claimAiReviewLock's fallback path (used when the cache adapter has no
atomic claim()) was a plain get-then-set pair: two concurrent callers
can both observe an absent key before either writes, and both believe
they claimed the lock -- defeating the dedupe guarantee for exactly
the double-LLM-call race this lock exists to prevent. The same latent
gap already existed in the sibling claimAgentMaintenanceLock fallback.

Extract a shared claimTransientLock helper used by both locks. Its
fallback now writes a token unique to the attempt, then reads the key
back -- since a correctly-behaved key-value store serializes writes to
a single key, only the caller whose token survives the final read
actually won; every other concurrent caller reads a different (later)
token and correctly backs off. Still fails open on a missing cache or
any read/write error, matching the existing defense-in-depth design.

* fix(queue): stop pretending the transient-lock fallback can serialize without atomic claim()

The prior fallback (write a unique token, then re-read to verify) does
not close the race: caller A can write its token, read it back, and
return true entirely before caller B's later write/read also returns
true -- both callers can still "win" under real concurrent
interleavings, which defeats the whole point of the lock.

There is no way to build real mutual exclusion out of separate
get/set calls without an atomic primitive. Rather than keep pretending
to serialize, claimTransientLock now requires the cache adapter's
native claim() for any exclusivity at all; without it, every caller
proceeds (fail open), matching this lock family's existing
defense-in-depth philosophy. Self-host's Redis-backed cache always
implements claim(), so this is a documented limitation for a
hypothetical future adapter, not a live production gap.

Author: JSONbored

src/queue/processors.ts

@@ -2446,6 +2446,34 @@ async function ciHeadShaResolutionCoalesced(
   );
 }
 
+/**
+ * Best-effort exclusive claim against the self-host transient cache, shared by every per-PR/per-review advisory
+ * lock below. Requires the store's native atomic claim() (Redis SET NX) to provide any real exclusivity — it is
+ * the only way to close the race between two concurrent callers each observing an absent key. A plain
+ * get-then-set pair CANNOT close that race in general, even with an extra write-then-verify re-read: caller A
+ * can write its own token, read it straight back, and return true entirely BEFORE caller B's later write/read
+ * also completes and also returns true — both callers "win" (#confirmed-bug). Rather than pretend to serialize
+ * via a check that silently fails under exactly the concurrent load this lock exists to guard against, an
+ * adapter without claim() gets NO exclusivity from this helper: every caller proceeds. This is honest about the
+ * limitation rather than a false guarantee, and costs nothing in practice — self-host's Redis-backed cache (the
+ * only cache adapter this codebase ships) always implements claim(), so this is a documented limitation for a
+ * hypothetical future adapter, not a live gap. A missing cache or a thrown claim() also fails OPEN (returns
+ * true) — every lock built on this helper is defense-in-depth, never the primary safety gate, and must never
+ * itself block real work from running.
+ */
+async function claimTransientLock(
+  env: Env,
+  key: string,
+  ttlSeconds: number,
+): Promise<boolean> {
+  if (!env.SELFHOST_TRANSIENT_CACHE?.claim) return true; // no atomic primitive — nothing to serialize against.
+  try {
+    return await env.SELFHOST_TRANSIENT_CACHE.claim(key, "1", ttlSeconds);
+  } catch {
+    return true; // fail open — see the doc comment above.
+  }
+}
+
 // Per-PR advisory lock around maybeRunAgentMaintenance's plan-and-execute critical section (#2129). The TTL is a
 // crash-safety backstop only — the normal path releases explicitly in a finally block within a few seconds — so
 // it is sized well above any realistic pass duration (matches CI_COALESCE_WINDOW_SECONDS, an already-vetted
@@ -2466,25 +2494,11 @@ export async function claimAgentMaintenanceLock(
   repoFullName: string,
   prNumber: number,
 ): Promise<boolean> {
-  const key = agentMaintenanceLockKey(repoFullName, prNumber);
-  // Atomic claim (#2129): a get-then-set pair has a window between the read and the write where two concurrent
-  // passes for the SAME PR can both observe an absent key and both claim it, defeating the serializer entirely.
-  // env.SELFHOST_TRANSIENT_CACHE.claim performs the check-and-set as one operation (Redis SET NX server-side),
-  // closing that window. Falls back to the non-atomic get/set pair only for a cache adapter that hasn't
-  // implemented claim yet — strictly no worse than this function's prior behavior.
-  if (env.SELFHOST_TRANSIENT_CACHE?.claim) {
-    try {
-      return await env.SELFHOST_TRANSIENT_CACHE.claim(key, "1", AGENT_MAINTENANCE_LOCK_TTL_SECONDS);
-    } catch {
-      return true; // fail open — see the doc comment above.
-    }
-  }
-  // getTransientKey/putTransientKey already fail open internally (a missing cache or a thrown read/write error
-  // both resolve rather than throw), so this never needs its own try/catch — a cache fault surfaces here as
-  // "no lock held", which correctly falls through to claiming it.
-  if (await getTransientKey(env, key)) return false; // another pass is already in-flight for this PR
-  await putTransientKey(env, key, "1", AGENT_MAINTENANCE_LOCK_TTL_SECONDS);
-  return true;
+  return claimTransientLock(
+    env,
+    agentMaintenanceLockKey(repoFullName, prNumber),
+    AGENT_MAINTENANCE_LOCK_TTL_SECONDS,
+  );
 }
 
 /** Best-effort release, called from a finally block so the lock frees promptly instead of waiting out the TTL. */
@@ -2502,6 +2516,54 @@ export async function releaseAgentMaintenanceLock(
   }
 }
 
+// Per-(repo, PR, head SHA) advisory lock around runAiReviewForAdvisory's expensive grounding/RAG/enrichment/LLM
+// section (#confirmed-bug: a webhook pass and an agent-regate-pr sweep pass can independently reach this same
+// code for the SAME PR at the SAME head SHA, both miss the cache, and both fire a real LLM call — which can
+// return DIFFERENT verdicts). The TTL is a crash-safety backstop only (see AI_REVIEW_LOCK_TTL_SECONDS below), not
+// a throughput bound — same philosophy as AGENT_MAINTENANCE_LOCK_TTL_SECONDS (#2129/#2368).
+const AI_REVIEW_LOCK_TTL_SECONDS = 1_800; // 30 minutes — see justification below.
+
+function aiReviewLockKey(repoFullName: string, prNumber: number, headSha: string, mode: string): string {
+  return `ai-review-lock:${repoFullName.toLowerCase()}#${prNumber}@${headSha.toLowerCase()}:${mode}`;
+}
+
+/**
+ * Claim the per-(repo, PR, head SHA, mode) advisory lock before the expensive grounding/RAG/enrichment/LLM
+ * section of runAiReviewForAdvisory. Returns false when another pass already holds it for this exact head (the
+ * caller must treat this as "another pass is already reviewing this head" and return the inconclusive-hold shape
+ * below — the next webhook/sweep tick, or the pass that IS running, is the backstop that populates the cache).
+ * A missing cache or cache hiccup fails OPEN (returns true — the lock is defense-in-depth, never the primary
+ * safety gate, and must never itself block a real review from running).
+ */
+export async function claimAiReviewLock(
+  env: Env,
+  repoFullName: string,
+  prNumber: number,
+  headSha: string,
+  mode: string,
+): Promise<boolean> {
+  return claimTransientLock(
+    env,
+    aiReviewLockKey(repoFullName, prNumber, headSha, mode),
+    AI_REVIEW_LOCK_TTL_SECONDS,
+  );
+}
+
+/** Best-effort release, called from a finally block so the lock frees promptly instead of waiting out the TTL. */
+export async function releaseAiReviewLock(
+  env: Env,
+  repoFullName: string,
+  prNumber: number,
+  headSha: string,
+  mode: string,
+): Promise<void> {
+  try {
+    await env.SELFHOST_TRANSIENT_CACHE?.del?.(aiReviewLockKey(repoFullName, prNumber, headSha, mode));
+  } catch {
+    // best-effort; the TTL is the backstop if release fails
+  }
+}
+
 /** Read the CI head SHA off a `check_suite`/`check_run` `completed` payload (the event node carries `head_sha`;
  *  `check_run` also nests it under `check_suite.head_sha`). Returns "" when absent. The payload type doesn't model
  *  these events, so we narrow off `Record<string, unknown>` the same way the `pull_requests[]` read does. */
@@ -4677,6 +4739,39 @@ export async function runAiReviewForAdvisory(
     }))
   )
     return undefined;
+  // Per-(repo, PR, head SHA, mode) advisory lock (#confirmed-bug, mirrors #2129/#2368's claimAgentMaintenanceLock):
+  // a webhook pass and an agent-regate-pr sweep pass can independently reach this point for the SAME PR at the
+  // SAME head, both miss the cache (neither has written yet), and both fire a real, wasteful LLM call that can
+  // return different verdicts. Claim before the expensive section below; a pass that loses the race returns the
+  // same inconclusive-hold shape the "AI produced no usable verdict" path already returns, so the gate is held
+  // (neutral) for a human rather than either pass's independently-decided verdict racing the other's cache write.
+  if (
+    !(await claimAiReviewLock(
+      env,
+      args.repoFullName,
+      args.pr.number,
+      args.advisory.headSha,
+      args.settings.aiReviewMode,
+    ))
+  ) {
+    const findings: AdvisoryFinding[] = [
+      {
+        code: "ai_review_inconclusive",
+        severity: "warning",
+        title: "AI review already in progress for this PR head",
+        detail: "Another Gittensory pass is already running the AI review for this exact PR head. This pass is skipping to avoid a duplicate LLM call.",
+        action: "The gate is held for a human reviewer rather than passed automatically; it re-evaluates once the in-flight review completes or on the next update.",
+      },
+    ];
+    args.advisory.findings.push(...findings);
+    return {
+      notes: "AI review is already running for this PR head in another Gittensory pass. Gittensory is holding this PR for manual review until that pass completes.",
+      reviewerCount: 0,
+      inlineFindings: [],
+      findings,
+      cacheable: false,
+    };
+  }
   try {
     // BYOK: decrypt the maintainer's provider key only for confirmed contributors when opted in. Falls back to free Workers AI when
     // no key is configured or the encryption secret is unavailable (getDecryptedRepositoryAiKey → null).
@@ -4968,6 +5063,14 @@ export async function runAiReviewForAdvisory(
       head_sha: args.advisory.headSha,
     });
     return undefined;
+  } finally {
+    await releaseAiReviewLock(
+      env,
+      args.repoFullName,
+      args.pr.number,
+      args.advisory.headSha,
+      args.settings.aiReviewMode,
+    );
   }
 }
 

test/unit/ai-review-advisory.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
-import { buildAiReviewDiff, runAiReviewForAdvisory, shouldStartAiReviewForAdvisory } from "../../src/queue/processors";
+import { buildAiReviewDiff, claimAiReviewLock, runAiReviewForAdvisory, shouldStartAiReviewForAdvisory } from "../../src/queue/processors";
 import { BEST_REVIEW_MODELS, INCOHERENT_DIFF_ASSESSMENT } from "../../src/services/ai-review";
 import * as sentryModule from "../../src/selfhost/sentry";
 import { upsertRepositoryAiKey } from "../../src/db/repositories";
@@ -524,6 +524,38 @@ describe("runAiReviewForAdvisory", () => {
     expect(adv.findings.map((f) => f.code)).toEqual(["ai_review_inconclusive"]);
   });
 
+  it("#confirmed-bug: defers to an already-held AI review lock and never invokes the AI when another pass is in-flight for this exact head", async () => {
+    const adv = advisory();
+    let aiCalls = 0;
+    const env = aiEnv(async () => {
+      aiCalls += 1;
+      return { response: notesOnlyJson() };
+    });
+    // Simulate a webhook pass already in-flight for this exact (repo, PR, head, mode) tuple — the caller under
+    // test (a sweep-shaped pass, say) must defer instead of racing it with a second, independently-decided
+    // LLM call.
+    expect(await claimAiReviewLock(env, "acme/widgets", 3, "sha3", "block")).toBe(true);
+
+    const result = await runAiReviewForAdvisory(env, {
+      settings: { aiReviewMode: "block" } as RepositorySettings,
+      advisory: adv,
+      repoFullName: "acme/widgets",
+      pr,
+      author: "alice",
+      confirmedContributor: true,
+    });
+
+    expect(aiCalls).toBe(0); // the AI mock was never invoked — the lock short-circuited before the LLM call
+    expect(result).toMatchObject({
+      reviewerCount: 0,
+      inlineFindings: [],
+      cacheable: false,
+      findings: [expect.objectContaining({ code: "ai_review_inconclusive" })],
+    });
+    expect(result?.notes).toContain("AI review is already running for this PR head in another Gittensory pass");
+    expect(adv.findings.map((f) => f.code)).toEqual(["ai_review_inconclusive"]);
+  });
+
   it("withholds unstructured AI text while holding the PR for manual review", async () => {
     const adv = advisory();
     const result = await runAiReviewForAdvisory(aiEnv(async () => ({ response: "Looks coherent, but please verify the new cache branch before merging." })), {

test/unit/gate-check-policy.test.ts

@@ -131,6 +131,46 @@ describe("AI fail-closed hold (#ai-fail-closed)", () => {
     expect(result.blockers.map((blocker) => blocker.code)).toContain("secret_leak");
   });
 
+  it("holds the gate NEUTRAL (never a failure-close) when the AI review lock is held by another in-flight pass (#confirmed-bug)", () => {
+    // Same code, different finding text — the lock-contention finding constructed by runAiReviewForAdvisory's
+    // new claim-failure branch. advisory.ts only keys on `code`, so this proves the mechanism end-to-end for
+    // the new finding shape without needing to touch advisory.ts.
+    const adv: Advisory = {
+      ...missingIssueAdvisory(),
+      findings: [
+        {
+          code: "ai_review_inconclusive",
+          title: "AI review already in progress for this PR head",
+          severity: "warning",
+          detail: "Another Gittensory pass is already running the AI review for this exact PR head. This pass is skipping to avoid a duplicate LLM call.",
+          action: "The gate is held for a human reviewer rather than passed automatically; it re-evaluates once the in-flight review completes or on the next update.",
+        },
+      ],
+    };
+    const result = evaluateGateCheck(adv, gateCheckPolicy(settings(), null, true));
+    expect(result.conclusion).toBe("neutral");
+    expect(result.blockers).toEqual([]);
+  });
+
+  it("a deterministic hard blocker (secret_leak) still FAILS even when the AI review is held by lock contention (#confirmed-bug)", () => {
+    const adv: Advisory = {
+      ...missingIssueAdvisory(),
+      findings: [
+        { code: "secret_leak", title: "Possible leaked secret", severity: "critical", detail: "a committed token", action: "remove and rotate it" },
+        {
+          code: "ai_review_inconclusive",
+          title: "AI review already in progress for this PR head",
+          severity: "warning",
+          detail: "Another Gittensory pass is already running the AI review for this exact PR head. This pass is skipping to avoid a duplicate LLM call.",
+          action: "The gate is held for a human reviewer rather than passed automatically; it re-evaluates once the in-flight review completes or on the next update.",
+        },
+      ],
+    };
+    const result = evaluateGateCheck(adv, gateCheckPolicy(settings(), null, true));
+    expect(result.conclusion).toBe("failure");
+    expect(result.blockers.map((blocker) => blocker.code)).toContain("secret_leak");
+  });
+
   it("an enforced pre-merge check (pre_merge_check_required) hard-blocks; the advisory variant never does (#review-pre-merge-checks)", () => {
     const enforced: Advisory = {
       ...missingIssueAdvisory(),