From 6cf1c6d6643b5c8124d72c8d63f960fc8666ad16 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Fri, 31 Jul 2026 03:45:02 -0700 Subject: [PATCH 1/2] fix(cost): one estimateNeurons, and record the calls actually made estimateNeurons existed four times and extractAiText three, byte-identical copies that had already drifted: ai-review.ts exports a version taking a `calls` multiplier, while ai-chat-qa, ai-summaries and ai-intent-router each kept a private copy without it. That is not just duplication. `estimatedNeurons` is BOTH the pre-flight budget gate and the value recorded into ai_usage_events, which sumAiEstimatedNeuronsSince adds up as the shared daily neuron budget. ai-chat-qa retries its provider call on an empty completion, so a retry spent two calls and reported one -- hiding usage from the backstop whose job is to notice runaway usage, in the one direction that matters. Both helpers move to a leaf module both sides import. Deliberately not imported from ai-review.ts: that file is 3.6k lines, and three small services should not depend on the review engine to divide a number by four. The chat surface's import-isolation contract (#4595 req 10) still holds -- the new module reaches no write path -- and its test still passes. The pre-flight gate keeps the single-call estimate, which is all that is knowable before the first attempt. The RECORDED figure is now computed after the loop from the calls actually made, so the two no longer pretend to be the same number. `calls` floors at 1, so a miscount can never report zero spend for work that happened. Found by the maintainability audit (#10170). Closes #10169 --- src/services/ai-chat-qa.ts | 29 +++++++-------- src/services/ai-intent-router.ts | 16 ++------- src/services/ai-summaries.ts | 18 ++-------- src/services/ai-usage-estimate.ts | 41 +++++++++++++++++++++ test/unit/ai-chat-qa.test.ts | 4 +-- test/unit/ai-intent-router.test.ts | 4 +-- test/unit/ai-summaries.test.ts | 2 +- test/unit/ai-usage-estimate.test.ts | 56 +++++++++++++++++++++++++++++ 8 files changed, 120 insertions(+), 50 deletions(-) create mode 100644 src/services/ai-usage-estimate.ts create mode 100644 test/unit/ai-usage-estimate.test.ts diff --git a/src/services/ai-chat-qa.ts b/src/services/ai-chat-qa.ts index 59d587a6b3..c14c98291e 100644 --- a/src/services/ai-chat-qa.ts +++ b/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; - 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; diff --git a/src/services/ai-intent-router.ts b/src/services/ai-intent-router.ts index d2a0a10af3..eba4f81a9a 100644 --- a/src/services/ai-intent-router.ts +++ b/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; - 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; diff --git a/src/services/ai-summaries.ts b/src/services/ai-summaries.ts index f6c11b2303..c3e92084f8 100644 --- a/src/services/ai-summaries.ts +++ b/src/services/ai-summaries.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 { JsonValue } from "../types"; import type { AgentRunBundle } from "./agent-orchestrator"; @@ -35,7 +36,7 @@ export async function summarizeAgentBundleWithAi(env: Env, bundle: AgentRunBundl const maxOutputTokens = clampNumber(Number(env.AI_MAX_OUTPUT_TOKENS || 256), 64, 512); const signalBundle = compactAgentSignalBundle(bundle, visibility); const prompt = buildPrompt(signalBundle, visibility); - const estimatedNeurons = estimateNeurons(prompt, maxOutputTokens); + const estimatedNeurons = estimateNeurons(prompt.length, maxOutputTokens); // Resolve the SHARED daily neuron budget exactly like ai-review.ts / ai-slop.ts (#1369): all three // AI features sum into one `sumAiEstimatedNeuronsSince` counter, so the old `|| 10000` default + // 1M ceiling here starved summaries into quota_exceeded once shared usage crossed 10k — well under the @@ -162,20 +163,7 @@ function buildPrompt(signalBundle: Record, visibility: AiSumm ].join("\n"); } -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; - 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 sanitizeAiText(value: string, visibility: AiSummaryVisibility): string { const sanitized = value @@ -302,7 +290,7 @@ export async function rewriteSignalBundleWithAi(env: Env, req: AiRewriteRequest) const model = env.WORKERS_AI_SUMMARY_MODEL || ""; const maxOutputTokens = clampNumber(Number(env.AI_MAX_OUTPUT_TOKENS || 256), 64, 512); const prompt = buildBundlePrompt(req.bundle, req.visibility); - const estimatedNeurons = estimateNeurons(prompt, maxOutputTokens); + const estimatedNeurons = estimateNeurons(prompt.length, maxOutputTokens); // Resolve the SHARED daily neuron budget exactly like ai-review.ts / ai-slop.ts (#1369): all three // AI features sum into one `sumAiEstimatedNeuronsSince` counter, so the old `|| 10000` default + // 1M ceiling here starved summaries into quota_exceeded once shared usage crossed 10k — well under the diff --git a/src/services/ai-usage-estimate.ts b/src/services/ai-usage-estimate.ts new file mode 100644 index 0000000000..ff0367316f --- /dev/null +++ b/src/services/ai-usage-estimate.ts @@ -0,0 +1,41 @@ +// Shared AI-call usage estimation and response extraction (#10169). +// +// Both helpers below existed as private copies in several services -- estimateNeurons in four places, +// extractAiText in three, byte-identical to each other. They had already drifted in the way copy-paste always +// does: ai-review.ts's copy grew a `calls` multiplier when it needed one, and the three copies that had been +// pasted earlier did not follow. +// +// That drift was not cosmetic. `estimatedNeurons` is BOTH the pre-flight budget check and the value recorded +// into ai_usage_events, which sumAiEstimatedNeuronsSince adds up as the shared daily neuron budget -- so a +// service that retried its provider call reported half the neurons it actually spent, into the counter meant +// to notice exactly that. See ai-chat-qa's retry loop. +// +// Kept in its own leaf module rather than imported from ai-review.ts: that file is 3.6k lines, and three +// small services should not take a dependency on the review engine to divide a number by four. + +/** + * PURE. Rough neuron cost of an AI call, or of `calls` identical ones. + * + * Chars/4 is the usual token approximation and 0.035 the Workers-AI neuron factor; both are estimates by + * construction -- the point is a consistent, comparable number across every feature that shares the budget, + * not accuracy against a bill. + * + * `calls` is the parameter the private copies dropped. Passing the ACTUAL number of provider calls made is + * what keeps the recorded figure honest when a service retries. + */ +export function estimateNeurons(promptChars: number, maxOutputTokens: number, calls = 1): number { + const inputTokens = Math.ceil(promptChars / 4); + return Math.max(1, Math.ceil((inputTokens + maxOutputTokens) * 0.035) * Math.max(1, calls)); +} + +/** Pull usable text out of a provider response whose shape varies by provider/model. Fail-soft: anything + * unrecognised yields "" so the caller's own empty-answer handling decides, rather than throwing here. */ +export function extractAiText(response: unknown): string { + if (typeof response === "string") return response; + if (!response || typeof response !== "object") return ""; + const record = response as Record; + 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 ""; +} diff --git a/test/unit/ai-chat-qa.test.ts b/test/unit/ai-chat-qa.test.ts index 63f07564da..39dee0d86a 100644 --- a/test/unit/ai-chat-qa.test.ts +++ b/test/unit/ai-chat-qa.test.ts @@ -417,8 +417,8 @@ describe("__chatQaInternals", () => { }); it("estimates neurons from prompt length and output tokens, with a floor of 1", () => { - expect(estimateNeurons("a".repeat(400), 256)).toBe(13); - expect(estimateNeurons("", 0)).toBe(1); + expect(estimateNeurons("a".repeat(400).length, 256)).toBe(13); + expect(estimateNeurons("".length, 0)).toBe(1); }); it("extracts text from every recognized response shape and falls back to empty otherwise", () => { diff --git a/test/unit/ai-intent-router.test.ts b/test/unit/ai-intent-router.test.ts index 64edacde27..cfd1c56339 100644 --- a/test/unit/ai-intent-router.test.ts +++ b/test/unit/ai-intent-router.test.ts @@ -189,8 +189,8 @@ describe("__intentRouterInternals", () => { }); it("estimates neurons from prompt length and output tokens, with a floor of 1", () => { - expect(estimateNeurons("a".repeat(400), 32)).toBeGreaterThanOrEqual(1); - expect(estimateNeurons("", 0)).toBe(1); + expect(estimateNeurons("a".repeat(400).length, 32)).toBeGreaterThanOrEqual(1); + expect(estimateNeurons("".length, 0)).toBe(1); }); it("extracts text from every recognized response shape and falls back to empty otherwise", () => { diff --git a/test/unit/ai-summaries.test.ts b/test/unit/ai-summaries.test.ts index adce687bc6..fff40fac05 100644 --- a/test/unit/ai-summaries.test.ts +++ b/test/unit/ai-summaries.test.ts @@ -276,7 +276,7 @@ describe("Workers AI summaries", () => { expect(__aiSummaryInternals.extractAiText({ result: "result" })).toBe("result"); expect(__aiSummaryInternals.extractAiText({ nope: 1 })).toBe(""); expect(__aiSummaryInternals.extractAiText(null)).toBe(""); - expect(__aiSummaryInternals.estimateNeurons("abcd".repeat(100), 128)).toBeGreaterThan(0); + expect(__aiSummaryInternals.estimateNeurons("abcd".repeat(100).length, 128)).toBeGreaterThan(0); expect(__aiSummaryInternals.sanitizeAiText("wallet hotkey payout", "public")).not.toMatch(/wallet|hotkey|payout/i); expect(__aiSummaryInternals.containsPublicForbiddenText("raw trust score")).toBe(true); expect(__aiSummaryInternals.compactAgentSignalBundle(bundleFixture(), "public").actions).toHaveLength(1); diff --git a/test/unit/ai-usage-estimate.test.ts b/test/unit/ai-usage-estimate.test.ts new file mode 100644 index 0000000000..6041270f02 --- /dev/null +++ b/test/unit/ai-usage-estimate.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from "vitest"; + +import { estimateNeurons, extractAiText } from "../../src/services/ai-usage-estimate"; + +// #10169: estimateNeurons existed in four places and extractAiText in three, byte-identical copies that had +// already drifted -- ai-review.ts's grew a `calls` multiplier when it needed one, and the three pasted +// earlier did not follow. +// +// The drift had teeth. `estimatedNeurons` is BOTH the pre-flight budget gate and the value recorded into +// ai_usage_events, which sumAiEstimatedNeuronsSince adds up as the shared daily neuron budget. A service that +// retried its provider call reported one call's neurons for two calls' spend -- hiding usage from the +// backstop whose entire job is to notice runaway usage. + +describe("estimateNeurons (#10169)", () => { + it("matches the formula every caller previously reimplemented", () => { + // 400 chars -> 100 input tokens; (100 + 256) * 0.035 = 12.46 -> 13. + expect(estimateNeurons(400, 256)).toBe(13); + }); + + it("never returns 0 — a call that happened must cost something", () => { + expect(estimateNeurons(0, 0)).toBe(1); + }); + + it("REGRESSION: scales with the number of calls actually made", () => { + // The parameter the private copies dropped. Without it a retry is invisible to the budget. + expect(estimateNeurons(400, 256, 2)).toBe(26); + expect(estimateNeurons(400, 256, 3)).toBe(39); + }); + + it("defaults to a single call, so the un-retried path is unchanged", () => { + expect(estimateNeurons(400, 256)).toBe(estimateNeurons(400, 256, 1)); + }); + + it("treats 0 or negative calls as one — a floor, never a way to report zero spend", () => { + // Guards the direction that matters: under-reporting is what hides spend. + expect(estimateNeurons(400, 256, 0)).toBe(13); + expect(estimateNeurons(400, 256, -5)).toBe(13); + }); +}); + +describe("extractAiText", () => { + it("reads the shapes different providers return", () => { + expect(extractAiText("plain")).toBe("plain"); + expect(extractAiText({ response: "r" })).toBe("r"); + expect(extractAiText({ text: "t" })).toBe("t"); + expect(extractAiText({ result: "x" })).toBe("x"); + }); + + it("FAIL-SOFT: anything unrecognised is empty, never a throw", () => { + // The caller's own empty-answer handling decides; throwing here would turn a odd-shaped response into a + // failed review. + for (const value of [null, undefined, 42, {}, { other: "no" }, []]) { + expect(extractAiText(value), JSON.stringify(value)).toBe(""); + } + }); +}); From 9e3fba45c31fdaa6a3d1d16b759149ec889e9332 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Fri, 31 Jul 2026 03:51:48 -0700 Subject: [PATCH 2/2] refactor(review): one configured-repo-set resolver instead of five copies Assembling "which repos does a fleet-wide pass consider" 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 and every one of them rebuilt it. Only the ASSEMBLY moves. What each caller does next -- resolving settings, requiring a real installation, applying its own eligibility rule -- genuinely differs and stays at the call site. Extracting further would push those differences into flag parameters, which is how a shared helper ends up worse than the duplication it replaced. The merge semantics were load-bearing and untested: the convergence list wins on name, but carries over a local row's installationId when one exists. Losing that would make an installed repo look uninstalled, and every caller that requires an installation would skip it -- a fleet-wide sweep quietly doing nothing. Now covered directly, including the case-insensitive de-duplication and the omitted-not-null installationId shape the call sites' spreads depend on. processors.ts drops from 16,913 to 16,886 lines. Small, but the point is the four remaining copies that can no longer drift. Refs #10170 --- src/queue/processors.ts | 47 +++++---------------- src/review/configured-repo-set.ts | 44 +++++++++++++++++++ src/review/pr-reconciliation.ts | 20 +++------ src/review/sweep-watchdog.ts | 20 +++------ test/unit/configured-repo-set.test.ts | 61 +++++++++++++++++++++++++++ 5 files changed, 127 insertions(+), 65 deletions(-) create mode 100644 src/review/configured-repo-set.ts create mode 100644 test/unit/configured-repo-set.test.ts diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 353d7bd2e9..88a7e0563b 100644 --- a/src/queue/processors.ts +++ b/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(); - 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 => { 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(); - 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(); - 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 => { const repoFullName = repo.fullName; diff --git a/src/review/configured-repo-set.ts b/src/review/configured-repo-set.ts new file mode 100644 index 0000000000..11e2699192 --- /dev/null +++ b/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 { + const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo])); + const byKey = new Map(); + 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()]; +} diff --git a/src/review/pr-reconciliation.ts b/src/review/pr-reconciliation.ts index e18c277214..bc6196c12e 100644 --- a/src/review/pr-reconciliation.ts +++ b/src/review/pr-reconciliation.ts @@ -10,9 +10,10 @@ // 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"; @@ -20,7 +21,7 @@ import { resolveLoopOverSelfRepoFullName } from "../config/loopover-repo-focus-m 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> { - const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo])); - const byKey = new Map(); - 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 diff --git a/src/review/sweep-watchdog.ts b/src/review/sweep-watchdog.ts index f1e6a19340..5cf796ea52 100644 --- a/src/review/sweep-watchdog.ts +++ b/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> { - const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo])); - const byKey = new Map(); - 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 diff --git a/test/unit/configured-repo-set.test.ts b/test/unit/configured-repo-set.test.ts new file mode 100644 index 0000000000..20f45e3509 --- /dev/null +++ b/test/unit/configured-repo-set.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it, vi } from "vitest"; + +// #10170: the repo-set assembly was duplicated five times (three inside processors.ts) and, being inline in +// each caller, was never tested on its own -- only incidentally, through whichever sweep happened to cover it. +// Its merge semantics are load-bearing: get them wrong and a fleet-wide pass silently skips repos, or treats +// an uninstalled one as installed. + +vi.mock("../../src/db/repositories", () => ({ listRepositories: vi.fn() })); +vi.mock("../../src/review/cutover-gate", () => ({ listConvergenceRepos: vi.fn() })); + +const { listRepositories } = await import("../../src/db/repositories"); +const { listConvergenceRepos } = await import("../../src/review/cutover-gate"); +const { resolveConfiguredRepoCandidates } = await import("../../src/review/configured-repo-set"); + +const env = {} as Env; +const setup = (rows: unknown[], convergence: string[]) => { + vi.mocked(listRepositories).mockResolvedValue(rows as never); + vi.mocked(listConvergenceRepos).mockReturnValue(convergence as never); +}; + +describe("resolveConfiguredRepoCandidates (#10170)", () => { + it("returns locally-known repos, carrying installationId", async () => { + setup([{ fullName: "acme/widgets", installationId: 42 }], []); + expect(await resolveConfiguredRepoCandidates(env)).toEqual([{ fullName: "acme/widgets", installationId: 42 }]); + }); + + it("includes a convergence repo that has no local row", async () => { + // The whole reason the two sources are merged: an allowlisted repo is in scope before it is ever synced. + setup([], ["acme/newrepo"]); + expect(await resolveConfiguredRepoCandidates(env)).toEqual([{ fullName: "acme/newrepo" }]); + }); + + it("OMITS installationId rather than setting it null when unknown", async () => { + // Every call site spreads this under exactOptionalPropertyTypes, and callers gate on + // `typeof repo.installationId === "number"`. A null would satisfy neither. + setup([], ["acme/newrepo"]); + const [repo] = await resolveConfiguredRepoCandidates(env); + expect("installationId" in (repo ?? {})).toBe(false); + }); + + it("REGRESSION: a convergence entry carries over the local row's installationId", async () => { + // The convergence list holds only names. Losing the installationId here would make a genuinely installed + // repo look uninstalled, and every caller that requires a real installation would skip it -- a + // fleet-wide sweep quietly doing nothing. + setup([{ fullName: "acme/widgets", installationId: 7 }], ["acme/widgets"]); + expect(await resolveConfiguredRepoCandidates(env)).toEqual([{ fullName: "acme/widgets", installationId: 7 }]); + }); + + it("de-duplicates case-insensitively, and the convergence spelling wins", async () => { + // GitHub full names are case-insensitive; two entries for one repo would double every per-repo read. + setup([{ fullName: "Acme/Widgets", installationId: 7 }], ["acme/widgets"]); + const out = await resolveConfiguredRepoCandidates(env); + expect(out).toHaveLength(1); + expect(out[0]).toEqual({ fullName: "acme/widgets", installationId: 7 }); + }); + + it("is empty when both sources are", async () => { + setup([], []); + expect(await resolveConfiguredRepoCandidates(env)).toEqual([]); + }); +});