diff --git a/src/env.d.ts b/src/env.d.ts index 794f644ca1..9d8be57d7f 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -247,6 +247,12 @@ declare global { * for just that repo. Default OFF — unset/false means the cron tick enqueues NO watchdog job (does no new * work), so the worker is byte-identical to today. */ GITTENSORY_SWEEP_WATCHDOG?: string; + /** Self-heal: when truthy, a short-interval cron list-diffs GitHub's open PR numbers against the local + * table for every acting-autonomy repo and catches up (fetch + upsert + regate) any PR number GitHub has + * that the local table doesn't — a silently-lost "opened" webhook, caught within minutes instead of the + * 6-hour backfillRegisteredRepositories freshness window. Default OFF — unset/false means the cron tick + * enqueues NO reconciliation job, so the worker is byte-identical to today. */ + GITTENSORY_PR_RECONCILIATION?: string; /** Convergence (RAG retrieval): when truthy, the AI reviewer prompt gains a RELEVANT EXISTING CODE / DOCS * section — at review time the codebase vector index is queried for code/docs semantically related to the * PR's changed files (callers, related modules, existing conventions) and appended as additive reference diff --git a/src/github/backfill.ts b/src/github/backfill.ts index 0082b692ad..0d8d5b92f2 100644 --- a/src/github/backfill.ts +++ b/src/github/backfill.ts @@ -3132,6 +3132,52 @@ export async function fetchLivePullRequest( return result.status === "ok" ? result.data : undefined; } +// Bounded at 1000 open PRs (10 pages of 100) — far beyond any real repo's open-PR count; a pathological repo +// can't spin this loop forever. +const RECONCILE_OPEN_PRS_MAX_PAGES = 10; + +export type OpenPrReconciliationResult = { + repoFullName: string; + remoteOpenCount: number; + localOpenCount: number; + /** PR numbers GitHub reports open that have NO local pull_requests row at all — a webhook silently lost. */ + missingNumbers: number[]; +}; + +/** + * Fast open-PR reconciliation (#audit-open-pr-reconciliation): a CHEAP list-only GitHub read (PR numbers only, + * via the Link-header-paginated `/pulls?state=open` list — never a per-PR detail fetch) diffed against the + * local `pull_requests` table. Exists to catch a silently-lost "PR opened" webhook (#3782/#3793, 2026-07-06: + * two contributor PRs had zero trace anywhere for ~2 hours) within minutes, instead of waiting on + * `backfillRegisteredRepositories`'s 6-hour freshness window. + * + * FAIL-OPEN: a total read failure (repo not found, or the very first list page erroring) returns an all-zero, + * empty-`missingNumbers` result — it must never falsely report every local PR as "missing" just because GitHub + * was briefly unreachable. A LATER page failing mid-crawl keeps the pages already fetched (a partial remote + * list can only under-report divergence, never wrongly flag a real local PR as missing). + */ +export async function reconcileOpenPullRequests(env: Env, repoFullName: string): Promise { + const empty: OpenPrReconciliationResult = { repoFullName, remoteOpenCount: 0, localOpenCount: 0, missingNumbers: [] }; + const repo = await getRepository(env, repoFullName); + if (!repo) return empty; + const token = await tokenForRepo(env, repo); + const admissionKey = repoAdmissionKeyForToken(env, repo, token); + const remoteNumbers: number[] = []; + for (let page = 1; page <= RECONCILE_OPEN_PRS_MAX_PAGES; page += 1) { + const result = await githubJsonWithHeaders>(env, repoFullName, `/pulls?state=open&per_page=100&page=${page}`, token, githubRateLimitOptions(admissionKey)).catch(() => undefined); + if (!result) { + if (page === 1) return empty; + break; + } + remoteNumbers.push(...result.data.map((pr) => pr.number)); + if (!hasNextPage(result.link)) break; + } + const localOpen = await listOpenPullRequests(env, repoFullName); + const localNumbers = new Set(localOpen.map((pr) => pr.number)); + const missingNumbers = remoteNumbers.filter((number) => !localNumbers.has(number)); + return { repoFullName, remoteOpenCount: remoteNumbers.length, localOpenCount: localOpen.length, missingNumbers }; +} + // #2537: durable, webhook-invalidated cache for the bare PR-state read (GET /pulls/{n}). Unlike the request-local // LiveGithubFacts memo (queue/processors.ts), this survives ACROSS webhook deliveries / sweep ticks, cutting // repeat /pulls/{n} calls for an unchanged PR at the freshness-guard/readiness/dup-winner call sites. NEVER used diff --git a/src/index.ts b/src/index.ts index f264b4ebdf..d1c5d791f2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -6,6 +6,7 @@ import { processJob } from "./queue/processors"; import { isOrbBrokerEnabled } from "./orb/broker"; import { isOpsEnabled } from "./review/ops-wire"; import { isSweepWatchdogEnabled } from "./review/sweep-watchdog"; +import { isPrReconciliationEnabled } from "./review/pr-reconciliation"; import { isRagEnabled } from "./review/rag-wire"; import { isSelfTuneEnabled } from "./review/selftune-wire"; import { @@ -106,6 +107,10 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController): const hour = scheduledAt.getUTCHours(); const isHourly = minute === 0; const isFullSyncWindow = isHourly && hour % 6 === 0; + // Self-heal (#audit-open-pr-reconciliation): every 10 minutes — much tighter than the hourly ops-alerts/ + // selftune cadence, since this exists specifically to catch a silently-lost webhook within minutes rather + // than up to 6 hours (backfillRegisteredRepositories's freshness window). + const isReconciliationWindow = minute % 10 === 0; // The light auto-maintain sweep runs EVERY cron tick (~every 2 min) so an approved+clean PR MERGES and a // red-CI non-owner PR CLOSES promptly — reviewbot parity (its cron fired every minute). It re-fetches LIVE CI + // mergeable and only ACTS (merge/close/hold); it never re-runs the AI, so it is cheap enough for this cadence. @@ -171,6 +176,10 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController): // classified (MAINTENANCE_JOB_TYPES) so it defers under live-work pressure like every other periodic sweep here. if (selfHostedReviews) jobs.push({ type: "backlog-convergence-sweep", requestedBy: "schedule" }); } + // Self-heal (flag GITTENSORY_PR_RECONCILIATION). Every 10 minutes — see isReconciliationWindow above. + // Enqueued ONLY when the flag is ON — flag-OFF (default) this job is never created, so the cron tick does + // ZERO new work and the enqueued set is byte-identical to today. + if (selfHostedReviews && isReconciliationWindow && isPrReconciliationEnabled(env)) jobs.push({ type: "reconcile-open-prs", requestedBy: "schedule" }); if (isHourly) { jobs.push({ type: "refresh-registry", requestedBy: "schedule" }); jobs.push({ type: "refresh-scoring-model", requestedBy: "schedule" }); diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 0dc9401a17..9c45b21dfa 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -493,6 +493,7 @@ import { resolveUnlinkedIssueMatchDisposition } from "../review/unlinked-issue-g import { DEFAULT_SCREENSHOT_TABLE_GATE, evaluateScreenshotTableGate } from "../review/screenshot-table-gate"; import { isOpsEnabled, runOpsAlerts } from "../review/ops-wire"; import { isSweepWatchdogEnabled, runSweepLivenessWatchdog } from "../review/sweep-watchdog"; +import { isPrReconciliationEnabled, runOpenPrReconciliation } from "../review/pr-reconciliation"; import { isSelfTuneEnabled, runSelfTune } from "../review/selftune-wire"; import { isCloseHoldOnly, @@ -1119,6 +1120,12 @@ export async function processJob(env: Env, message: JobMessage): Promise { // work here too. Fails safe internally — never throws into the queue. if (isSweepWatchdogEnabled(env)) await runSweepLivenessWatchdog(env); return; + case "reconcile-open-prs": + // Self-heal (flag GITTENSORY_PR_RECONCILIATION). Defense-in-depth: the cron only ENQUEUES this when the + // flag is ON, but a stale in-flight job that lands after a flag-flip must still no-op, so flag-OFF does + // zero work here too. Fails safe internally — never throws into the queue. + if (isPrReconciliationEnabled(env)) await runOpenPrReconciliation(env); + return; case "selftune": // Convergence (self-improve / auto-tune, flag GITTENSORY_REVIEW_SELFTUNE). Defense-in-depth: the cron only // ENQUEUES this when the flag is ON, but a stale in-flight job that lands after a flag-flip must still diff --git a/src/review/pr-reconciliation.ts b/src/review/pr-reconciliation.ts new file mode 100644 index 0000000000..da9cced3f7 --- /dev/null +++ b/src/review/pr-reconciliation.ts @@ -0,0 +1,124 @@ +// Self-heal (flag-gated by GITTENSORY_PR_RECONCILIATION). A "PR opened" webhook that silently vanishes (no +// error, no audit event, no trace anywhere — see reconcileOpenPullRequests's doc comment for the 2026-07-06 +// incident that motivated this) previously wasn't caught until backfillRegisteredRepositories's opportunistic +// resync, which skips any repo whose sync is "fresh" (up to 6 hours). This module runs a much tighter, dedicated +// reconciliation on its own short cron cadence: list-diff GitHub's open PR numbers against the local table for +// every acting-autonomy repo, and for any number GitHub has that the local table doesn't, immediately catch it +// up (fetch full details, upsert, and enqueue a normal regate — the SAME pipeline a real webhook would have fed). +// +// Default OFF (like every other convergence capability) — flag-OFF this module is never invoked and the cron +// enqueues no reconciliation job, byte-identical to today. + +import { githubRateLimitAdmissionKeyForToken } from "../github/client"; +import { createInstallationToken } from "../github/app"; +import { fetchLivePullRequest, reconcileOpenPullRequests } from "../github/backfill"; +import { getRepository, listRepositories, upsertPullRequestFromGitHub } from "../db/repositories"; +import { isAgentConfigured } from "../settings/autonomy"; +import { resolveRepositorySettings } from "../settings/repository-settings"; +import { incr } from "../selfhost/metrics"; +import type { JobMessage } from "../types"; +import { errorMessage } from "../utils/json"; +import { isConvergenceRepoAllowed, listConvergenceRepos } from "./cutover-gate"; + +/** True when fast open-PR reconciliation is enabled. Flag-OFF (default) → the caller never invokes it, so the + * cron enqueues no reconciliation job and the queue processor no-ops on a stale in-flight one. */ +export function isPrReconciliationEnabled(env: { GITTENSORY_PR_RECONCILIATION?: string | undefined }): boolean { + return /^(1|true|yes|on)$/i.test(env.GITTENSORY_PR_RECONCILIATION ?? ""); +} + +/** The same acting-autonomy repo set fanOutAgentRegateSweepJobs sweeps (mirrors sweep-watchdog.ts's own copy of + * this selection) — this reconciliation only makes sense for repos gittensory is actually reviewing. */ +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 } : {}), + }); + } + const configured: Array<{ fullName: string; installationId?: number }> = []; + for (const repo of byKey.values()) { + try { + const settings = await resolveRepositorySettings(env, repo.fullName); + if (isConvergenceRepoAllowed(env, repo.fullName) || isAgentConfigured(settings.autonomy)) configured.push(repo); + } catch { + /* a settings blip on one repo must not abort the whole reconciliation scan */ + } + } + return configured; +} + +/** Catch up ONE missing PR number: fetch its full live payload, upsert it into the local table, and enqueue a + * normal regate — the exact pipeline a real "PR opened" webhook would have fed it into, so it gets reviewed, + * labeled, and gated exactly like any other PR. Best-effort: a fetch/upsert failure is logged and skipped (the + * next reconciliation tick retries it; it is not lost by this catch-up failing). */ +async function catchUpMissingPullRequest(env: Env, repoFullName: string, installationId: number, prNumber: number): Promise { + try { + const token = (await createInstallationToken(env, installationId).catch(() => undefined)) ?? env.GITHUB_PUBLIC_TOKEN; + const admissionKey = githubRateLimitAdmissionKeyForToken(env, token, installationId); + const live = await fetchLivePullRequest(env, repoFullName, prNumber, token, admissionKey); + if (!live) { + console.error(JSON.stringify({ level: "error", event: "open_pr_reconciliation_catch_up_fetch_failed", repository: repoFullName, prNumber })); + return; + } + await upsertPullRequestFromGitHub(env, repoFullName, live); + const message: JobMessage = { type: "agent-regate-pr", deliveryId: `reconcile:${repoFullName}#${prNumber}`, repoFullName, prNumber, installationId }; + await env.JOBS.send(message); + } catch (error) { + console.error(JSON.stringify({ level: "error", event: "open_pr_reconciliation_catch_up_failed", repository: repoFullName, prNumber, message: errorMessage(error).slice(0, 200) })); + } +} + +export interface OpenPrDivergence { + repoFullName: string; + remoteOpenCount: number; + localOpenCount: number; + missingNumbers: number[]; +} + +/** + * The reconciliation scan, run on the cron tick. FAILS SAFE: a per-repo error is logged and the scan continues; + * a top-level error is swallowed (this is best-effort self-heal, never a reason to fail the queue). Only an + * INSTALLED repo (installationId present) is reconciled — a registered-but-uninstalled repo never gets a per-PR + * fan-out regardless (#sweep-uninstalled-budget-waste), so reconciling it would just spend the shared + * GITHUB_PUBLIC_TOKEN budget with no actionable outcome. Returns the divergences found (for tests / a caller + * that wants to inspect them further). + * + * Caller MUST gate this on {@link isPrReconciliationEnabled} — it is invoked only from the flag-ON cron path, so + * flag-OFF this function is never reached and the cron does zero new work. + */ +export async function runOpenPrReconciliation(env: Env): Promise { + const found: OpenPrDivergence[] = []; + try { + const repos = await watchedRepos(env); + for (const repo of repos) { + try { + if (typeof repo.installationId !== "number") continue; + const result = await reconcileOpenPullRequests(env, repo.fullName); + if (result.missingNumbers.length === 0) continue; + found.push({ repoFullName: repo.fullName, remoteOpenCount: result.remoteOpenCount, localOpenCount: result.localOpenCount, missingNumbers: result.missingNumbers }); + incr("gittensory_open_pr_reconciliation_missing_total", { repo: repo.fullName }, result.missingNumbers.length); + console.error( + JSON.stringify({ + level: "error", + event: "open_pr_reconciliation_divergence", + repository: repo.fullName, + remoteOpenCount: result.remoteOpenCount, + localOpenCount: result.localOpenCount, + missingNumbers: result.missingNumbers, + }), + ); + for (const prNumber of result.missingNumbers) await catchUpMissingPullRequest(env, repo.fullName, repo.installationId, prNumber); + } catch (error) { + console.error(JSON.stringify({ level: "error", event: "open_pr_reconciliation_repo_error", repository: repo.fullName, message: errorMessage(error).slice(0, 200) })); + } + } + } catch (error) { + console.error(JSON.stringify({ level: "error", event: "open_pr_reconciliation_error", message: errorMessage(error).slice(0, 200) })); + } + return found; +} diff --git a/src/selfhost/maintenance-admission.ts b/src/selfhost/maintenance-admission.ts index c4b20b2a08..caa07a38f4 100644 --- a/src/selfhost/maintenance-admission.ts +++ b/src/selfhost/maintenance-admission.ts @@ -58,6 +58,7 @@ export const MAINTENANCE_JOB_TYPES: ReadonlySet = new Set([ "notify-deliver", "ops-alerts", "sweep-liveness-watchdog", + "reconcile-open-prs", "selftune", "rag-index-repo", "backlog-convergence-sweep", diff --git a/src/types.ts b/src/types.ts index 9bb1133d24..21bb989871 100644 --- a/src/types.ts +++ b/src/types.ts @@ -196,6 +196,15 @@ export type JobMessage = type: "sweep-liveness-watchdog"; requestedBy: "schedule" | "api" | "test"; } + | { + // Self-heal (flag-gated by GITTENSORY_PR_RECONCILIATION). List-diff GitHub's open PR numbers against the + // local table for every acting-autonomy repo — a much tighter cadence than backfillRegisteredRepositories's + // 6-hour freshness window — and catch up (fetch + upsert + regate) any PR number GitHub has that the local + // table doesn't (a silently-lost "opened" webhook). Enqueued on a short interval ONLY when the flag is ON + // (index.ts), so flag-OFF this job never exists. + type: "reconcile-open-prs"; + requestedBy: "schedule" | "api" | "test"; + } | { // Convergence (self-improve / auto-tune, flag-gated by GITTENSORY_REVIEW_SELFTUNE). Run the ported // self-improvement loop over gittensory's review-outcome data — compute tuning recommendations, diff --git a/test/unit/backfill.test.ts b/test/unit/backfill.test.ts index 1ff34f4e1f..f2807d0727 100644 --- a/test/unit/backfill.test.ts +++ b/test/unit/backfill.test.ts @@ -45,6 +45,7 @@ import { isOwnReviewThreadAuthor, isRateLimitedGitHubFailure, mergeRequiredCiContexts, + reconcileOpenPullRequests, refreshContributorActivity, refreshInstallationHealth, refreshPullRequestDetails, @@ -5823,6 +5824,99 @@ describe("GitHub backfill", () => { }); }); + describe("reconcileOpenPullRequests (#audit-open-pr-reconciliation)", () => { + it("returns an all-zero result for a repo that does not exist", async () => { + const env = createTestEnv(); + expect(await reconcileOpenPullRequests(env, "owner/missing")).toEqual({ repoFullName: "owner/missing", remoteOpenCount: 0, localOpenCount: 0, missingNumbers: [] }); + }); + + it("reports no missing numbers when the local table already has every remote-open PR", async () => { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); + await seedRegisteredRepo(env); + await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url.includes("/pulls?")) return Response.json([{ number: 1 }]); + return Response.json([]); + }); + + const result = await reconcileOpenPullRequests(env, "JSONbored/gittensory"); + + expect(result).toEqual({ repoFullName: "JSONbored/gittensory", remoteOpenCount: 1, localOpenCount: 1, missingNumbers: [] }); + }); + + it("REGRESSION (#3782/#3793): reports a PR number GitHub has open that has no local row at all", async () => { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); + await seedRegisteredRepo(env); + await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url.includes("/pulls?")) return Response.json([{ number: 1 }, { number: 7 }]); // #7 opened but never made it into the local table + return Response.json([]); + }); + + const result = await reconcileOpenPullRequests(env, "JSONbored/gittensory"); + + expect(result).toEqual({ repoFullName: "JSONbored/gittensory", remoteOpenCount: 2, localOpenCount: 1, missingNumbers: [7] }); + }); + + it("paginates the open-PR list past the first 100 via the Link header", async () => { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); + await seedRegisteredRepo(env); + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url.includes("/pulls?")) { + const page = Number(new URL(url).searchParams.get("page") ?? "1"); + if (page === 1) { + return Response.json( + Array.from({ length: 100 }, (_, i) => ({ number: i + 1 })), + { headers: { link: '; rel="next"' } }, + ); + } + return Response.json([{ number: 101 }]); + } + return Response.json([]); + }); + + const result = await reconcileOpenPullRequests(env, "JSONbored/gittensory"); + + expect(result.remoteOpenCount).toBe(101); + expect(result.missingNumbers).toEqual(expect.arrayContaining([1, 101])); + }); + + it("fails open (all-zero result) when the FIRST page fails, so a GitHub hiccup never falsely reports every local PR as missing", async () => { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); + await seedRegisteredRepo(env); + await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + vi.stubGlobal("fetch", async () => new Response("down", { status: 500 })); + + expect(await reconcileOpenPullRequests(env, "JSONbored/gittensory")).toEqual({ repoFullName: "JSONbored/gittensory", remoteOpenCount: 0, localOpenCount: 0, missingNumbers: [] }); + }); + + it("keeps the pages already fetched when a LATER page fails mid-crawl (a partial remote list can only under-report, never falsely flag a real local PR)", async () => { + const env = createTestEnv({ GITHUB_PUBLIC_TOKEN: "public-token" }); + await seedRegisteredRepo(env); + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url.includes("/pulls?")) { + const page = Number(new URL(url).searchParams.get("page") ?? "1"); + if (page === 1) { + return Response.json( + Array.from({ length: 100 }, (_, i) => ({ number: i + 1 })), + { headers: { link: '; rel="next"' } }, + ); + } + return new Response("down", { status: 500 }); + } + return Response.json([]); + }); + + const result = await reconcileOpenPullRequests(env, "JSONbored/gittensory"); + + expect(result.remoteOpenCount).toBe(100); // page 1's 100 items are kept despite page 2 failing + }); + }); + }); async function seedRegisteredRepo(env: Env) { diff --git a/test/unit/index.test.ts b/test/unit/index.test.ts index baacaf7f56..3f7b65b72f 100644 --- a/test/unit/index.test.ts +++ b/test/unit/index.test.ts @@ -645,6 +645,47 @@ describe("worker entrypoint", () => { expect(sent.some((m) => m.type === "sweep-liveness-watchdog")).toBe(false); }); + it("enqueues the reconcile-open-prs job every 10 minutes ONLY when GITTENSORY_PR_RECONCILIATION is ON (flag-OFF is byte-identical)", async () => { + const sentFor = async (flag?: string, isoTime = "2026-05-25T05:10:00.000Z"): Promise> => { + const sent: Array = []; + const env = createTestEnv({ + ...(flag === undefined ? {} : { GITTENSORY_PR_RECONCILIATION: flag }), + JOBS: { + async send(message: import("../../src/types").JobMessage) { + sent.push(message); + }, + } as unknown as Queue, + }); + const waitUntil: Promise[] = []; + await worker.scheduled(controllerFor(isoTime), env, executionContext(waitUntil)); + await Promise.all(waitUntil); + return sent; + }; + + // Flag OFF (default) → no reconcile-open-prs job; the enqueued set is unchanged from today. + expect((await sentFor()).some((m) => m.type === "reconcile-open-prs")).toBe(false); + expect((await sentFor("false")).some((m) => m.type === "reconcile-open-prs")).toBe(false); + // Flag ON, on a 10-minute boundary → exactly one reconcile-open-prs job. + const on = await sentFor("true"); + expect(on.filter((m) => m.type === "reconcile-open-prs")).toEqual([{ type: "reconcile-open-prs", requestedBy: "schedule" }]); + }); + + it("does NOT enqueue reconcile-open-prs outside the 10-minute window even when GITTENSORY_PR_RECONCILIATION is ON", async () => { + const sent: Array = []; + const env = createTestEnv({ + GITTENSORY_PR_RECONCILIATION: "true", + JOBS: { + async send(message: import("../../src/types").JobMessage) { + sent.push(message); + }, + } as unknown as Queue, + }); + const waitUntil: Promise[] = []; + await worker.scheduled(controllerFor("2026-05-25T05:14:00.000Z"), env, executionContext(waitUntil)); // not a 10-minute boundary + await Promise.all(waitUntil); + expect(sent.some((m) => m.type === "reconcile-open-prs")).toBe(false); + }); + it("enqueues selftune hourly only when GITTENSORY_REVIEW_SELFTUNE is ON", async () => { const sentFor = async ( selfTuneFlag?: string, diff --git a/test/unit/pr-reconciliation.test.ts b/test/unit/pr-reconciliation.test.ts new file mode 100644 index 0000000000..ce19bb1637 --- /dev/null +++ b/test/unit/pr-reconciliation.test.ts @@ -0,0 +1,157 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { isPrReconciliationEnabled, runOpenPrReconciliation } from "../../src/review/pr-reconciliation"; +import { getPullRequest, upsertRepositoryFromGitHub, upsertRepositorySettings } from "../../src/db/repositories"; +import * as backfillModule from "../../src/github/backfill"; +import * as repositoriesModule from "../../src/db/repositories"; +import { counterValue, resetMetrics } from "../../src/selfhost/metrics"; +import { createTestEnv } from "../helpers/d1"; + +describe("isPrReconciliationEnabled — default OFF, truthy convention", () => { + it("matches the codebase's shared truthy-string convention", () => { + for (const off of [undefined, "", "false", "no", "0", "off"]) expect(isPrReconciliationEnabled({ GITTENSORY_PR_RECONCILIATION: off })).toBe(false); + for (const on of ["1", "true", "yes", "on", "TRUE", "On"]) expect(isPrReconciliationEnabled({ GITTENSORY_PR_RECONCILIATION: on })).toBe(true); + }); +}); + +describe("runOpenPrReconciliation (#audit-open-pr-reconciliation)", () => { + afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllGlobals(); + }); + + it("REGRESSION (#3782/#3793): catches up a missing PR — fetches it, upserts it, and enqueues a regate", async () => { + resetMetrics(); + 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 upsertRepositoryFromGitHub(env, { name: "lost-repo", full_name: "owner/lost-repo", private: false, owner: { login: "owner" } }, 9400); + await upsertRepositorySettings(env, { repoFullName: "owner/lost-repo", autonomy: { merge: "auto" } }); + vi.spyOn(backfillModule, "reconcileOpenPullRequests").mockResolvedValueOnce({ repoFullName: "owner/lost-repo", remoteOpenCount: 1, localOpenCount: 0, missingNumbers: [7] }); + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url.includes("/pulls/7")) return Response.json({ number: 7, title: "Lost PR", state: "open", user: { login: "c" }, head: { sha: "a7" }, labels: [], body: "" }); + return Response.json({}); + }); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + const found = await runOpenPrReconciliation(env); + + expect(found).toEqual([{ repoFullName: "owner/lost-repo", remoteOpenCount: 1, localOpenCount: 0, missingNumbers: [7] }]); + expect(counterValue("gittensory_open_pr_reconciliation_missing_total", { repo: "owner/lost-repo" })).toBe(1); + const logged = errors.mock.calls.map((c) => String(c[0])).find((line) => line.includes("open_pr_reconciliation_divergence")); + expect(logged).toBeDefined(); + expect(JSON.parse(logged!)).toMatchObject({ level: "error", event: "open_pr_reconciliation_divergence", repository: "owner/lost-repo", missingNumbers: [7] }); + expect(sent).toEqual([expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/lost-repo", prNumber: 7, installationId: 9400 })]); + const stored = await getPullRequest(env, "owner/lost-repo", 7); + expect(stored).toMatchObject({ number: 7, title: "Lost PR" }); + }); + + it("takes no action when the list-diff finds no divergence", 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 upsertRepositoryFromGitHub(env, { name: "clean-repo", full_name: "owner/clean-repo", private: false, owner: { login: "owner" } }, 9401); + await upsertRepositorySettings(env, { repoFullName: "owner/clean-repo", autonomy: { merge: "auto" } }); + vi.spyOn(backfillModule, "reconcileOpenPullRequests").mockResolvedValueOnce({ repoFullName: "owner/clean-repo", remoteOpenCount: 1, localOpenCount: 1, missingNumbers: [] }); + + const found = await runOpenPrReconciliation(env); + + expect(found).toEqual([]); + expect(sent).toEqual([]); + }); + + it("never reconciles a registered-but-uninstalled repo (#sweep-uninstalled-budget-waste)", async () => { + const env = createTestEnv({ GITTENSORY_REVIEW_REPOS: "owner/no-install" }); + await upsertRepositoryFromGitHub(env, { name: "no-install", full_name: "owner/no-install", private: false, owner: { login: "owner" } }); // no installation id + await upsertRepositorySettings(env, { repoFullName: "owner/no-install", autonomy: { merge: "auto" } }); + const reconcileSpy = vi.spyOn(backfillModule, "reconcileOpenPullRequests"); + + const found = await runOpenPrReconciliation(env); + + expect(found).toEqual([]); + expect(reconcileSpy).not.toHaveBeenCalled(); + }); + + it("watches an ALLOWLISTED (GITTENSORY_REVIEW_REPOS) installed repo even with no autonomy configured", async () => { + const env = createTestEnv({ GITTENSORY_REVIEW_REPOS: "owner/allowlisted-repo" }); + await upsertRepositoryFromGitHub(env, { name: "allowlisted-repo", full_name: "owner/allowlisted-repo", private: false, owner: { login: "owner" } }, 9407); + const reconcileSpy = vi.spyOn(backfillModule, "reconcileOpenPullRequests").mockResolvedValueOnce({ repoFullName: "owner/allowlisted-repo", remoteOpenCount: 0, localOpenCount: 0, missingNumbers: [] }); + + await runOpenPrReconciliation(env); + + expect(reconcileSpy).toHaveBeenCalledWith(env, "owner/allowlisted-repo"); + }); + + it("skips a repo that is neither allowlisted nor agent-configured", async () => { + const env = createTestEnv(); + await upsertRepositoryFromGitHub(env, { name: "plain-repo", full_name: "owner/plain-repo", private: false, owner: { login: "owner" } }, 9402); + const reconcileSpy = vi.spyOn(backfillModule, "reconcileOpenPullRequests"); + + const found = await runOpenPrReconciliation(env); + + expect(found).toEqual([]); + expect(reconcileSpy).not.toHaveBeenCalled(); + }); + + it("fails safe per-repo: a load error on one repo is logged and the scan continues to the next repo", 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 upsertRepositoryFromGitHub(env, { name: "erroring-repo", full_name: "owner/erroring-repo", private: false, owner: { login: "owner" } }, 9403); + await upsertRepositorySettings(env, { repoFullName: "owner/erroring-repo", autonomy: { merge: "auto" } }); + await upsertRepositoryFromGitHub(env, { name: "ok-repo", full_name: "owner/ok-repo", private: false, owner: { login: "owner" } }, 9404); + await upsertRepositorySettings(env, { repoFullName: "owner/ok-repo", autonomy: { merge: "auto" } }); + vi.spyOn(backfillModule, "reconcileOpenPullRequests").mockImplementation(async (_env, repoFullName) => { + if (repoFullName === "owner/erroring-repo") throw new Error("GitHub read error"); + return { repoFullName, remoteOpenCount: 0, localOpenCount: 0, missingNumbers: [] }; + }); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + const found = await runOpenPrReconciliation(env); + + expect(found).toEqual([]); // ok-repo had no divergence, but the scan reached it despite erroring-repo's failure + expect(errors.mock.calls.some((call) => String(call[0]).includes("open_pr_reconciliation_repo_error") && String(call[0]).includes("owner/erroring-repo"))).toBe(true); + expect(sent).toEqual([]); + }); + + it("fails safe at the top level: a total scan failure is logged and returns an empty result instead of throwing", async () => { + const env = createTestEnv(); + const listSpy = vi.spyOn(repositoriesModule, "listRepositories").mockRejectedValueOnce(new Error("D1 unavailable")); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await expect(runOpenPrReconciliation(env)).resolves.toEqual([]); + + expect(errors.mock.calls.some((call) => String(call[0]).includes("open_pr_reconciliation_error"))).toBe(true); + listSpy.mockRestore(); + }); + + it("logs open_pr_reconciliation_catch_up_fetch_failed and does not throw when the missing PR's live fetch fails", async () => { + const env = createTestEnv(); + await upsertRepositoryFromGitHub(env, { name: "fetch-fails", full_name: "owner/fetch-fails", private: false, owner: { login: "owner" } }, 9405); + await upsertRepositorySettings(env, { repoFullName: "owner/fetch-fails", autonomy: { merge: "auto" } }); + vi.spyOn(backfillModule, "reconcileOpenPullRequests").mockResolvedValueOnce({ repoFullName: "owner/fetch-fails", remoteOpenCount: 1, localOpenCount: 0, missingNumbers: [9] }); + vi.stubGlobal("fetch", async () => new Response("down", { status: 500 })); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await expect(runOpenPrReconciliation(env)).resolves.toEqual([{ repoFullName: "owner/fetch-fails", remoteOpenCount: 1, localOpenCount: 0, missingNumbers: [9] }]); + + expect(errors.mock.calls.some((call) => String(call[0]).includes("open_pr_reconciliation_catch_up_fetch_failed") && String(call[0]).includes("owner/fetch-fails"))).toBe(true); + expect(await getPullRequest(env, "owner/fetch-fails", 9)).toBeNull(); + }); + + it("logs open_pr_reconciliation_catch_up_failed and does not throw when the enqueue itself fails", async () => { + const env = createTestEnv({ + JOBS: { + async send() { + throw new Error("queue send error"); + }, + } as unknown as Queue, + }); + await upsertRepositoryFromGitHub(env, { name: "send-fails", full_name: "owner/send-fails", private: false, owner: { login: "owner" } }, 9406); + await upsertRepositorySettings(env, { repoFullName: "owner/send-fails", autonomy: { merge: "auto" } }); + vi.spyOn(backfillModule, "reconcileOpenPullRequests").mockResolvedValueOnce({ repoFullName: "owner/send-fails", remoteOpenCount: 1, localOpenCount: 0, missingNumbers: [3] }); + vi.stubGlobal("fetch", async () => Response.json({ number: 3, title: "PR3", state: "open", user: { login: "c" }, head: { sha: "a3" }, labels: [], body: "" })); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await expect(runOpenPrReconciliation(env)).resolves.toEqual([{ repoFullName: "owner/send-fails", remoteOpenCount: 1, localOpenCount: 0, missingNumbers: [3] }]); + + expect(errors.mock.calls.some((call) => String(call[0]).includes("open_pr_reconciliation_catch_up_failed") && String(call[0]).includes("owner/send-fails"))).toBe(true); + }); +}); diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index b310f57e5e..f61042d03d 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -22181,6 +22181,29 @@ describe("queue processors", () => { expect(sent).toEqual([expect.objectContaining({ type: "agent-regate-sweep", repoFullName: "owner/stale-repo", installationId: 9311 })]); }); + it("reconcile-open-prs job no-ops when GITTENSORY_PR_RECONCILIATION is OFF (does no scan)", async () => { + const env = createTestEnv(); // flag unset → OFF + await upsertRepositoryFromGitHub(env, { name: "stale-repo", full_name: "owner/stale-repo", private: false, owner: { login: "owner" } }, 9410); + await upsertRepositorySettings(env, { repoFullName: "owner/stale-repo", autonomy: { merge: "auto" } }); + const reconcileSpy = vi.spyOn(backfillModule, "reconcileOpenPullRequests"); + + await processJob(env, { type: "reconcile-open-prs", requestedBy: "test" }); + + expect(reconcileSpy).not.toHaveBeenCalled(); + }); + + it("reconcile-open-prs job runs the reconciliation scan when GITTENSORY_PR_RECONCILIATION is ON", async () => { + const env = createTestEnv({ GITTENSORY_PR_RECONCILIATION: "true" }); + await upsertRepositoryFromGitHub(env, { name: "stale-repo", full_name: "owner/stale-repo", private: false, owner: { login: "owner" } }, 9411); + await upsertRepositorySettings(env, { repoFullName: "owner/stale-repo", autonomy: { merge: "auto" } }); + const reconcileSpy = vi.spyOn(backfillModule, "reconcileOpenPullRequests").mockResolvedValue({ repoFullName: "owner/stale-repo", remoteOpenCount: 0, localOpenCount: 0, missingNumbers: [] }); + + await processJob(env, { type: "reconcile-open-prs", requestedBy: "test" }); + + expect(reconcileSpy).toHaveBeenCalledWith(env, "owner/stale-repo"); + reconcileSpy.mockRestore(); + }); + describe("type label decoupling (#label-decoupling)", () => { function stubTypeLabelFetch(prNumber: number, seen: { posted: string[]; removed: string[]; checkRunCreated: boolean }) { vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { diff --git a/worker-configuration.d.ts b/worker-configuration.d.ts index 7a6d5207f3..163cc474d3 100644 --- a/worker-configuration.d.ts +++ b/worker-configuration.d.ts @@ -1,5 +1,5 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: 9e992b669fb5fd96f79df4b2f0f44770) +// Generated by Wrangler by running `wrangler types` (hash: 273aa035a633a16371217457cfc17ff6) // Runtime types generated with workerd@1.20260701.1 2026-05-28 nodejs_compat interface __BaseEnv_Env { DB: D1Database; @@ -26,6 +26,7 @@ interface __BaseEnv_Env { GITTENSORY_REVIEW_REPUTATION: "false"; GITTENSORY_REVIEW_OPS: "false"; GITTENSORY_SWEEP_WATCHDOG: "false"; + GITTENSORY_PR_RECONCILIATION: "false"; GITTENSORY_REVIEW_RAG: "false"; GITTENSORY_REVIEW_IMPACT_MAP: "false"; GITTENSORY_REVIEW_CULTURE_PROFILE: "false"; @@ -55,7 +56,7 @@ type StringifyValues> = { [Binding in keyof EnvType]: EnvType[Binding] extends string ? EnvType[Binding] : string; }; declare namespace NodeJS { - interface ProcessEnv extends StringifyValues> {} + interface ProcessEnv extends StringifyValues> {} } // Begin runtime types diff --git a/wrangler.jsonc b/wrangler.jsonc index d546fecc3f..9f40294548 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -83,6 +83,12 @@ // of requiring a human to notice and manually trigger it. Default OFF — flag-OFF the cron enqueues no // watchdog job, byte-identical to today. "GITTENSORY_SWEEP_WATCHDOG": "false", + // Self-heal: a short-interval cron list-diffs GitHub's open PR numbers against the local table for every + // acting-autonomy repo, and catches up (fetch + upsert + regate) any PR number GitHub has that the local + // table doesn't — a silently-lost "opened" webhook, caught within minutes instead of the 6-hour + // backfillRegisteredRepositories freshness window. Default OFF — flag-OFF the cron enqueues no + // reconciliation job, byte-identical to today. + "GITTENSORY_PR_RECONCILIATION": "false", // Convergence (RAG retrieval): at review time, query the codebase vector index for code/docs semantically // related to the PR's changed files and append a RELEVANT EXISTING CODE / DOCS section to the reviewer // prompt — additive reference context (callers, related modules, conventions), exactly like grounding.