fix(sweep): dedup the re-gate fan-out so a burst collapses to one effective sweep

The ~2-min cron enqueues a fan-out job each tick, but fanOutAgentRegateSweepJobs
had no global dedup. When a burst of fan-out jobs ran at once — a deploy-restart
cron catch-up, or fan-out jobs that queued behind a heavy per-PR re-review
backlog and then drained together — EACH one enqueued a per-repo sweep before the
per-repo dispatch-stamp in-flight guard could engage, producing redundant
overlapping sweeps (observed ~3x on the metagraphed dry-run). Those redundant
sweeps tripled the per-PR load, which delayed the next fan-out, which then burst
in turn — a self-sustaining cascade.

Add an atomic fan-out dedup: claimRegateFanoutSlot does a conditional UPDATE on
the global_agent_controls singleton (new last_regate_fanout_at column, migration
0063) that matches only when the last fan-out is unset or older than the dedup
window. D1 serializes writes, so a burst collapses to exactly ONE winner per
window; the rest get 0 changes and skip (audited deduped). One effective fan-out
per window keeps the per-PR load bounded, which stops the backlog that was
delaying subsequent fan-outs — breaking the cascade at its source. Fail-open on a
driver error so the fleet never stalls.

Author: JSONbored

migrations/0063_sweep_fanout_marker.sql

@@ -0,0 +1,16 @@
+-- Fan-out dedup marker: collapse a burst of re-gate fan-out jobs to ONE effective fan-out per window.
+--
+-- BEFORE: the ~2-min cron enqueues an agent-regate-sweep fan-out job each tick; fanOutAgentRegateSweepJobs runs it
+-- with no global dedup. When a burst of fan-out jobs runs at once — a deploy-restart cron catch-up, or fan-out
+-- jobs that queued behind a heavy per-PR re-review backlog and then drained together — EACH one enqueues a
+-- per-repo sweep before the per-repo dispatch-stamp in-flight guard (0062 / #audit-sweep-dispatch-stamp) can
+-- engage, producing redundant overlapping sweeps (observed ~3x on the metagraphed dry-run).
+--
+-- AFTER: claimRegateFanoutSlot performs an atomic conditional UPDATE on this singleton column — D1 serializes
+-- writes, so only ONE concurrent fan-out's UPDATE matches the "unset or older than the dedup window" predicate
+-- and proceeds; the rest get 0 changes and skip. One effective fan-out per window keeps the per-PR load bounded,
+-- which in turn stops the backlog that was delaying subsequent fan-outs (the cascade that caused the bursts).
+--
+-- Reuses the global_agent_controls singleton (0059); nullable / no default → backward-compatible (NULL = no
+-- fan-out has claimed the slot yet, so the first one proceeds).
+ALTER TABLE global_agent_controls ADD COLUMN last_regate_fanout_at TEXT;

src/db/repositories.ts

@@ -1948,6 +1948,27 @@ export async function isGlobalAgentFrozen(env: Env): Promise<boolean> {
   }
 }
 
+/** Atomic re-gate fan-out dedup (#audit-fanout-dedup): claim the global fan-out slot for this window. The
+ *  conditional UPDATE on the singleton matches only when the last fan-out is unset or older than `windowMs`. D1
+ *  serializes writes, so when a BURST of fan-out jobs runs at once (a deploy-restart cron catch-up, or fan-out
+ *  jobs that queued behind a per-PR backlog and drained together) exactly ONE wins the slot (changes === 1); the
+ *  rest get 0 changes and skip, collapsing the burst to a single effective fan-out. Fail-open on a driver error
+ *  (return true → the sweep still runs, degrading to the pre-dedup behaviour rather than stalling the fleet). */
+export async function claimRegateFanoutSlot(env: Env, now: string, windowMs: number): Promise<boolean> {
+  const threshold = new Date(Date.parse(now) - windowMs).toISOString();
+  try {
+    const result = await env.DB.prepare(
+      "UPDATE global_agent_controls SET last_regate_fanout_at = ?1 WHERE id = 'singleton' AND (last_regate_fanout_at IS NULL OR last_regate_fanout_at < ?2)",
+    )
+      .bind(now, threshold)
+      .run();
+    /* v8 ignore next -- D1 update metadata normally includes changes; the ?? 0 fallback protects driver anomalies. */
+    return Number(result.meta.changes ?? 0) === 1;
+  } catch {
+    return true;
+  }
+}
+
 /** Flip the DB-backed global kill-switch (operator emergency brake; no redeploy required). */
 export async function setGlobalAgentFrozen(env: Env, frozen: boolean, updatedBy?: string | null): Promise<void> {
   await env.DB.prepare(

src/queue/processors.ts

@@ -41,6 +41,7 @@ import {
   persistAdvisory,
   markPullRequestsRegated,
   getLatestRegatedAt,
+  claimRegateFanoutSlot,
   recordAgentCommandFeedback,
   recordAuditEvent,
   recordGateBlockOutcome,
@@ -115,7 +116,7 @@ import { isAuthorizedGitHubSessionLogin, parseGitHubLoginList } from "../auth/se
 import { commandAuthorizationAllowedRoles, commandAuthorizationNeedsMinerDetection } from "../settings/command-authorization";
 import { autonomyRequiresApproval, isAgentConfigured, resolveAutonomy } from "../settings/autonomy";
 import { isGlobalAgentPause, resolveAgentActionMode } from "../settings/agent-execution";
-import { isRegateSweepDraining, selectRegateCandidates } from "../settings/agent-sweep";
+import { SWEEP_FANOUT_DEDUP_MS, isRegateSweepDraining, selectRegateCandidates } from "../settings/agent-sweep";
 import { MAINTENANCE_RESERVED_HEADROOM, delayUntil, shouldWaitForGitHubRateLimit } from "../github/rate-limit";
 import { downgradeCloseToHold, downgradeMergeToHold, isProtectedAutomationAuthor, planAgentMaintenanceActions, type PlannedAgentAction } from "../settings/agent-actions";
 import { executeAgentMaintenanceActions, pendingClosureLabelApplied } from "../services/agent-action-executor";
@@ -432,8 +433,20 @@ async function fanOutRepoSignalSnapshotJobs(env: Env, requestedBy: "schedule" |
 // sweep job for every repo that opted the agent in (an acting autonomy level). Mirrors the signal-snapshot
 // fan-out so each repo's sweep runs as its own bounded, retryable queue message.
 async function fanOutAgentRegateSweepJobs(env: Env, requestedBy: "schedule" | "api" | "test"): Promise<void> {
-  const repositories = await listRepositories(env);
   const now = nowIso();
+  // Atomic fan-out dedup (#audit-fanout-dedup): collapse a BURST of fan-out jobs to a SINGLE effective fan-out per
+  // window, so a deploy-restart cron catch-up (or fan-out jobs delayed behind a per-PR backlog then drained
+  // together) cannot each enqueue a redundant per-repo sweep before the per-repo dispatch-stamp guard engages.
+  if (!(await claimRegateFanoutSlot(env, now, SWEEP_FANOUT_DEDUP_MS))) {
+    await recordAuditEvent(env, {
+      eventType: "agent.sweep.fanout",
+      outcome: "denied",
+      detail: "re-gate fan-out deduped: another fan-out already claimed this window",
+      metadata: { requestedBy, deduped: true },
+    });
+    return;
+  }
+  const repositories = await listRepositories(env);
   const configured: string[] = [];
   let skippedDraining = 0;
   for (const repo of repositories) {

src/settings/agent-sweep.ts

@@ -16,6 +16,12 @@ export const SWEEP_MAX_PRS = 25;
 // approved PRs unmerged for up to an hour.
 export const SWEEP_FRESHNESS_MS = 2 * 60 * 1000;
 
+// Fan-out dedup window (#audit-fanout-dedup): a burst of fan-out jobs within this window collapses to ONE
+// effective fan-out. Kept BELOW the ~2-min cron cadence so a legitimate next-tick fan-out is never skipped, but
+// well above the few-seconds spread of a burst (a deploy-restart cron catch-up, or fan-out jobs that queued
+// behind a per-PR backlog and drained together).
+export const SWEEP_FANOUT_DEDUP_MS = 90 * 1000;
+
 /**
  * Select the open PRs a single repo sweep should recompute: drop drafts and anything a webhook touched within
  * `freshnessWindowMs` of `now` (don't race an in-flight review), then take the `max` PRs the sweep has gone

test/unit/db-parsers.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it } from "vitest";
 import {
+  claimRegateFanoutSlot,
   countRecentDeadLetters,
   getLatestScorePreview,
   getRepoAuthorPullRequestHistory,
@@ -113,6 +114,22 @@ describe("database row parser hardening", () => {
     expect(rows.find((p) => p.number === 6)?.lastRegatedAt ?? null).toBeNull(); // #6 not in the batch → untouched
   });
 
+  it("claimRegateFanoutSlot collapses a burst to one winner per window (#audit-fanout-dedup)", async () => {
+    const env = createTestEnv();
+    const W = 90 * 1000;
+    expect(await claimRegateFanoutSlot(env, "2026-06-25T01:00:00.000Z", W)).toBe(true); // first claim wins (marker NULL)
+    expect(await claimRegateFanoutSlot(env, "2026-06-25T01:00:05.000Z", W)).toBe(false); // +5s, inside window → loses
+    expect(await claimRegateFanoutSlot(env, "2026-06-25T01:00:50.000Z", W)).toBe(false); // +50s, still inside → loses
+    expect(await claimRegateFanoutSlot(env, "2026-06-25T01:01:31.000Z", W)).toBe(true); // +91s, outside window → wins again
+    expect(await claimRegateFanoutSlot(env, "2026-06-25T01:01:40.000Z", W)).toBe(false); // back inside the new window → loses
+  });
+
+  it("claimRegateFanoutSlot fails open (returns true) on a DB error so the fleet never stalls", async () => {
+    const env = createTestEnv();
+    const broken = { ...env, DB: null } as unknown as typeof env;
+    expect(await claimRegateFanoutSlot(broken, "2026-06-25T01:00:00.000Z", 90 * 1000)).toBe(true);
+  });
+
   it("REGRESSION: a later GitHub sync does NOT clobber last_regated_at (omitted from the upsert SET clause)", async () => {
     const env = createTestEnv();
     await upsertPullRequestFromGitHub(env, "owner/repo", { number: 6, title: "First", state: "open", user: { login: "bob" }, labels: [] });

test/unit/queue.test.ts

@@ -891,6 +891,25 @@ describe("queue processors", () => {
     expect(JSON.parse(fanout?.metadata_json ?? "{}")).toMatchObject({ repoCount: 1, skippedDraining: 1 });
   });
 
+  it("INVARIANT (#audit-fanout-dedup): a BURST of fan-outs collapses to ONE — the second claims nothing and audits denied", 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: 9400, 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" } }, 9400);
+    await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" } });
+    await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "PR7", state: "open", user: { login: "c" }, head: { sha: "a7" }, labels: [], body: "" });
+    vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z"));
+
+    await processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" }); // first fan-out claims the window
+    expect(sent.some((m) => m.type === "agent-regate-sweep" && m.repoFullName === "owner/agent-repo")).toBe(true);
+
+    sent.length = 0;
+    await processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" }); // burst sibling in the same window → deduped
+    expect(sent.filter((m) => m.type === "agent-regate-sweep")).toEqual([]); // enqueues no redundant sweep
+    const denied = await env.DB.prepare("select count(*) as n from audit_events where event_type='agent.sweep.fanout' and outcome='denied'").first<{ n: number }>();
+    expect(denied?.n).toBe(1);
+  });
+
   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 });