fix(queue): serialize concurrent webhook/sweep passes for the same PR

src/env.d.ts

@@ -28,6 +28,12 @@ declare global {
       get(key: string): Promise<string | null>;
       set(key: string, value: string, ttlSeconds: number): Promise<void>;
       del?(key: string): Promise<void>;
+      /** Atomic "set only if absent": returns true when this call newly claimed the key, false when it was
+       *  already held by someone else. Unlike a get-then-set pair, there is no window where two concurrent
+       *  callers can both observe an absent key and both claim it — the store (e.g. Redis SET NX) performs the
+       *  check-and-set as one operation. Optional so a cache adapter that hasn't implemented it yet still
+       *  type-checks; callers fall back to the non-atomic get/set pair when absent (#2129). */
+      claim?(key: string, value: string, ttlSeconds: number): Promise<boolean>;
     };
     /** TODO (convergence follow-up): a per-PR LOCK Durable Object (`SubmissionLock` mutex) is a separate,
      *  more-involved sub-task — it needs the ported DO class + its own migration tag, not just a binding here.

src/queue/processors.ts

@@ -1694,6 +1694,48 @@ async function maybeRunAgentMaintenance(
   if (pr.isDraft) return;
   if (!gate) return;
 
+  // Per-PR mutual exclusion (#2129): a webhook re-review and a sweep-driven agent-regate-pr job use different
+  // coalesce-key shapes (jobCoalesceKey never matches one against the other) and QUEUE_CONCURRENCY explicitly
+  // overlaps I/O-bound jobs, so two passes for the SAME PR can both reach this point concurrently, each with its
+  // own independently-timed live CI/mergeable/reviewDecision read. If those reads disagree, both could plan and
+  // execute DIFFERENT actions for the same PR. Claim a short-TTL advisory lock before the plan-and-execute
+  // critical section (extracted below so the try/finally doesn't force-reindent that whole block); a pass that
+  // loses the race defers cleanly — the next webhook/sweep tick is the backstop. Lightweight stand-in for the
+  // per-PR SubmissionLock Durable Object noted as a longer-term TODO in env.d.ts.
+  if (!(await claimAgentMaintenanceLock(env, repoFullName, pr.number))) return;
+  try {
+    await runAgentMaintenancePlanAndExecute(env, {
+      installationId,
+      repoFullName,
+      repo: args.repo,
+      pr,
+      settings,
+      otherOpenPullRequests,
+      gate,
+      liveFacts: args.liveFacts,
+    });
+  } finally {
+    await releaseAgentMaintenanceLock(env, repoFullName, pr.number);
+  }
+}
+
+/** The plan-and-execute critical section of {@link maybeRunAgentMaintenance}, extracted so the caller's
+ *  per-PR lock (#2129) wraps it in a try/finally without reindenting this whole block. */
+async function runAgentMaintenancePlanAndExecute(
+  env: Env,
+  args: {
+    installationId: number;
+    repoFullName: string;
+    repo: Awaited<ReturnType<typeof getRepository>>;
+    pr: PullRequestRecord;
+    settings: RepositorySettings;
+    otherOpenPullRequests: PullRequestRecord[];
+    gate: ReturnType<typeof evaluateGateCheck>;
+    liveFacts: LiveGithubFacts;
+  },
+): Promise<void> {
+  const { installationId, repoFullName, pr, settings, otherOpenPullRequests, gate } = args;
+
   // Convergence safety: feed the planner the PR's changed paths + the repo's hard-guardrail globs so guarded
   // paths force manual review, and flag owner-authored PRs so they are never auto-closed (standing rule).
   // FIX B: resolve files via the shared resolver so an EMPTY stored list (the maintenance ran before the
@@ -2307,6 +2349,62 @@ async function ciHeadShaResolutionCoalesced(
   );
 }
 
+// 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
+// value for a comparable-scale operation in this file), not to bound throughput.
+const AGENT_MAINTENANCE_LOCK_TTL_SECONDS = 60;
+
+function agentMaintenanceLockKey(repoFullName: string, prNumber: number): string {
+  return `agent-maintenance-lock:${repoFullName.toLowerCase()}#${prNumber}`;
+}
+
+/**
+ * Claim the per-PR advisory lock. Returns false when another pass already holds it (caller must skip this pass
+ * — the next webhook/sweep tick is the backstop). A missing cache or cache hiccup fails OPEN (returns true —
+ * the lock is a defense-in-depth serializer, not the primary safety gate, and must never itself block actuation).
+ */
+export async function claimAgentMaintenanceLock(
+  env: Env,
+  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;
+}
+
+/** Best-effort release, called from a finally block so the lock frees promptly instead of waiting out the TTL. */
+export async function releaseAgentMaintenanceLock(
+  env: Env,
+  repoFullName: string,
+  prNumber: number,
+): Promise<void> {
+  try {
+    await env.SELFHOST_TRANSIENT_CACHE?.del?.(
+      agentMaintenanceLockKey(repoFullName, prNumber),
+    );
+  } 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. */

src/selfhost/redis-cache.ts

@@ -16,6 +16,13 @@ export function createRedisCache(redis: Redis) {
     async del(key: string): Promise<void> {
       await redis.del(key);
     },
+    // Redis performs the existence check and the write as a single atomic command server-side (SET ... NX), so
+    // two concurrent callers racing on the same key can never both receive "OK" -- unlike a get-then-set pair,
+    // which has a window between the read and the write where both callers can observe an absent key.
+    async claim(key: string, value: string, ttlSeconds: number): Promise<boolean> {
+      const result = await redis.set(key, value, "EX", ttlSeconds, "NX");
+      return result === "OK";
+    },
   };
 }
 

test/helpers/d1.ts

@@ -89,6 +89,15 @@ export function createTestEnv(overrides: Partial<Env> = {}): Env {
       async del(key: string) {
         transientCache.delete(key);
       },
+      // Mirrors createRedisCache's atomic claim (#2129): the check-and-set below has no `await` between the
+      // `has` read and the `set` write, so it completes synchronously within one microtask — a concurrent
+      // caller can never observe the key as absent partway through another caller's claim, matching Redis's
+      // SET NX server-side atomicity.
+      async claim(key: string, value: string) {
+        if (transientCache.has(key)) return false;
+        transientCache.set(key, value);
+        return true;
+      },
     },
     // Per-repo review allowlist: default to the test repos so flag-ON wiring tests activate the
     // gated review features. Override to "" to assert the dormant (no-repo) default.

test/unit/queue.test.ts

@@ -44,7 +44,7 @@ import {
   upsertRepositoryFromGitHub,
   putCachedAiReview,
 } from "../../src/db/repositories";
-import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, contributorEvidenceBatchSize, processJob } from "../../src/queue/processors";
+import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAgentMaintenanceLock, contributorEvidenceBatchSize, processJob, releaseAgentMaintenanceLock } from "../../src/queue/processors";
 import { aiReviewCacheInputFingerprint } from "../../src/review/ai-review-cache-input";
 import { upsertRepoFocusManifest } from "../../src/signals/focus-manifest-loader";
 import { normalizeRegistryPayload } from "../../src/registry/normalize";
@@ -2922,6 +2922,132 @@ describe("queue processors", () => {
     expect(denied?.n).toBe(1);
   });
 
+  it("claimAgentMaintenanceLock claims when free, denies when held (per-PR, not per-repo), and release frees it again (#2129)", async () => {
+    const env = createTestEnv({});
+    // First claim for this PR succeeds — no prior pass in-flight.
+    expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
+    // A second, concurrent pass for the SAME PR (regardless of what triggered it — webhook or sweep) is denied
+    // while the first is still in-flight — exactly the race #2129 describes, since job-coalesce keys never
+    // match across trigger shapes but this lock is keyed purely on repo+PR.
+    expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(false);
+    // A DIFFERENT PR in the same repo is unaffected — the lock is per-PR, not a repo-wide serializer.
+    expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 8)).toBe(true);
+    // Release (the finally block's job) frees the PR — a subsequent pass can claim it again.
+    await releaseAgentMaintenanceLock(env, "owner/agent-repo", 7);
+    expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
+  });
+
+  it("claimAgentMaintenanceLock fails OPEN on a broken transient cache — never itself blocks actuation (#2129)", async () => {
+    const env = createTestEnv({
+      SELFHOST_TRANSIENT_CACHE: {
+        get: async () => { throw new Error("cache read error"); },
+        set: async () => { throw new Error("cache write error"); },
+        del: async () => { throw new Error("cache delete error"); },
+      },
+    });
+    expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
+    await expect(releaseAgentMaintenanceLock(env, "owner/agent-repo", 7)).resolves.toBeUndefined();
+  });
+
+  it("claimAgentMaintenanceLock fails OPEN when the atomic claim primitive itself throws (#2368)", async () => {
+    const env = createTestEnv({
+      SELFHOST_TRANSIENT_CACHE: {
+        get: async () => null,
+        set: async () => undefined,
+        claim: async () => { throw new Error("redis unavailable"); },
+      },
+    });
+    expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
+  });
+
+  it("REGRESSION (#2368): claimAgentMaintenanceLock uses an atomic check-and-set, so two genuinely concurrent claims for the SAME PR can never both succeed", async () => {
+    // #2368: a get-then-set pair has a window between the read and the write where two concurrent callers can
+    // both observe an absent key and both claim it — exactly what the per-PR lock exists to prevent. This test
+    // races two claims for the same PR via Promise.all (both kick off before either resolves) against the
+    // default test cache's claim(), which mirrors createRedisCache's atomic SET NX: the check-and-set happens
+    // with no `await` boundary in between, so it is impossible for both callers to see "unclaimed".
+    const env = createTestEnv({});
+    const [first, second] = await Promise.all([
+      claimAgentMaintenanceLock(env, "owner/agent-repo", 7),
+      claimAgentMaintenanceLock(env, "owner/agent-repo", 7),
+    ]);
+    expect([first, second].filter(Boolean)).toHaveLength(1);
+  });
+
+  it("REGRESSION (#2368): claimAgentMaintenanceLock calls the atomic claim primitive, not a separate get+set pair, when the cache supports it", async () => {
+    const calls: string[] = [];
+    const env = createTestEnv({
+      SELFHOST_TRANSIENT_CACHE: {
+        get: async () => { calls.push("get"); return null; },
+        set: async () => { calls.push("set"); },
+        claim: async () => { calls.push("claim"); return true; },
+      },
+    });
+    expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
+    expect(calls).toEqual(["claim"]); // never falls through to the racy get/set pair when claim is available
+  });
+
+  it("claimAgentMaintenanceLock falls back to the get/set pair and still denies a held key when the cache has no claim() (#2368)", async () => {
+    // A cache adapter that hasn't implemented the atomic claim() primitive yet must still deny a second claim —
+    // just via the older, non-atomic get-then-set pair (documented as strictly no worse than this function's
+    // prior behavior), not by skipping the check entirely.
+    const values = new Map<string, string>();
+    const env = createTestEnv({
+      SELFHOST_TRANSIENT_CACHE: {
+        get: async (key: string) => values.get(key) ?? null,
+        set: async (key: string, value: string) => { values.set(key, value); },
+      },
+    });
+    expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
+    expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(false);
+  });
+
+  it("INVARIANT (#2129 per-PR lock): a maintenance pass defers when another pass already holds the PR's lock", async () => {
+    const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() });
+    await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } });
+    await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001);
+    await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" });
+    await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Clean PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1" });
+    let mergeCalls = 0;
+    vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => {
+      const url = input.toString();
+      if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" });
+      if (url.includes("/pulls/7/files")) return Response.json([{ filename: "src/a.ts", status: "modified", additions: 1, deletions: 0, changes: 1, patch: "@@\n+export const ok = true;" }]);
+      if (url.includes("/pulls/7/merge")) {
+        mergeCalls += 1;
+        return new Response(null, { status: 204 });
+      }
+      if (url.includes("/pulls/7/reviews") && init?.method === "POST") return Response.json({ id: 1 });
+      if (url.includes("/pulls/7/reviews")) return Response.json([]);
+      // Only the bare PR resource (no sub-path) — the more specific checks above already claimed
+      // /pulls/7/files, /pulls/7/merge, and /pulls/7/reviews.
+      if (/\/pulls\/7(\?|$)/.test(url)) return Response.json({ number: 7, title: "Clean PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1", mergeable_state: "clean" });
+      if (url.includes("/commits/a7/check-runs")) return Response.json({ total_count: 0, check_runs: [] });
+      if (url.includes("/commits/a7/status")) return Response.json({ state: "success", statuses: [] });
+      if (url.includes("/commits/a7/check-suites")) return Response.json({ check_suites: [] });
+      if (url.includes("/issues/1")) return Response.json({ number: 1, title: "Issue", state: "open", labels: [], user: { login: "reporter" } });
+      if (url.includes("/branches/")) return Response.json({ protected: false, protection: { required_status_checks: { contexts: [] } } });
+      if (url.includes(".gittensory.yml")) return new Response("Not Found", { status: 404 });
+      if (url.endsWith("/check-runs") && init?.method === "POST") return Response.json({ id: 1 });
+      if (url === "https://api.gittensor.io/miners") return Response.json([]);
+      if (url.endsWith("/graphql")) return Response.json({ data: {} });
+      return Response.json({});
+    });
+    vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z"));
+
+    // Simulate a webhook pass already in-flight for this exact PR — a github-webhook:pr-refresh job's coalesce
+    // key never matches agent-regate-pr's, so the two would never dedup against each other pre-#2129; the new
+    // per-PR advisory lock is what makes a second, independently-triggered pass defer instead of racing it.
+    await env.SELFHOST_TRANSIENT_CACHE?.set("agent-maintenance-lock:owner/agent-repo#7", "1", 60);
+
+    await processJob(env, { type: "agent-regate-pr", deliveryId: "race-sweep", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001 });
+
+    // The held lock made this pass skip its plan-and-execute critical section entirely — no mutation attempted.
+    expect(mergeCalls).toBe(0);
+    const actionAudits = await env.DB.prepare("select count(*) as n from audit_events where event_type like 'agent.action.%'").first<{ n: number }>();
+    expect(actionAudits?.n).toBe(0);
+  });
+
   it("the sweep stamps the marker INLINE when the repo has no installation (audit-only, still converges) (#audit-sweep-fanout)", 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 });