From 3094d0d534db32b7cce94deec5bcc82a1a8f9b17 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 6 Jul 2026 06:48:20 -0700 Subject: [PATCH] feat(review): add hourly sweep-liveness watchdog with self-heal re-enqueue (#3808) Nothing previously noticed when the scheduled regate sweep stopped advancing a repo's last-regated marker -- the 2026-07-06 incident stayed silent for hours until a human queried the database directly. Add an hourly watchdog (flag-gated by GITTENSORY_SWEEP_WATCHDOG, default OFF) that scans the same acting-autonomy repo set the sweep itself covers. A repo with open PRs whose sweep marker hasn't advanced within a 45-minute window gets a structured sweep_liveness_stale log (Sentry- visible) and a single targeted agent-regate-sweep re-enqueue -- the same message shape the normal fan-out sends, so this only nudges the existing sweep rather than bypassing its own gating/dedup logic. A registered-but- uninstalled repo is never watched, since it can never get a per-PR fan-out regardless of nudging. Fails safe per-repo and at the top level. --- src/env.d.ts | 6 + src/index.ts | 6 + src/queue/processors.ts | 7 + src/review/sweep-watchdog.ts | 122 +++++++++++++++++ src/selfhost/maintenance-admission.ts | 1 + src/types.ts | 9 ++ test/unit/index.test.ts | 41 ++++++ test/unit/queue.test.ts | 24 ++++ test/unit/sweep-watchdog.test.ts | 190 ++++++++++++++++++++++++++ worker-configuration.d.ts | 5 +- wrangler.jsonc | 7 + 11 files changed, 416 insertions(+), 2 deletions(-) create mode 100644 src/review/sweep-watchdog.ts create mode 100644 test/unit/sweep-watchdog.test.ts diff --git a/src/env.d.ts b/src/env.d.ts index 1145fd6a1e..794f644ca1 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -241,6 +241,12 @@ declare global { * byte-identical to today. NOTE: this is read-only OBSERVABILITY only; the auto-tune / config-mutation * self-improve loop (src/review/auto-apply.ts) is deliberately NOT wired here — see ops-wire.ts. */ GITTENSORY_REVIEW_OPS?: string; + /** Self-heal: when truthy, an hourly watchdog scans the SAME acting-autonomy repo set the scheduled regate + * sweep covers for a repo whose sweep marker hasn't advanced despite having open PRs to regate, emits a + * structured `sweep_liveness_stale` log (Sentry-visible), and re-enqueues a targeted `agent-regate-sweep` + * 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; /** 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/index.ts b/src/index.ts index cd9be9e4a8..f264b4ebdf 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,6 +5,7 @@ import { processDlqBatch } from "./queue/dlq"; import { processJob } from "./queue/processors"; import { isOrbBrokerEnabled } from "./orb/broker"; import { isOpsEnabled } from "./review/ops-wire"; +import { isSweepWatchdogEnabled } from "./review/sweep-watchdog"; import { isRagEnabled } from "./review/rag-wire"; import { isSelfTuneEnabled } from "./review/selftune-wire"; import { @@ -179,6 +180,11 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController): // review-outcome data. 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 && isOpsEnabled(env)) jobs.push({ type: "ops-alerts", requestedBy: "schedule" }); + // Self-heal (flag GITTENSORY_SWEEP_WATCHDOG). Hourly liveness check over the same repo set the scheduled + // regate sweep covers — re-enqueues a targeted sweep for any repo whose sweep marker has gone stale despite + // having open PRs to regate. 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 && isSweepWatchdogEnabled(env)) jobs.push({ type: "sweep-liveness-watchdog", requestedBy: "schedule" }); // Convergence (self-improve / auto-tune, flag GITTENSORY_REVIEW_SELFTUNE). Hourly self-improvement tick over // gittensory's own review-outcome data: compute tuning recommendations, shadow-soak any strictly-tightening // one, and auto-promote it to live only after the soak window passes the gate (TIGHTENING-ONLY, audited). diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 12881b26c3..29d03245cb 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -491,6 +491,7 @@ import { import { DEFAULT_UNLINKED_ISSUE_GUARDRAIL } from "../review/unlinked-issue-guardrail-config"; import { resolveUnlinkedIssueMatchDisposition } from "../review/unlinked-issue-guardrail"; import { isOpsEnabled, runOpsAlerts } from "../review/ops-wire"; +import { isSweepWatchdogEnabled, runSweepLivenessWatchdog } from "../review/sweep-watchdog"; import { isSelfTuneEnabled, runSelfTune } from "../review/selftune-wire"; import { isCloseHoldOnly, @@ -1111,6 +1112,12 @@ export async function processJob(env: Env, message: JobMessage): Promise { // flag-OFF does zero work here too. Read-only telemetry — never throws into the queue. if (isOpsEnabled(env)) await runOpsAlerts(env); return; + case "sweep-liveness-watchdog": + // Self-heal (flag GITTENSORY_SWEEP_WATCHDOG). 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 (isSweepWatchdogEnabled(env)) await runSweepLivenessWatchdog(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/sweep-watchdog.ts b/src/review/sweep-watchdog.ts new file mode 100644 index 0000000000..46ba651ebd --- /dev/null +++ b/src/review/sweep-watchdog.ts @@ -0,0 +1,122 @@ +// Self-heal (flag-gated by GITTENSORY_SWEEP_WATCHDOG). The scheduled regate sweep (fanOutAgentRegateSweepJobs / +// sweepRepoRegate, src/queue/processors.ts) advances every acting-autonomy repo's `last_regated_at` marker on +// every successful sweep tick. When the sweep stops advancing that marker for a repo — a stalled cron, a wedged +// per-repo failure that keeps recurring, a rate-limit floor that never clears — nothing in the system previously +// noticed on its own; the 2026-07-06 incident stayed silent for hours until a human queried the database. This +// watchdog closes that gap: it runs the SAME repo-selection the sweep itself uses, and for any repo with open +// PRs whose sweep marker hasn't advanced within the staleness window, it (a) emits a structured Sentry-visible +// log, and (b) re-enqueues a single targeted `agent-regate-sweep` for just that repo — the same message shape +// the normal fan-out sends, so this is a pure "nudge," never a bypass of the sweep's own gating/dedup logic. +// +// 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 { isAgentConfigured } from "../settings/autonomy"; +import { resolveRepositorySettings } from "../settings/repository-settings"; +import type { JobMessage } from "../types"; +import { errorMessage, nowIso } from "../utils/json"; +import { isConvergenceRepoAllowed, listConvergenceRepos } from "./cutover-gate"; + +/** True when the sweep-liveness watchdog is enabled. Flag-OFF (default) → the caller never invokes it, so the + * cron enqueues no watchdog job and the queue processor no-ops on a stale in-flight one (defense-in-depth, + * mirrors isOpsEnabled). */ +export function isSweepWatchdogEnabled(env: { GITTENSORY_SWEEP_WATCHDOG?: string | undefined }): boolean { + return /^(1|true|yes|on)$/i.test(env.GITTENSORY_SWEEP_WATCHDOG ?? ""); +} + +/** A repo's sweep is stale when it has open PRs to regate but its last-regated marker either never advanced or + * hasn't advanced within the staleness window. A repo with NO open PRs is never stale — there is nothing for + * the sweep to do, so a `null` marker there means "nothing to regate," not "the sweep stopped working." */ +export const SWEEP_STALENESS_THRESHOLD_MS = 45 * 60 * 1000; + +export function isSweepStale(input: { openPullRequestCount: number; lastRegatedAt: string | null; nowMs: number }): boolean { + if (input.openPullRequestCount === 0) return false; + const lastMs = input.lastRegatedAt ? Date.parse(input.lastRegatedAt) : NaN; + if (!Number.isFinite(lastMs)) return true; + return input.nowMs - lastMs > SWEEP_STALENESS_THRESHOLD_MS; +} + +/** The same acting-autonomy repo set fanOutAgentRegateSweepJobs sweeps: the convergence allowlist + * (GITTENSORY_REVIEW_REPOS) union the webhook-registered repos with acting autonomy, deduped case-insensitively. + * Deliberately mirrors that function's own selection so the watchdog can never watch a DIFFERENT set of repos + * than the sweep actually covers. */ +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 watchdog scan */ + } + } + return configured; +} + +export interface StaleSweepRepo { + repoFullName: string; + installationId?: number | undefined; + openPullRequestCount: number; + lastRegatedAt: string | null; + ageMs: number; +} + +/** + * The watchdog 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) gets a self-heal re-enqueue — a registered-but-uninstalled repo never + * gets a per-PR fan-out regardless (#sweep-uninstalled-budget-waste), so re-enqueuing its sweep would just spend + * the shared GITHUB_PUBLIC_TOKEN budget on a sweep that can never act. Returns the stale repos found (for tests / + * a caller that wants to act further). + * + * Caller MUST gate this on {@link isSweepWatchdogEnabled} — 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 runSweepLivenessWatchdog(env: Env): Promise { + const found: StaleSweepRepo[] = []; + const nowMs = Date.parse(nowIso()); + try { + const repos = await watchedRepos(env); + for (const repo of repos) { + try { + if (typeof repo.installationId !== "number") continue; + const [openPullRequestCount, lastRegatedAt] = await Promise.all([countOpenPullRequests(env, repo.fullName), getLatestRegatedAt(env, repo.fullName)]); + if (!isSweepStale({ openPullRequestCount, lastRegatedAt, nowMs })) continue; + const lastMs = lastRegatedAt ? Date.parse(lastRegatedAt) : NaN; + const ageMs = Number.isFinite(lastMs) ? nowMs - lastMs : Number.POSITIVE_INFINITY; + found.push({ repoFullName: repo.fullName, installationId: repo.installationId, openPullRequestCount, lastRegatedAt, ageMs }); + console.error( + JSON.stringify({ + level: "error", + event: "sweep_liveness_stale", + repository: repo.fullName, + openPullRequestCount, + lastRegatedAt, + ageMs: Number.isFinite(ageMs) ? ageMs : null, + }), + ); + const message: JobMessage = { type: "agent-regate-sweep", requestedBy: "schedule", repoFullName: repo.fullName, installationId: repo.installationId }; + await env.JOBS.send(message).catch((error) => { + console.error(JSON.stringify({ level: "error", event: "sweep_liveness_reenqueue_failed", repository: repo.fullName, error: errorMessage(error) })); + }); + } catch (error) { + console.error(JSON.stringify({ level: "error", event: "sweep_liveness_repo_error", repository: repo.fullName, message: errorMessage(error).slice(0, 200) })); + } + } + } catch (error) { + console.error(JSON.stringify({ level: "error", event: "sweep_liveness_error", message: errorMessage(error).slice(0, 200) })); + } + return found; +} diff --git a/src/selfhost/maintenance-admission.ts b/src/selfhost/maintenance-admission.ts index 5f87fa7e1a..c4b20b2a08 100644 --- a/src/selfhost/maintenance-admission.ts +++ b/src/selfhost/maintenance-admission.ts @@ -57,6 +57,7 @@ export const MAINTENANCE_JOB_TYPES: ReadonlySet = new Set([ "notify-evaluate", "notify-deliver", "ops-alerts", + "sweep-liveness-watchdog", "selftune", "rag-index-repo", "backlog-convergence-sweep", diff --git a/src/types.ts b/src/types.ts index 81ce0c7411..053c11f7fe 100644 --- a/src/types.ts +++ b/src/types.ts @@ -187,6 +187,15 @@ export type JobMessage = type: "ops-alerts"; requestedBy: "schedule" | "api" | "test"; } + | { + // Self-heal (flag-gated by GITTENSORY_SWEEP_WATCHDOG). Scan the SAME acting-autonomy repo set the + // scheduled regate sweep covers for a stalled per-repo sweep (open PRs present, but none regated within + // the staleness window) — emit a structured `sweep_liveness_stale` log AND re-enqueue a targeted + // `agent-regate-sweep` for just that repo. Enqueued hourly by the cron ONLY when the flag is ON + // (index.ts), so flag-OFF this job never exists. + type: "sweep-liveness-watchdog"; + 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/index.test.ts b/test/unit/index.test.ts index 4e2cf3bf23..baacaf7f56 100644 --- a/test/unit/index.test.ts +++ b/test/unit/index.test.ts @@ -604,6 +604,47 @@ describe("worker entrypoint", () => { expect(sent.some((m) => m.type === "ops-alerts")).toBe(false); }); + it("enqueues the sweep-liveness-watchdog job hourly ONLY when GITTENSORY_SWEEP_WATCHDOG is ON (flag-OFF is byte-identical)", async () => { + const sentFor = async (watchdogFlag?: string): Promise> => { + const sent: Array = []; + const env = createTestEnv({ + ...(watchdogFlag === undefined ? {} : { GITTENSORY_SWEEP_WATCHDOG: watchdogFlag }), + 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:00:00.000Z"), env, executionContext(waitUntil)); + await Promise.all(waitUntil); + return sent; + }; + + // Flag OFF (default) → no sweep-liveness-watchdog job; the enqueued set is unchanged from today. + expect((await sentFor()).some((m) => m.type === "sweep-liveness-watchdog")).toBe(false); + expect((await sentFor("false")).some((m) => m.type === "sweep-liveness-watchdog")).toBe(false); + // Flag ON → exactly one sweep-liveness-watchdog job, enqueued in the hourly window. + const on = await sentFor("true"); + expect(on.filter((m) => m.type === "sweep-liveness-watchdog")).toEqual([{ type: "sweep-liveness-watchdog", requestedBy: "schedule" }]); + }); + + it("does NOT enqueue sweep-liveness-watchdog outside the hourly window even when GITTENSORY_SWEEP_WATCHDOG is ON", async () => { + const sent: Array = []; + const env = createTestEnv({ + GITTENSORY_SWEEP_WATCHDOG: "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:15:00.000Z"), env, executionContext(waitUntil)); // non-hourly + await Promise.all(waitUntil); + expect(sent.some((m) => m.type === "sweep-liveness-watchdog")).toBe(false); + }); + it("enqueues selftune hourly only when GITTENSORY_REVIEW_SELFTUNE is ON", async () => { const sentFor = async ( selfTuneFlag?: string, diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 544f844d11..3704ab9759 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -21946,6 +21946,30 @@ describe("queue processors", () => { errors.mockRestore(); }); + it("sweep-liveness-watchdog job no-ops when GITTENSORY_SWEEP_WATCHDOG is OFF (does no scan, no re-enqueue)", 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 }); // flag unset → OFF + await upsertRepositoryFromGitHub(env, { name: "stale-repo", full_name: "owner/stale-repo", private: false, owner: { login: "owner" } }, 9310); + await upsertRepositorySettings(env, { repoFullName: "owner/stale-repo", autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, "owner/stale-repo", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + + await processJob(env, { type: "sweep-liveness-watchdog", requestedBy: "test" }); + + expect(sent).toEqual([]); + }); + + it("sweep-liveness-watchdog job runs the liveness scan and re-enqueues a stale repo when GITTENSORY_SWEEP_WATCHDOG is ON", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ GITTENSORY_SWEEP_WATCHDOG: "true", JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); + await upsertRepositoryFromGitHub(env, { name: "stale-repo", full_name: "owner/stale-repo", private: false, owner: { login: "owner" } }, 9311); + await upsertRepositorySettings(env, { repoFullName: "owner/stale-repo", autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, "owner/stale-repo", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + + await processJob(env, { type: "sweep-liveness-watchdog", requestedBy: "test" }); + + expect(sent).toEqual([expect.objectContaining({ type: "agent-regate-sweep", repoFullName: "owner/stale-repo", installationId: 9311 })]); + }); + 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/test/unit/sweep-watchdog.test.ts b/test/unit/sweep-watchdog.test.ts new file mode 100644 index 0000000000..1df3827963 --- /dev/null +++ b/test/unit/sweep-watchdog.test.ts @@ -0,0 +1,190 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { isSweepStale, isSweepWatchdogEnabled, runSweepLivenessWatchdog, SWEEP_STALENESS_THRESHOLD_MS } from "../../src/review/sweep-watchdog"; +import { markPullRequestsRegated, upsertPullRequestFromGitHub, upsertRepositoryFromGitHub, upsertRepositorySettings } from "../../src/db/repositories"; +import * as repositoriesModule from "../../src/db/repositories"; +import { createTestEnv } from "../helpers/d1"; + +describe("isSweepWatchdogEnabled — default OFF, truthy convention", () => { + it("matches the codebase's shared truthy-string convention", () => { + for (const off of [undefined, "", "false", "no", "0", "off"]) expect(isSweepWatchdogEnabled({ GITTENSORY_SWEEP_WATCHDOG: off })).toBe(false); + for (const on of ["1", "true", "yes", "on", "TRUE", "On"]) expect(isSweepWatchdogEnabled({ GITTENSORY_SWEEP_WATCHDOG: on })).toBe(true); + }); +}); + +describe("isSweepStale (#audit-sweep-fanout-isolation follow-up)", () => { + const NOW = Date.parse("2026-07-06T12:00:00.000Z"); + + it("a repo with NO open PRs is never stale, regardless of the marker", () => { + expect(isSweepStale({ openPullRequestCount: 0, lastRegatedAt: null, nowMs: NOW })).toBe(false); + expect(isSweepStale({ openPullRequestCount: 0, lastRegatedAt: "2020-01-01T00:00:00.000Z", nowMs: NOW })).toBe(false); + }); + + it("a repo with open PRs and NO regate marker at all is stale (never regated)", () => { + expect(isSweepStale({ openPullRequestCount: 1, lastRegatedAt: null, nowMs: NOW })).toBe(true); + }); + + it("a repo with open PRs and an unparseable marker is stale (fails toward stale, not silently healthy)", () => { + expect(isSweepStale({ openPullRequestCount: 1, lastRegatedAt: "not-a-date", nowMs: NOW })).toBe(true); + }); + + it("a repo regated within the staleness window is NOT stale", () => { + const lastRegatedAt = new Date(NOW - (SWEEP_STALENESS_THRESHOLD_MS - 1000)).toISOString(); + expect(isSweepStale({ openPullRequestCount: 1, lastRegatedAt, nowMs: NOW })).toBe(false); + }); + + it("a repo NOT regated within the staleness window IS stale", () => { + const lastRegatedAt = new Date(NOW - (SWEEP_STALENESS_THRESHOLD_MS + 1000)).toISOString(); + expect(isSweepStale({ openPullRequestCount: 1, lastRegatedAt, nowMs: NOW })).toBe(true); + }); +}); + +describe("runSweepLivenessWatchdog (#audit-sweep-fanout-isolation follow-up)", () => { + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("re-enqueues a targeted sweep + logs sweep_liveness_stale for an installed repo with open PRs whose marker never advanced", 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: "stale-repo", full_name: "owner/stale-repo", private: false, owner: { login: "owner" } }, 9300); + await upsertRepositorySettings(env, { repoFullName: "owner/stale-repo", autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, "owner/stale-repo", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + const found = await runSweepLivenessWatchdog(env); + + expect(found).toEqual([expect.objectContaining({ repoFullName: "owner/stale-repo", installationId: 9300, openPullRequestCount: 1 })]); + expect(sent).toEqual([expect.objectContaining({ type: "agent-regate-sweep", requestedBy: "schedule", repoFullName: "owner/stale-repo", installationId: 9300 })]); + const logged = errors.mock.calls.map((c) => String(c[0])).find((line) => line.includes("sweep_liveness_stale") && line.includes("owner/stale-repo")); + expect(logged).toBeDefined(); + expect(JSON.parse(logged!)).toMatchObject({ level: "error", event: "sweep_liveness_stale", repository: "owner/stale-repo" }); + }); + + it("REGRESSION: reports a finite ageMs for a repo that WAS regated once but fell outside the staleness window (not just a never-regated null marker)", async () => { + vi.useFakeTimers(); + const start = new Date("2026-07-06T10:00:00.000Z"); + vi.setSystemTime(start); + 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: "aged-repo", full_name: "owner/aged-repo", private: false, owner: { login: "owner" } }, 9306); + await upsertRepositorySettings(env, { repoFullName: "owner/aged-repo", autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, "owner/aged-repo", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + await markPullRequestsRegated(env, "owner/aged-repo", [1]); // stamps last_regated_at = start + vi.setSystemTime(new Date(start.getTime() + SWEEP_STALENESS_THRESHOLD_MS + 60_000)); // now outside the window + + const found = await runSweepLivenessWatchdog(env); + + expect(found).toEqual([expect.objectContaining({ repoFullName: "owner/aged-repo", lastRegatedAt: start.toISOString(), ageMs: SWEEP_STALENESS_THRESHOLD_MS + 60_000 })]); + expect(Number.isFinite(found[0]?.ageMs)).toBe(true); + expect(sent).toEqual([expect.objectContaining({ type: "agent-regate-sweep", repoFullName: "owner/aged-repo" })]); + }); + + it("watches an ALLOWLISTED (GITTENSORY_REVIEW_REPOS) installed repo even with no autonomy configured, and skips a plain repo that is neither allowlisted nor agent-configured", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ + GITTENSORY_REVIEW_REPOS: "owner/allowlisted-repo", + JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue, + }); + // Allowlisted + installed, but NO autonomy config at all — isConvergenceRepoAllowed alone must still watch it. + await upsertRepositoryFromGitHub(env, { name: "allowlisted-repo", full_name: "owner/allowlisted-repo", private: false, owner: { login: "owner" } }, 9307); + await upsertPullRequestFromGitHub(env, "owner/allowlisted-repo", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + // Neither allowlisted nor agent-configured — must be excluded entirely, regardless of its own staleness. + await upsertRepositoryFromGitHub(env, { name: "plain-repo", full_name: "owner/plain-repo", private: false, owner: { login: "owner" } }, 9308); + await upsertPullRequestFromGitHub(env, "owner/plain-repo", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + + const found = await runSweepLivenessWatchdog(env); + + expect(found.map((f) => f.repoFullName)).toEqual(["owner/allowlisted-repo"]); + expect(sent).toEqual([expect.objectContaining({ repoFullName: "owner/allowlisted-repo", installationId: 9307 })]); + }); + + it("fails safe at the top level: a total scan failure (e.g. listRepositories throwing) 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(runSweepLivenessWatchdog(env)).resolves.toEqual([]); + + expect(errors.mock.calls.some((call) => String(call[0]).includes("sweep_liveness_error"))).toBe(true); + listSpy.mockRestore(); + }); + + it("does NOT re-enqueue a repo regated within the staleness window", 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: "fresh-repo", full_name: "owner/fresh-repo", private: false, owner: { login: "owner" } }, 9301); + await upsertRepositorySettings(env, { repoFullName: "owner/fresh-repo", autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, "owner/fresh-repo", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + await markPullRequestsRegated(env, "owner/fresh-repo", [1]); + + const found = await runSweepLivenessWatchdog(env); + + expect(found).toEqual([]); + expect(sent).toEqual([]); + }); + + it("does NOT flag a repo with zero open PRs, even with no regate marker at all", 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: "quiet-repo", full_name: "owner/quiet-repo", private: false, owner: { login: "owner" } }, 9302); + await upsertRepositorySettings(env, { repoFullName: "owner/quiet-repo", autonomy: { merge: "auto" } }); + + const found = await runSweepLivenessWatchdog(env); + + expect(found).toEqual([]); + expect(sent).toEqual([]); + }); + + it("never flags a registered-but-uninstalled repo (#sweep-uninstalled-budget-waste) — no per-PR fan-out could ever help it", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ GITTENSORY_REVIEW_REPOS: "owner/no-install", JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); + 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" } }); + await upsertPullRequestFromGitHub(env, "owner/no-install", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + + const found = await runSweepLivenessWatchdog(env); + + expect(found).toEqual([]); + expect(sent).toEqual([]); + }); + + 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" } }, 9303); + 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" } }, 9304); + await upsertRepositorySettings(env, { repoFullName: "owner/ok-repo", autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, "owner/erroring-repo", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + await upsertPullRequestFromGitHub(env, "owner/ok-repo", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + const countSpy = vi.spyOn(repositoriesModule, "countOpenPullRequests").mockImplementation(async (_env, fullName) => { + if (fullName === "owner/erroring-repo") throw new Error("D1 read error"); + return 1; + }); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + const found = await runSweepLivenessWatchdog(env); + + expect(found).toEqual([expect.objectContaining({ repoFullName: "owner/ok-repo" })]); // erroring-repo's failure did not block ok-repo + expect(errors.mock.calls.some((call) => String(call[0]).includes("sweep_liveness_repo_error") && String(call[0]).includes("owner/erroring-repo"))).toBe(true); + countSpy.mockRestore(); + }); + + it("logs sweep_liveness_reenqueue_failed and does not throw when the re-enqueue send 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" } }, 9305); + await upsertRepositorySettings(env, { repoFullName: "owner/send-fails", autonomy: { merge: "auto" } }); + await upsertPullRequestFromGitHub(env, "owner/send-fails", { number: 1, title: "PR1", state: "open", user: { login: "c" }, head: { sha: "a1" }, labels: [], body: "" }); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await expect(runSweepLivenessWatchdog(env)).resolves.toEqual([expect.objectContaining({ repoFullName: "owner/send-fails" })]); // still reported as found even though the re-enqueue itself failed + expect(errors.mock.calls.some((call) => String(call[0]).includes("sweep_liveness_reenqueue_failed") && String(call[0]).includes("owner/send-fails"))).toBe(true); + }); +}); diff --git a/worker-configuration.d.ts b/worker-configuration.d.ts index 4ec12f8f54..7a6d5207f3 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: 1101774c5b8456e1e46f2b918558b1aa) +// Generated by Wrangler by running `wrangler types` (hash: 9e992b669fb5fd96f79df4b2f0f44770) // Runtime types generated with workerd@1.20260701.1 2026-05-28 nodejs_compat interface __BaseEnv_Env { DB: D1Database; @@ -25,6 +25,7 @@ interface __BaseEnv_Env { GITTENSORY_REVIEW_GROUNDING: "false"; GITTENSORY_REVIEW_REPUTATION: "false"; GITTENSORY_REVIEW_OPS: "false"; + GITTENSORY_SWEEP_WATCHDOG: "false"; GITTENSORY_REVIEW_RAG: "false"; GITTENSORY_REVIEW_IMPACT_MAP: "false"; GITTENSORY_REVIEW_CULTURE_PROFILE: "false"; @@ -54,7 +55,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 556deeb50a..d546fecc3f 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -76,6 +76,13 @@ // aggregate. Read-only observability; the auto-tune/config-mutation self-improve loop is NOT wired here. // Default OFF — flag-OFF the cron enqueues no ops job and the endpoint 404s, byte-identical to today. "GITTENSORY_REVIEW_OPS": "false", + // Self-heal (#audit-sweep-fanout-isolation follow-up): an hourly watchdog over the SAME acting-autonomy repo + // set the scheduled regate sweep covers — a repo with open PRs whose last-regated marker hasn't advanced in + // over the staleness window gets a structured `sweep_liveness_stale` log (Sentry-visible) AND a single + // targeted `agent-regate-sweep` re-enqueue for just that repo, so a stalled sweep recovers on its own instead + // 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", // 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.