fix(selfhost): backpressure scheduled regate sweeps

src/index.ts

@@ -7,11 +7,16 @@ import { isOrbBrokerEnabled } from "./orb/broker";
 import { isOpsEnabled } from "./review/ops-wire";
 import { isRagEnabled } from "./review/rag-wire";
 import { isSelfTuneEnabled } from "./review/selftune-wire";
-import { isGitHubBudgetBackgroundJob } from "./selfhost/queue-common";
+import {
+  isGitHubBudgetBackgroundJob,
+  queueSnapshotBacklog,
+  queueSnapshotFromBinding,
+} from "./selfhost/queue-common";
 import { isReviewExecutionJob, isSelfHostedReviewRuntime } from "./selfhost/review-runtime";
 import type { JobMessage } from "./types";
 
 const app = createApp();
+const REGATE_BACKPRESSURE_TYPES = ["agent-regate-pr", "agent-regate-sweep"] as const;
 
 export { RateLimiter };
 
@@ -99,11 +104,26 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
   // tick (~2 min) retries, and after the bucket resets the sweep resumes. Webhooks never pre-yield.
   const jobs: JobMessage[] = [];
   const selfHostedReviews = isSelfHostedReviewRuntime(env);
+  const queueSnapshot = selfHostedReviews
+    ? await queueSnapshotFromBinding(env.JOBS).catch((error) => {
+        console.warn(
+          JSON.stringify({
+            level: "warn",
+            event: "selfhost_queue_snapshot_failed",
+            error: error instanceof Error ? error.message : "unknown error",
+          }),
+        );
+        return null;
+      })
+    : null;
+  const regateBacklog = queueSnapshotBacklog(queueSnapshot, REGATE_BACKPRESSURE_TYPES);
   let sweepThrottledUntil: string | undefined;
   if (selfHostedReviews) {
     sweepThrottledUntil = await shouldWaitForGitHubRateLimit(env, MAINTENANCE_RESERVED_HEADROOM);
     if (sweepThrottledUntil) {
       console.log(JSON.stringify({ event: "regate_sweep_throttled", resetAt: sweepThrottledUntil }));
+    } else if (regateBacklog > 0) {
+      console.log(JSON.stringify({ event: "regate_sweep_backlog_deferred", backlog: regateBacklog }));
     } else {
       jobs.push({ type: "agent-regate-sweep", requestedBy: "schedule" });
     }
@@ -120,8 +140,10 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
     // 30-min tick retries, and after the bucket resets the backfill resumes. The cheap single-call health jobs
     // (repair-data-fidelity, refresh-installation-health) stay unconditional — they cost ~one call and keep
     // installation/health state fresh even while the budget is reserved.
-    if (selfHostedReviews && !sweepThrottledUntil) {
+    if (selfHostedReviews && !sweepThrottledUntil && regateBacklog === 0) {
       jobs.push({ type: "backfill-registered-repos", requestedBy: "schedule", mode: isFullSyncWindow ? "full" : "light" });
+    } else if (selfHostedReviews && regateBacklog > 0) {
+      console.log(JSON.stringify({ event: "backfill_backlog_deferred", backlog: regateBacklog }));
     } else if (selfHostedReviews) {
       console.log(JSON.stringify({ event: "backfill_throttled", resetAt: sweepThrottledUntil }));
     }

src/queue/processors.ts

@@ -209,6 +209,10 @@ import {
   delayUntil,
   shouldWaitForGitHubRateLimit,
 } from "../github/rate-limit";
+import {
+  queueSnapshotBacklog,
+  queueSnapshotFromBinding,
+} from "../selfhost/queue-common";
 import {
   downgradeCloseToHold,
   downgradeMergeToHold,
@@ -408,6 +412,7 @@ import { errorMessage, nowIso } from "../utils/json";
 
 const OFFICIAL_MINER_DETECTION_TTL_MS = 5 * 60 * 1000;
 const OFFICIAL_MINER_DETECTION_UNAVAILABLE_TTL_MS = 60 * 1000;
+const PER_PR_REGATE_BACKPRESSURE_TYPES = ["agent-regate-pr"] as const;
 const PR_PUBLIC_SURFACE_ACTIONS = new Set([
   "opened",
   "reopened",
@@ -791,7 +796,7 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
         await fanOutAgentRegateSweepJobs(env, message.requestedBy);
         return;
       }
-      await sweepRepoRegate(env, message.repoFullName);
+      await sweepRepoRegate(env, message.repoFullName, message.requestedBy);
       return;
     case "agent-regate-pr":
       // One bounded re-gate unit fanned out by the sweep (#audit-sweep-fanout): re-review + stamp a single PR.
@@ -1020,6 +1025,11 @@ async function fanOutAgentRegateSweepJobs(
   });
 }
 
+async function currentRegateBacklog(env: Env): Promise<number> {
+  const snapshot = await queueSnapshotFromBinding(env.JOBS).catch(() => null);
+  return queueSnapshotBacklog(snapshot, PER_PR_REGATE_BACKPRESSURE_TYPES);
+}
+
 // Convergence (RAG / codebase index, flag GITTENSORY_REVIEW_RAG). The dispatch for the `rag-index-repo` job.
 // Caller already gated on isRagEnabled(env).
 //   - No repoFullName → cron fan-out: enqueue one FULL re-index job per registered + cutover-allowlisted repo.
@@ -1142,6 +1152,7 @@ async function maybeEnqueueRagReindexForMergedPr(
 async function sweepRepoRegate(
   env: Env,
   repoFullName: string | undefined,
+  requestedBy: "schedule" | "api" | "test",
 ): Promise<void> {
   if (!repoFullName) return;
   const settings = await resolveRepositorySettings(env, repoFullName);
@@ -1171,6 +1182,19 @@ async function sweepRepoRegate(
     });
     return;
   }
+  const regateBacklog = requestedBy === "schedule" ? await currentRegateBacklog(env) : 0;
+  if (regateBacklog > 0) {
+    await recordAuditEvent(env, {
+      eventType: "agent.sweep.regate",
+      actor: "gittensory",
+      targetKey: repoFullName,
+      outcome: "queued",
+      detail:
+        "re-gate sweep deferred: prior scheduled re-gate work is still pending or processing",
+      metadata: { repoFullName, mode, deferred: true, regateBacklog },
+    });
+    return;
+  }
   const [repo, openPullRequests] = await Promise.all([
     getRepository(env, repoFullName),
     listOpenPullRequests(env, repoFullName),

src/selfhost/pg-queue.ts

@@ -15,6 +15,7 @@ import {
   githubRateLimitAdmissionTargetForJob,
   githubRateLimitMetricContext,
   githubRateLimitRetryDelayMs,
+  buildSelfHostQueueSnapshot,
   jobCoalesceKey,
   jobPriority,
   queueBackgroundConcurrency,
@@ -25,6 +26,7 @@ import {
   rateLimitRetryDelayWithJitter,
   matchesGitHubRateLimitAdmissionTarget,
   type GitHubRateLimitAdmissionTarget,
+  type SelfHostQueueSnapshot,
 } from "./queue-common";
 import type { JobMessage } from "../types";
 
@@ -60,6 +62,7 @@ export interface PgDurableQueue {
   size(): Promise<number>;
   deadCount(): Promise<number>;
   stats(): Promise<Record<string, number>>;
+  snapshot(): Promise<SelfHostQueueSnapshot>;
 }
 
 interface JobRow {
@@ -570,6 +573,14 @@ export function createPgQueue(
     async stats() {
       return readQueueStats();
     },
+    async snapshot() {
+      const res = await pool.query(
+        `SELECT payload, status, run_after FROM ${TABLE} WHERE status IN ('pending','processing','dead')`,
+      );
+      return buildSelfHostQueueSnapshot(
+        res.rows as Array<{ payload: string; status: string; run_after: string | number }>,
+      );
+    },
   };
 
   async function reclaimExpiredProcessingJobs(): Promise<number> {

src/selfhost/queue-common.ts

@@ -20,6 +20,24 @@ const DEFAULT_PROCESSING_TIMEOUT_MS = 30 * 60_000;
 const DEFAULT_BACKGROUND_CONCURRENCY = 1;
 export const FOREGROUND_QUEUE_PRIORITY_FLOOR = 8;
 
+export type SelfHostQueueJobStatus = "pending" | "processing" | "dead";
+
+export type SelfHostQueueSnapshotRow = {
+  type: string;
+  status: SelfHostQueueJobStatus;
+  count: number;
+  due: number;
+};
+
+export type SelfHostQueueSnapshot = {
+  totals: Record<SelfHostQueueJobStatus, number> & { due: number };
+  byType: SelfHostQueueSnapshotRow[];
+};
+
+export interface SelfHostQueueIntrospection {
+  snapshot(): SelfHostQueueSnapshot | Promise<SelfHostQueueSnapshot>;
+}
+
 // Webhook-driven work (a fresh PR -> its review) jumps ahead of heavy background jobs. Per-PR review refreshes
 // sit just below real webhooks, and sweep fan-out sits below those so stale surfaces are repaired during bursts.
 // Bot-generated comment edits are background noise; keeping them with real webhooks lets panel edits starve repair.
@@ -95,6 +113,61 @@ export function isGitHubBudgetBackgroundJob(message: JobMessage): boolean {
   return GITHUB_BUDGET_BACKGROUND_TYPES.has(message.type);
 }
 
+export function buildSelfHostQueueSnapshot(
+  rows: Iterable<{ payload?: unknown; status?: unknown; run_after?: unknown; runAfter?: unknown }>,
+  nowMs = Date.now(),
+): SelfHostQueueSnapshot {
+  const totals = { pending: 0, processing: 0, dead: 0, due: 0 };
+  const byKey = new Map<string, SelfHostQueueSnapshotRow>();
+  for (const row of rows) {
+    const status = queueStatus(row.status);
+    if (!status) continue;
+    const type = typeof row.payload === "string" ? (extractPayloadType(row.payload) ?? "unknown") : "unknown";
+    const runAfter = queueRunAfterMs(row.run_after ?? row.runAfter);
+    const due = status === "pending" && (runAfter === null || runAfter <= nowMs) ? 1 : 0;
+    const key = `${type}\0${status}`;
+    const current = byKey.get(key) ?? { type, status, count: 0, due: 0 };
+    current.count += 1;
+    current.due += due;
+    byKey.set(key, current);
+    totals[status] += 1;
+    totals.due += due;
+  }
+  return {
+    totals,
+    byType: [...byKey.values()].sort((a, b) => a.type.localeCompare(b.type) || a.status.localeCompare(b.status)),
+  };
+}
+
+export function queueSnapshotBacklog(
+  snapshot: SelfHostQueueSnapshot | null | undefined,
+  types: readonly string[],
+  statuses: readonly SelfHostQueueJobStatus[] = ["pending", "processing"],
+): number {
+  if (!snapshot) return 0;
+  const typeSet = new Set(types);
+  const statusSet = new Set(statuses);
+  return snapshot.byType.reduce(
+    (sum, row) => sum + (typeSet.has(row.type) && statusSet.has(row.status) ? row.count : 0),
+    0,
+  );
+}
+
+export async function queueSnapshotFromBinding(binding: Queue): Promise<SelfHostQueueSnapshot | null> {
+  const snapshot = (binding as Queue & Partial<SelfHostQueueIntrospection>).snapshot;
+  if (typeof snapshot !== "function") return null;
+  return snapshot.call(binding);
+}
+
+function queueStatus(value: unknown): SelfHostQueueJobStatus | null {
+  return value === "pending" || value === "processing" || value === "dead" ? value : null;
+}
+
+function queueRunAfterMs(value: unknown): number | null {
+  const parsed = typeof value === "number" ? value : typeof value === "string" ? Number(value) : null;
+  return parsed !== null && Number.isFinite(parsed) ? parsed : null;
+}
+
 function githubObservedRateLimitDelayMs(
   observation:
     | { remaining?: unknown; reset_at?: unknown; resetAt?: unknown }

src/selfhost/sqlite-queue.ts

@@ -16,6 +16,7 @@ import {
   githubRateLimitAdmissionTargetForJob,
   githubRateLimitMetricContext,
   githubRateLimitRetryDelayMs,
+  buildSelfHostQueueSnapshot,
   jobCoalesceKey,
   jobPriority,
   queueBackgroundConcurrency,
@@ -26,6 +27,7 @@ import {
   rateLimitRetryDelayWithJitter,
   matchesGitHubRateLimitAdmissionTarget,
   type GitHubRateLimitAdmissionTarget,
+  type SelfHostQueueSnapshot,
 } from "./queue-common";
 import type { JobMessage } from "../types";
 
@@ -62,6 +64,7 @@ export interface DurableQueue {
   size(): number;
   deadCount(): number;
   stats(): Record<string, number>;
+  snapshot(): SelfHostQueueSnapshot;
 }
 
 interface JobRow {
@@ -518,6 +521,14 @@ export function createSqliteQueue(
     stats() {
       return readQueueStats(driver);
     },
+    snapshot() {
+      return buildSelfHostQueueSnapshot(
+        driver.query(
+          `SELECT payload, status, run_after FROM ${TABLE} WHERE status IN ('pending','processing','dead')`,
+          [],
+        ).rows as Array<{ payload: string; status: string; run_after: number }>,
+      );
+    },
   };
 }
 

src/settings/agent-sweep.ts

@@ -12,12 +12,11 @@ import type { PullRequestRecord } from "../types";
 // REST bucket. Each fanned-out per-PR re-review costs ~9 REST GETs (1 resync `GET /pulls/{n}`, then required-
 // contexts + CI aggregate in prReadyForReview, then required-contexts + merge-state + CI aggregate + files in
 // auto-maintain). The sweep re-arms every ~2 min (≈30 ticks/hr) and fans out per repo, so the worst-case hourly
-// sweep cost is `SWEEP_MAX_PRS × repos × 9 × 30`. At the old cap of 25 over 3 self-host repos that is ~200k/hr —
-// far over budget, the cause of the `regate_sweep_throttled` exhaustion. A cap of 6 bounds the worst case to
-// `6 × 3 × 9 × 30 ≈ 4.9k/hr` (and steady state is far lower — the freshness skip + the in-flight drain guard stop
-// re-regating just-touched PRs), reserving headroom for webhooks. A 30-PR backlog still fully converges in
-// ceil(30/6)=5 sweeps (~10 min), so no PR's merge/close is meaningfully delayed.
-export const SWEEP_MAX_PRS = 6;
+// sweep cost is `SWEEP_MAX_PRS × repos × 9 × 30`. At the old cap of 25 over 3 self-host repos that is ~200k/hr.
+// A cap of 6 still consumed nearly the whole REST bucket across three active repos, so scheduled sweeps now run
+// at a smaller source budget and also skip while prior regate work is queued. A cap of 3 bounds the raw worst case
+// to `3 × 3 × 9 × 30 ≈ 2.4k/hr`, leaving budget for live webhooks, cache misses, and branch-protection reads.
+export const SWEEP_MAX_PRS = 3;
 
 // Skip-if-fresh window: a PR touched within this span was almost certainly just gated by its webhook, so the
 // sweep leaves it alone for that brief moment to avoid racing the in-flight webhook review. Kept SHORT (2 min)

test/unit/agent-sweep.test.ts

@@ -132,7 +132,7 @@ describe("selectRegateCandidates (#777 re-gate sweep selection)", () => {
 
   it("defaults: freshness window is two minutes and the cap is bounded for the shared REST budget (#audit-rate-headroom)", () => {
     expect(SWEEP_FRESHNESS_MS).toBe(2 * 60 * 1000);
-    expect(SWEEP_MAX_PRS).toBe(6); // lowered from 25: 6 × 3 repos × 9 GETs × 30 ticks/hr ≈ 4.9k/hr ≤ the 5000 bucket
+    expect(SWEEP_MAX_PRS).toBe(3); // 3 × 3 repos × 9 GETs × 30 ticks/hr ≈ 2.4k/hr, leaving headroom for webhooks
     const pulls = Array.from({ length: 40 }, (_, i) => pr({ number: i + 1, createdAt: minutesAgo(120 + i) }));
     expect(selectRegateCandidates({ pulls, now: NOW })).toHaveLength(SWEEP_MAX_PRS);
   });

test/unit/index.test.ts

@@ -276,6 +276,55 @@ describe("worker entrypoint", () => {
     expect(sent).toEqual([{ type: "agent-regate-sweep", requestedBy: "schedule" }]);
   });
 
+  it("does not enqueue scheduled sweep/backfill work while prior regate jobs are still queued", async () => {
+    const sent: Array<import("../../src/types").JobMessage> = [];
+    const env = createTestEnv({
+      JOBS: {
+        async send(message: import("../../src/types").JobMessage) {
+          sent.push(message);
+        },
+        snapshot: async () => ({
+          totals: { pending: 2, processing: 1, dead: 0, due: 2 },
+          byType: [
+            { type: "agent-regate-pr", status: "pending", count: 2, due: 2 },
+            { type: "agent-regate-sweep", status: "processing", count: 1, due: 0 },
+          ],
+        }),
+      } as unknown as Queue,
+    });
+    const waitUntil: Promise<unknown>[] = [];
+
+    await worker.scheduled(controllerFor("2026-05-25T05:30:00.000Z"), env, executionContext(waitUntil));
+    await Promise.all(waitUntil);
+
+    expect(sent).toEqual([
+      { type: "repair-data-fidelity", requestedBy: "schedule" },
+      { type: "refresh-installation-health", requestedBy: "schedule" },
+    ]);
+  });
+
+  it("fails open when queue introspection is unavailable so scheduled maintenance still runs", async () => {
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    const sent: Array<import("../../src/types").JobMessage> = [];
+    const env = createTestEnv({
+      JOBS: {
+        async send(message: import("../../src/types").JobMessage) {
+          sent.push(message);
+        },
+        snapshot: async () => {
+          throw new Error("snapshot unavailable");
+        },
+      } as unknown as Queue,
+    });
+    const waitUntil: Promise<unknown>[] = [];
+
+    await worker.scheduled(controllerFor("2026-05-25T05:14:00.000Z"), env, executionContext(waitUntil));
+    await Promise.all(waitUntil);
+
+    expect(sent).toEqual([{ type: "agent-regate-sweep", requestedBy: "schedule" }]);
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining("selfhost_queue_snapshot_failed"));
+  });
+
   it("does not enqueue review sweeps from a broker-only Cloudflare runtime", async () => {
     const sent: Array<import("../../src/types").JobMessage> = [];
     const env = createTestEnv({