fix(queue): isolate per-repo failures in the scheduled regate sweep fan-out

src/queue/processors.ts

@@ -1258,27 +1258,44 @@ async function fanOutAgentRegateSweepJobs(
   }
   const configured: Array<{ fullName: string; installationId?: number }> = [];
   let skippedDraining = 0;
+  let skippedErrored = 0;
   for (const repo of byKey.values()) {
     const repoFullName = repo.fullName;
-    const settings = await resolveRepositorySettings(env, repoFullName);
-    if (
-      !(
-        isConvergenceRepoAllowed(env, repoFullName) ||
-        isAgentConfigured(settings.autonomy)
+    // #audit-sweep-fanout-isolation: one repo's settings/draining-check failure (a transient D1 read error, say)
+    // must not throw out of this loop and abort the fan-out for EVERY OTHER already-iterated-and-pending repo —
+    // it previously did, since an uncaught throw here escapes the whole function before the dispatch loop below
+    // ever runs. Skip just this repo (it gets picked up again next tick) and keep going.
+    try {
+      const settings = await resolveRepositorySettings(env, repoFullName);
+      if (
+        !(
+          isConvergenceRepoAllowed(env, repoFullName) ||
+          isAgentConfigured(settings.autonomy)
+        )
       )
-    )
-      continue;
-    // In-flight guard (#audit-sweep-fanout): skip a repo whose prior sweep is still draining — its per-PR jobs are
-    // mid-flight and stamping last_regated_at as they run, so the freshest stamp being within the sweep window
-    // means a sweep is active. Re-arming now would enqueue duplicate per-PR jobs for the not-yet-drained
-    // candidates, so this is what finally stops the 2-min cron piling a second full sweep on an unfinished one.
-    if (
-      isRegateSweepDraining(await getLatestRegatedAt(env, repoFullName), now)
-    ) {
-      skippedDraining += 1;
-      continue;
+        continue;
+      // In-flight guard (#audit-sweep-fanout): skip a repo whose prior sweep is still draining — its per-PR jobs are
+      // mid-flight and stamping last_regated_at as they run, so the freshest stamp being within the sweep window
+      // means a sweep is active. Re-arming now would enqueue duplicate per-PR jobs for the not-yet-drained
+      // candidates, so this is what finally stops the 2-min cron piling a second full sweep on an unfinished one.
+      if (
+        isRegateSweepDraining(await getLatestRegatedAt(env, repoFullName), now)
+      ) {
+        skippedDraining += 1;
+        continue;
+      }
+      configured.push(repo);
+    } catch (error) {
+      skippedErrored += 1;
+      console.error(
+        JSON.stringify({
+          level: "error",
+          event: "sweep_fanout_repo_check_failed",
+          repository: repoFullName,
+          error: errorMessage(error),
+        }),
+      );
     }
-    configured.push(repo);
   }
   await Promise.all(
     configured.map((repo, index) => {
@@ -1289,15 +1306,27 @@ async function fanOutAgentRegateSweepJobs(
         ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}),
       };
       const delaySeconds = Math.min(index * 10, 600);
-      return delaySeconds > 0
-        ? env.JOBS.send(message, { delaySeconds })
-        : env.JOBS.send(message);
+      // #audit-sweep-fanout-isolation: one repo's dispatch failure (a transient queue-send error) must not reject
+      // this Promise.all and, with it, abort every OTHER repo's already-in-flight send AND the audit event below
+      // that records this fan-out's outcome. Swallow + log per-repo instead; a repo that fails to dispatch here
+      // simply gets picked up again next tick (it never got its convergence marker stamped, so it stays eligible).
+      const send = delaySeconds > 0 ? env.JOBS.send(message, { delaySeconds }) : env.JOBS.send(message);
+      return send.catch((error) => {
+        console.error(
+          JSON.stringify({
+            level: "error",
+            event: "sweep_fanout_dispatch_failed",
+            repository: repo.fullName,
+            error: errorMessage(error),
+          }),
+        );
+      });
     }),
   );
   await recordAuditEvent(env, {
     eventType: "agent.sweep.fanout",
     outcome: "queued",
-    metadata: { repoCount: configured.length, skippedDraining, requestedBy },
+    metadata: { repoCount: configured.length, skippedDraining, skippedErrored, requestedBy },
   });
 }
 

test/unit/queue.test.ts

@@ -768,6 +768,61 @@ describe("queue processors", () => {
     expect(sent).toEqual(expect.arrayContaining([expect.objectContaining({ type: "agent-regate-sweep", repoFullName: "owner/advisory-repo", installationId: 9102 })]));
   });
 
+  it("REGRESSION (#audit-sweep-fanout-isolation): one repo's settings-check failure does not abort the fan-out for every other repo", async () => {
+    const sent: import("../../src/types").JobMessage[] = [];
+    const env = createTestEnv({
+      GITTENSORY_REVIEW_REPOS: "",
+      JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue,
+    });
+    await upsertRepositoryFromGitHub(env, { name: "agent-a", full_name: "owner/agent-a", private: false, owner: { login: "owner" } });
+    await upsertRepositoryFromGitHub(env, { name: "agent-b", full_name: "owner/agent-b", private: false, owner: { login: "owner" } });
+    await upsertRepositorySettings(env, { repoFullName: "owner/agent-a", autonomy: { label: "auto" } });
+    await upsertRepositorySettings(env, { repoFullName: "owner/agent-b", autonomy: { label: "auto" } });
+    const realResolve = repositorySettingsModule.resolveRepositorySettings;
+    const errors = vi.spyOn(console, "error").mockImplementation(() => undefined);
+    const resolveSpy = vi.spyOn(repositorySettingsModule, "resolveRepositorySettings").mockImplementation(async (e, repoFullName) => {
+      if (repoFullName === "owner/agent-a") throw new Error("D1 read error");
+      return realResolve(e, repoFullName);
+    });
+
+    await processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" });
+
+    expect(sent).toEqual([expect.objectContaining({ type: "agent-regate-sweep", repoFullName: "owner/agent-b" })]); // agent-a's failure did not block agent-b
+    expect(errors.mock.calls.some((call) => String(call[0]).includes("sweep_fanout_repo_check_failed") && String(call[0]).includes("owner/agent-a"))).toBe(true);
+    const fanout = await env.DB.prepare("select outcome, metadata_json from audit_events where event_type = ?").bind("agent.sweep.fanout").first<{ outcome: string; metadata_json: string }>();
+    expect(fanout?.outcome).toBe("queued"); // the fan-out still completes and records its own outcome
+    expect(JSON.parse(fanout?.metadata_json ?? "{}")).toMatchObject({ repoCount: 1, skippedErrored: 1 });
+    errors.mockRestore();
+    resolveSpy.mockRestore();
+  });
+
+  it("REGRESSION (#audit-sweep-fanout-isolation): one repo's dispatch failure does not abort dispatch for every other repo, and the fan-out audit event still records", async () => {
+    const sent: import("../../src/types").JobMessage[] = [];
+    const env = createTestEnv({
+      GITTENSORY_REVIEW_REPOS: "",
+      JOBS: {
+        async send(m: import("../../src/types").JobMessage) {
+          if (m.type === "agent-regate-sweep" && m.repoFullName === "owner/agent-a") throw new Error("queue send error");
+          sent.push(m);
+        },
+      } as unknown as Queue,
+    });
+    await upsertRepositoryFromGitHub(env, { name: "agent-a", full_name: "owner/agent-a", private: false, owner: { login: "owner" } });
+    await upsertRepositoryFromGitHub(env, { name: "agent-b", full_name: "owner/agent-b", private: false, owner: { login: "owner" } });
+    await upsertRepositorySettings(env, { repoFullName: "owner/agent-a", autonomy: { label: "auto" } });
+    await upsertRepositorySettings(env, { repoFullName: "owner/agent-b", autonomy: { label: "auto" } });
+    const errors = vi.spyOn(console, "error").mockImplementation(() => undefined);
+
+    await processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" });
+
+    expect(sent).toEqual([expect.objectContaining({ type: "agent-regate-sweep", repoFullName: "owner/agent-b" })]); // agent-a's failed send did not block agent-b's
+    expect(errors.mock.calls.some((call) => String(call[0]).includes("sweep_fanout_dispatch_failed") && String(call[0]).includes("owner/agent-a"))).toBe(true);
+    const fanout = await env.DB.prepare("select outcome, metadata_json from audit_events where event_type = ?").bind("agent.sweep.fanout").first<{ outcome: string; metadata_json: string }>();
+    expect(fanout?.outcome).toBe("queued"); // reached — the dispatch failure did not throw the fan-out itself
+    expect(JSON.parse(fanout?.metadata_json ?? "{}")).toMatchObject({ repoCount: 2 }); // both PASSED their settings/draining checks regardless of dispatch outcome
+    errors.mockRestore();
+  });
+
   it("agent re-gate sweep recomputes stale open PR verdicts as an advisory audit, never publishing (#777)", async () => {
     const sent: import("../../src/types").JobMessage[] = [];
     const env = createTestEnv({