refactor: dedup estimateNeurons and the configured-repo-set assembly

src/queue/processors.ts

@@ -98,6 +98,7 @@ import {
   getLatestAdvisoryForPullRequest,
 } from "../db/repositories";
 import { withLinkedIssueMaintainerExemption, type LinkedIssueExemptionAuthor } from "../settings/linked-issue-exemption";
+import { resolveConfiguredRepoCandidates } from "../review/configured-repo-set";
 import { renameRepositoryIdentity } from "../db/repo-identity-rename";
 import {
   effectiveIssueCapForAccountAge,
@@ -632,7 +633,6 @@ import {
 } from "../review/reputation-wire";
 import {
   isConvergenceRepoAllowed,
-  listConvergenceRepos,
 } from "../review/cutover-gate";
 import {
   convergedFeatureActive,
@@ -940,24 +940,15 @@ export async function fanOutAgentRegateSweepJobs(
   // that can merge/close. The action layer (maybeRunAgentMaintenance) stays autonomy-gated, so an observe repo is
   // re-reviewed but never auto-actioned. This is what makes advisory reviews fire on existing open PRs without
   // depending on a fresh webhook per PR.
-  const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo]));
-  const byKey = new Map<string, { fullName: string; installationId?: number }>();
-  for (const repo of repositoriesByKey.values())
-    byKey.set(repo.fullName.toLowerCase(), { fullName: repo.fullName, ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}) });
-  for (const fullName of listConvergenceRepos(env)) {
-    const repo = repositoriesByKey.get(fullName.toLowerCase());
-    byKey.set(fullName.toLowerCase(), {
-      fullName,
-      ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}),
-    });
-  }
+  // #10170: the repo-set assembly was duplicated five times; callers keep their own eligibility rules.
+  const repoCandidates = await resolveConfiguredRepoCandidates(env);
   // #3899: resolve every repo's settings + drain-state CONCURRENTLY (bounded), not one at a time. Each repo
   // costs resolveRepositorySettings's own 3 parallel round-trips plus a 4th getLatestRegatedAt read; awaiting
   // that serially per repo made this whole prefix scale linearly with repo count, before the per-repo dispatch
   // below (already parallel) even started. Reuses the same bounded worker-pool helper loadRepoFocusManifests
   // already relies on for the same "many small per-repo D1/KV reads" shape.
   const outcomes = await mapWithConcurrencyLimit(
-    [...byKey.values()],
+    repoCandidates,
     SWEEP_FANOUT_RESOLUTION_CONCURRENCY,
     async (repo): Promise<SweepFanoutResolutionOutcome> => {
       const repoFullName = repo.fullName;
@@ -1312,18 +1303,9 @@ async function fanOutRagIndexJobs(
   // case-insensitively (a repo can be both known AND configured). Each candidate is then filtered by whether RAG is
   // active for it (`features.rag` override → LOOPOVER_REVIEW_REPOS allowlist default) just below, so this widens
   // ELIGIBILITY only — the convergedFeatureActive gate below is what actually controls indexing spend.
-  const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo]));
-  const byKey = new Map<string, { fullName: string; installationId?: number }>();
-  for (const repo of repositoriesByKey.values())
-    byKey.set(repo.fullName.toLowerCase(), { fullName: repo.fullName, ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}) });
-  for (const fullName of listConvergenceRepos(env)) {
-    const repo = repositoriesByKey.get(fullName.toLowerCase());
-    byKey.set(fullName.toLowerCase(), {
-      fullName,
-      ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}),
-    });
-  }
-  const candidates = [...byKey.values()];
+  // #10170: the repo-set assembly was duplicated five times; callers keep their own eligibility rules.
+  const repoCandidates = await resolveConfiguredRepoCandidates(env);
+  const candidates = repoCandidates;
   const ragActiveByRepo = await Promise.all(
     candidates.map((repo) => convergedFeatureActive(env, repo.fullName, "rag")),
   );
@@ -1733,21 +1715,12 @@ export async function fanOutBacklogConvergenceSweepJobs(
     });
     return;
   }
-  const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo]));
-  const byKey = new Map<string, { fullName: string; installationId?: number }>();
-  for (const repo of repositoriesByKey.values())
-    byKey.set(repo.fullName.toLowerCase(), { fullName: repo.fullName, ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}) });
-  for (const fullName of listConvergenceRepos(env)) {
-    const repo = repositoriesByKey.get(fullName.toLowerCase());
-    byKey.set(fullName.toLowerCase(), {
-      fullName,
-      ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}),
-    });
-  }
+  // #10170: the repo-set assembly was duplicated five times; callers keep their own eligibility rules.
+  const repoCandidates = await resolveConfiguredRepoCandidates(env);
   // #4502 (ports #3899): resolve every repo's settings + drain-state CONCURRENTLY (bounded), not one at a time —
   // mirrors fanOutAgentRegateSweepJobs's own port of this fix, the same "many small per-repo D1/KV reads" shape.
   const outcomes = await mapWithConcurrencyLimit(
-    [...byKey.values()],
+    repoCandidates,
     SWEEP_FANOUT_RESOLUTION_CONCURRENCY,
     async (repo): Promise<SweepFanoutResolutionOutcome> => {
       const repoFullName = repo.fullName;

src/review/configured-repo-set.ts

@@ -0,0 +1,44 @@
+// The set of repositories a fleet-wide pass should consider (#10170).
+//
+// Assembling this was duplicated FIVE times, byte-identical: three copies inside queue/processors.ts and one
+// each in review/pr-reconciliation.ts and review/sweep-watchdog.ts. Every fleet-wide sweep needs the same
+// answer to "which repos are in scope", and every one of them rebuilt it.
+//
+// Only the ASSEMBLY is shared. What each caller then does with the set -- resolving settings, requiring a
+// real installation, applying its own eligibility rule -- genuinely differs and deliberately stays at the
+// call site. Extracting more than this would force the callers' differences into flag parameters, which is
+// how a shared helper becomes worse than the duplication it replaced.
+//
+// Two sources are merged, and the ORDER matters: the convergence list wins. A repo named there is in scope
+// whether or not it has a local row, and when it has one, the row's installationId is carried over so a
+// caller that requires a real GitHub App installation can still tell. A locally-known repo absent from the
+// convergence list keeps its row as-is.
+
+import { listRepositories } from "../db/repositories";
+import { listConvergenceRepos } from "./cutover-gate";
+
+/** A repo in scope for a fleet-wide pass. `installationId` is absent -- not null -- when unknown, matching
+ *  the exactOptionalPropertyTypes shape every existing call site already spreads. */
+export type ConfiguredRepoCandidate = { fullName: string; installationId?: number };
+
+/**
+ * Every repository a fleet-wide pass should consider: locally-known rows plus the convergence allowlist,
+ * de-duplicated case-insensitively on the full name.
+ *
+ * Callers apply their own eligibility on top -- this deliberately does NOT decide whether a repo is
+ * agent-configured, installed, or due. It answers only "which repos exist for this pass".
+ */
+export async function resolveConfiguredRepoCandidates(env: Env): Promise<ConfiguredRepoCandidate[]> {
+  const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo]));
+  const byKey = new Map<string, ConfiguredRepoCandidate>();
+  for (const repo of repositoriesByKey.values())
+    byKey.set(repo.fullName.toLowerCase(), { fullName: repo.fullName, ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}) });
+  for (const fullName of listConvergenceRepos(env)) {
+    const repo = repositoriesByKey.get(fullName.toLowerCase());
+    byKey.set(fullName.toLowerCase(), {
+      fullName,
+      ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}),
+    });
+  }
+  return [...byKey.values()];
+}

src/review/pr-reconciliation.ts

@@ -10,17 +10,18 @@
 // enqueues no reconciliation job, byte-identical to today.
 
 import { githubRateLimitAdmissionKeyForToken } from "../github/client";
+import { resolveConfiguredRepoCandidates } from "./configured-repo-set";
 import { createInstallationToken } from "../github/app";
 import { fetchLivePullRequest, reconcileOpenPullRequests } from "../github/backfill";
-import { listRepositories, upsertPullRequestFromGitHub } from "../db/repositories";
+import { upsertPullRequestFromGitHub } from "../db/repositories";
 import { isAgentConfigured } from "../settings/autonomy";
 import { resolveRepositorySettings } from "../settings/repository-settings";
 import { loadRepoFocusManifest } from "../signals/focus-manifest-loader";
 import { resolveLoopOverSelfRepoFullName } from "../config/loopover-repo-focus-manifest";
 import { incr } from "../selfhost/metrics";
 import type { JobMessage } from "../types";
 import { errorMessage } from "../utils/json";
-import { isConvergenceRepoAllowed, listConvergenceRepos } from "./cutover-gate";
+import { isConvergenceRepoAllowed } from "./cutover-gate";
 import { deliveryIdFor } from "../queue/delivery-id";
 
 /** A manifest-sourced enable override (#6558 / #6275) -- the top-level `prReconciliation` block of the
@@ -84,19 +85,10 @@ export function clearPrReconciliationManifestOverrideCacheForTest(): void {
  *  manifest-load error fails OPEN (the repo stays watched), matching the surrounding settings-blip fail-safe
  *  below -- a config-read failure must never silently exclude a repo from monitoring. */
 async function watchedRepos(env: Env): Promise<Array<{ fullName: string; installationId?: number }>> {
-  const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo]));
-  const byKey = new Map<string, { fullName: string; installationId?: number }>();
-  for (const repo of repositoriesByKey.values())
-    byKey.set(repo.fullName.toLowerCase(), { fullName: repo.fullName, ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}) });
-  for (const fullName of listConvergenceRepos(env)) {
-    const repo = repositoriesByKey.get(fullName.toLowerCase());
-    byKey.set(fullName.toLowerCase(), {
-      fullName,
-      ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}),
-    });
-  }
+  // #10170: the repo-set assembly was duplicated five times; callers keep their own eligibility rules.
+  const repoCandidates = await resolveConfiguredRepoCandidates(env);
   const configured: Array<{ fullName: string; installationId?: number }> = [];
-  for (const repo of byKey.values()) {
+  for (const repo of repoCandidates) {
     try {
       const settings = await resolveRepositorySettings(env, repo.fullName);
       // #sweep-requires-installation: isAgentConfigured resolves the operator's global-default autonomy for

src/review/sweep-watchdog.ts

@@ -11,14 +11,15 @@
 // Default OFF (like every other *-wire-adjacent convergence capability) — flag-OFF this module is never invoked
 // and the cron enqueues no watchdog job, byte-identical to today.
 
-import { countOpenPullRequests, getLatestRegatedAt, listRepositories } from "../db/repositories";
+import { countOpenPullRequests, getLatestRegatedAt } from "../db/repositories";
+import { resolveConfiguredRepoCandidates } from "./configured-repo-set";
 import { isAgentConfigured } from "../settings/autonomy";
 import { resolveRepositorySettings } from "../settings/repository-settings";
 import { loadRepoFocusManifest } from "../signals/focus-manifest-loader";
 import { resolveLoopOverSelfRepoFullName } from "../config/loopover-repo-focus-manifest";
 import type { JobMessage } from "../types";
 import { errorMessage, nowIso } from "../utils/json";
-import { isConvergenceRepoAllowed, listConvergenceRepos } from "./cutover-gate";
+import { isConvergenceRepoAllowed } from "./cutover-gate";
 
 /** A manifest-sourced enable/threshold override (#6558 / #6275 / #6594) -- the top-level `sweepWatchdog`
  *  block of the loopover self-repo's `.loopover.yml` (see FocusManifestSweepWatchdogConfig). Distinct from the
@@ -128,19 +129,10 @@ export function isSweepStale(input: {
  *  (the repo stays watched), matching the surrounding settings-blip fail-safe below -- a config-read failure
  *  must never silently exclude a repo from monitoring. */
 async function watchedRepos(env: Env): Promise<Array<{ fullName: string; installationId?: number }>> {
-  const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo]));
-  const byKey = new Map<string, { fullName: string; installationId?: number }>();
-  for (const repo of repositoriesByKey.values())
-    byKey.set(repo.fullName.toLowerCase(), { fullName: repo.fullName, ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}) });
-  for (const fullName of listConvergenceRepos(env)) {
-    const repo = repositoriesByKey.get(fullName.toLowerCase());
-    byKey.set(fullName.toLowerCase(), {
-      fullName,
-      ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}),
-    });
-  }
+  // #10170: the repo-set assembly was duplicated five times; callers keep their own eligibility rules.
+  const repoCandidates = await resolveConfiguredRepoCandidates(env);
   const configured: Array<{ fullName: string; installationId?: number }> = [];
-  for (const repo of byKey.values()) {
+  for (const repo of repoCandidates) {
     try {
       const settings = await resolveRepositorySettings(env, repo.fullName);
       // #sweep-requires-installation: mirrors fanOutAgentRegateSweepJobs's own guard -- a repo with no real

src/services/ai-chat-qa.ts

@@ -1,4 +1,5 @@
 import { recordAiUsageEvent, recordAuditEvent, sumAiEstimatedNeuronsSince } from "../db/repositories";
+import { estimateNeurons, extractAiText } from "./ai-usage-estimate";
 import { sanitizePublicComment } from "../queue-intelligence";
 import type { AdvisoryAiRoutingConfig } from "../types";
 import type { AgentRunBundle } from "./agent-orchestrator";
@@ -130,7 +131,7 @@ export async function generateChatQaAnswer(env: Env, req: ChatQaRequest): Promis
   const model = env.WORKERS_AI_SUMMARY_MODEL || "";
   const maxOutputTokens = clampNumber(Number(env.AI_MAX_OUTPUT_TOKENS || 256), 64, 512);
   const prompt = buildChatPrompt(question, grounding);
-  const estimatedNeurons = estimateNeurons(prompt, maxOutputTokens);
+  const estimatedNeurons = estimateNeurons(prompt.length, maxOutputTokens);
   // Shared daily neuron budget: the SAME counter every AI feature sums into (ai-review / ai-slop / ai-summaries,
   // #1369). Default HIGH (10M) and clamp to 10M so chat Q&A never starves — or is starved by — the shared pool.
   const rawNeuronBudget = Number(env.AI_DAILY_NEURON_BUDGET);
@@ -155,7 +156,14 @@ export async function generateChatQaAnswer(env: Env, req: ChatQaRequest): Promis
     // without masking a real, persistent failure: if it's STILL empty on the second try, the loop falls
     // through with rawText === "" and the check below throws exactly as it always did.
     let rawText = "";
+    // #10169: count the calls actually made. `estimatedNeurons` above is the PRE-FLIGHT figure (one call, all
+    // that is knowable before the first attempt) and is what the budget gate compares -- but the same value
+    // was also being RECORDED, so a retry spent two calls and reported one into the very counter
+    // sumAiEstimatedNeuronsSince adds up as the shared daily budget. Under-reporting there is the wrong
+    // direction: it hides spend from the backstop meant to notice runaway usage.
+    let providerCalls = 0;
     for (let attempt = 0; attempt < 2 && !rawText; attempt += 1) {
+      providerCalls += 1;
       const response = await ai.run(model, {
         messages: [
           { role: "system", content: CHAT_QA_SYSTEM_PROMPT },
@@ -167,11 +175,13 @@ export async function generateChatQaAnswer(env: Env, req: ChatQaRequest): Promis
       rawText = extractAiText(response) ?? "";
     }
     if (!rawText) throw new Error("empty_chat_answer");
+    // Recorded spend, as opposed to the pre-flight estimate above.
+    const spentNeurons = estimateNeurons(prompt.length, maxOutputTokens, providerCalls);
     if (containsPublicForbiddenText(rawText)) {
-      await recordChatAi(env, req, { model, status: "unsafe", estimatedNeurons, detail: "chat answer failed public sanitizer", usedFrontier });
+      await recordChatAi(env, req, { model, status: "unsafe", estimatedNeurons: spentNeurons, detail: "chat answer failed public sanitizer", usedFrontier });
       return { status: "unsafe", model, estimatedNeurons, reason: "chat answer failed public sanitizer" };
     }
-    await recordChatAi(env, req, { model, status: "ok", estimatedNeurons, detail: "chat answer generated", usedFrontier });
+    await recordChatAi(env, req, { model, status: "ok", estimatedNeurons: spentNeurons, detail: "chat answer generated", usedFrontier });
     return { status: "ok", model, estimatedNeurons, text: rawText.trim() };
   } catch (error) {
     const reason = error instanceof Error ? error.message : "chat_answer_failed";
@@ -222,20 +232,7 @@ function containsPublicForbiddenText(value: string): boolean {
   return PUBLIC_FORBIDDEN_TEXT_PATTERN.test(value);
 }
 
-function estimateNeurons(prompt: string, maxOutputTokens: number): number {
-  const inputTokens = Math.ceil(prompt.length / 4);
-  return Math.max(1, Math.ceil((inputTokens + maxOutputTokens) * 0.035));
-}
 
-function extractAiText(response: unknown): string {
-  if (typeof response === "string") return response;
-  if (!response || typeof response !== "object") return "";
-  const record = response as Record<string, unknown>;
-  if (typeof record.response === "string") return record.response;
-  if (typeof record.text === "string") return record.text;
-  if (typeof record.result === "string") return record.result;
-  return "";
-}
 
 function clampNumber(value: number, min: number, max: number): number {
   if (!Number.isFinite(value)) return min;

src/services/ai-intent-router.ts

@@ -1,4 +1,5 @@
 import { recordAiUsageEvent, recordAuditEvent, sumAiEstimatedNeuronsSince } from "../db/repositories";
+import { estimateNeurons, extractAiText } from "./ai-usage-estimate";
 import { INTENT_ROUTABLE_COMMANDS, isIntentRoutableCommand, type IntentRoutableCommandName } from "../github/commands";
 import type { AdvisoryAiRoutingConfig } from "../types";
 
@@ -61,7 +62,7 @@ export async function classifyLoopOverIntent(env: Env, req: IntentRoutingRequest
   const model = env.WORKERS_AI_SUMMARY_MODEL || "";
   const maxOutputTokens = 32; // the entire valid output is a ~20-char JSON object; no legitimate reason to allow more
   const prompt = `Contributor message: ${text}`;
-  const estimatedNeurons = estimateNeurons(prompt, maxOutputTokens);
+  const estimatedNeurons = estimateNeurons(prompt.length, maxOutputTokens);
   // Shared daily neuron budget: the SAME counter every AI feature sums into (ai-review / ai-slop / ai-summaries /
   // ai-chat-qa, #1369). Default HIGH (10M) and clamp to 10M so intent routing never starves -- or is starved by --
   // the shared pool.
@@ -121,20 +122,7 @@ function extractCommandCandidate(rawText: string): unknown {
   }
 }
 
-function estimateNeurons(prompt: string, maxOutputTokens: number): number {
-  const inputTokens = Math.ceil(prompt.length / 4);
-  return Math.max(1, Math.ceil((inputTokens + maxOutputTokens) * 0.035));
-}
 
-function extractAiText(response: unknown): string {
-  if (typeof response === "string") return response;
-  if (!response || typeof response !== "object") return "";
-  const record = response as Record<string, unknown>;
-  if (typeof record.response === "string") return record.response;
-  if (typeof record.text === "string") return record.text;
-  if (typeof record.result === "string") return record.result;
-  return "";
-}
 
 function clampNumber(value: number, min: number, max: number): number {
   if (!Number.isFinite(value)) return min;