feat(review): add hourly sweep-liveness watchdog with self-heal re-enqueue

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

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).

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<void> {
       // 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

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<Array<{ fullName: string; installationId?: number }>> {
+  const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo]));
+  const byKey = new Map<string, { fullName: string; installationId?: number }>();
+  for (const repo of repositoriesByKey.values())
+    byKey.set(repo.fullName.toLowerCase(), { fullName: repo.fullName, ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}) });
+  for (const fullName of listConvergenceRepos(env)) {
+    const repo = repositoriesByKey.get(fullName.toLowerCase());
+    byKey.set(fullName.toLowerCase(), {
+      fullName,
+      ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}),
+    });
+  }
+  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<StaleSweepRepo[]> {
+  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;
+}

src/selfhost/maintenance-admission.ts

@@ -57,6 +57,7 @@ export const MAINTENANCE_JOB_TYPES: ReadonlySet<string> = new Set([
   "notify-evaluate",
   "notify-deliver",
   "ops-alerts",
+  "sweep-liveness-watchdog",
   "selftune",
   "rag-index-repo",
   "backlog-convergence-sweep",

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,

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<Array<import("../../src/types").JobMessage>> => {
+      const sent: Array<import("../../src/types").JobMessage> = [];
+      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<unknown>[] = [];
+      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<import("../../src/types").JobMessage> = [];
+    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<unknown>[] = [];
+    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,

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) => {