fix(observability): forward ops-alerts anomalies to Sentry, add a review-burst rule (#3766)

Two changes to the hourly ops-alerts anomaly scan:

1. Fix a Sentry-visibility gap found while investigating the CI-stuck
   repeat-review incident: every existing anomaly (gate false-positive
   spike, slop score inverting, recommendations not panning out) was
   logged via console.warn with an `ev` field. forwardStructuredLogToSentry
   only wraps console.log/console.error and keys the issue off a field
   literally named `event` -- so none of these anomalies has EVER reached
   Sentry, regardless of whether Sentry is active. They only ever reached
   Workers Logs, which nobody was watching -- part of why the CI-stuck
   bleed went unnoticed for 20+ hours. Switched to console.error + an
   `event` field, matching the convention already used by
   selfhost_ai_provider_failed / regate_repair_exhausted /
   ci_stuck_review_repeat_suppressed.

2. Add a 4th anomaly rule: "review burst" -- flags a PR that published
   more than 6 review surfaces within a 2-hour rolling window, the exact
   signature of a stuck-CI finalize loop or sweep retry storm. Backed by
   a new findHottestReviewTargetForRepo query over the existing
   audit_events ledger (github_app.pr_public_surface_published is a
   genuine INSERT-only event per publish pass, unlike ai_review_cache or
   review_audit's gate_decision rows, which are both upserted at a fixed
   head SHA and so cannot see a same-head repeat). This runs on the
   existing hourly ops-alerts cron, so a future recurrence of this class
   of bleed surfaces within an hour instead of requiring a human to
   notice and query the database directly.

Author: JSONbored

src/db/repositories.ts

@@ -1,4 +1,4 @@
-import { and, asc, desc, eq, gte, inArray, not, or, sql, type SQL } from "drizzle-orm";
+import { and, asc, desc, eq, gte, inArray, like, not, or, sql, type SQL } from "drizzle-orm";
 import { getDb } from "./client";
 import {
   activeReviewTracking,
@@ -2552,6 +2552,40 @@ export async function countRecentAuditEventsForActorAndTarget(env: Env, actor: s
   return row.count;
 }
 
+/** #orb-ci-stuck-repeat / #orb-retry-storm ops-alerts signal: the single PR within `repoFullName` that published
+ *  the most review surfaces in the last `sinceIso`-bounded window, and how many. `github_app.pr_public_surface_
+ *  published` is a genuine INSERT-only event (never upserted) recorded once per successful publish pass
+ *  (processors.ts's finishPublicSurfacePublication), so unlike ai_review_cache or review_audit's gate_decision
+ *  rows (both keyed + upserted on `(repo, pr, headSha)`, so a repeat pass at an UNCHANGED head silently
+ *  overwrites rather than accumulates), this correctly counts repeat publishes even when the head SHA never
+ *  changes -- exactly the shape of a stuck-CI or sweep retry-storm bleed. Returns null when the repo published
+ *  no surfaces in the window at all. */
+export async function findHottestReviewTargetForRepo(
+  env: Env,
+  repoFullName: string,
+  sinceIso: string,
+): Promise<{ targetKey: string; count: number } | null> {
+  const db = getDb(env.DB);
+  const [row] = await db
+    .select({ targetKey: auditEvents.targetKey, count: sql<number>`count(*)` })
+    .from(auditEvents)
+    .where(
+      and(
+        eq(auditEvents.eventType, "github_app.pr_public_surface_published"),
+        like(auditEvents.targetKey, `${repoFullName}#%`),
+        gte(auditEvents.createdAt, sinceIso),
+      ),
+    )
+    .groupBy(auditEvents.targetKey)
+    .orderBy(desc(sql`count(*)`))
+    .limit(1);
+  /* v8 ignore next -- the WHERE clause's `like(auditEvents.targetKey, ...)` can never match a NULL target_key
+   *  (SQL LIKE against NULL is NULL, never true), so a returned row always has a non-null targetKey; the
+   *  column's nullable TS type is a schema-wide default this specific query structurally rules out. */
+  if (!row || row.targetKey === null) return null;
+  return { targetKey: row.targetKey, count: row.count };
+}
+
 /** Moderation-rules engine (#selfhost-mod-engine): the actor's TOTAL violation count across every rule type in
  *  `eventTypes` and EVERY repo this install tracks (no targetKey/route scoping -- `audit_events` carries no
  *  repo/installation column at all, so this is inherently install-wide, mirroring the install-wide contributor

src/review/ops-wire.ts

@@ -14,17 +14,23 @@
 //     services/outcome-calibration.ts (buildRepoOutcomeCalibration).
 //
 // NOTIFY PATH: gittensory has NO Discord / operator webhook (notifications/service.ts is a per-recipient,
-// pull-based BADGE feed — the wrong channel for an operator anomaly). So, per the task ("Discord/webhook if
-// present, else a structured log"), an anomaly emits a structured `console.warn` log line (the house
-// `JSON.stringify({ ev: ... })` convention used across the worker) that Workers Logs/Observability surfaces.
+// pull-based BADGE feed — the wrong channel for an operator anomaly). So an anomaly emits a structured
+// `console.error` log line with an `event` field (#orb-ci-stuck-repeat: this was previously `console.warn` with
+// an `ev` field — forwardStructuredLogToSentry, src/selfhost/sentry.ts, only wraps console.log/console.error
+// -- never console.warn -- and keys the Sentry issue off a field literally named `event`, not `ev`. Under the
+// old shape, every anomaly this module ever found was invisible to Sentry regardless of whether Sentry was
+// active; it only ever reached Workers Logs, which is why a 20+-hour token-usage bleed went unnoticed until a
+// human queried the database directly. `console.error` + `event` is the same convention every other Sentry-
+// visible anomaly signal in this codebase already uses (selfhost_ai_provider_failed, regate_repair_exhausted,
+// ci_stuck_review_repeat_suppressed).
 //
 // DEFERRED (NOT implemented here): the auto-tune / auto-apply config-mutation self-improve loop. The ported
 // pure logic + D1 store already exist in src/review/auto-apply.ts, but actually CLOSING the loop (mutating a
 // live gate's tunables from the cron) is sensitive — it needs the `tunables_overrides` / `_shadow` /
 // `override_audit` D1 tables (none of which exist in gittensory's migrations yet) plus a careful soak/promote
 // design. This module is READ-ONLY observability: it reports drift; it never changes what blocks a live PR.
 
-import { listRepositories } from "../db/repositories";
+import { findHottestReviewTargetForRepo, listRepositories } from "../db/repositories";
 import { isAgentConfigured } from "../settings/autonomy";
 import { resolveRepositorySettings } from "../settings/repository-settings";
 import { loadGatePrecisionReport, type GatePrecisionReport } from "../services/gate-precision";
@@ -47,12 +53,24 @@ const GATE_FALSE_POSITIVE_THRESHOLD = 0.3;
 const RECOMMENDATION_NEGATIVE_THRESHOLD = 0.5;
 /** Don't judge the recommendation negative-rate off a trickle of resolved outcomes. */
 const MIN_RECOMMENDATION_RESOLVED = 5;
-
-/** One repo's outcome reports + the repo it covers — the input to the pure anomaly detector. */
+/** #orb-ci-stuck-repeat / #orb-retry-storm: more than this many published review surfaces for the SAME PR within
+ *  REVIEW_BURST_WINDOW_HOURS is not normal iteration (a human pushing a few follow-up commits tops out well
+ *  below this) -- it is the signature of a stuck-CI finalize loop or a sweep retry storm. Conservative on
+ *  purpose: an actively-iterated PR with several quick pushes should never trip this. */
+const REVIEW_BURST_THRESHOLD = 6;
+/** Rolling window the review-burst count is computed over. Short enough that the hourly ops-alerts cron catches
+ *  a live bleed within one or two ticks, not the 20+ hours it took a human to notice the incident this exists
+ *  to prevent from recurring. */
+const REVIEW_BURST_WINDOW_HOURS = 2;
+
+/** One repo's outcome reports + the repo it covers — the input to the pure anomaly detector. `reviewBurst` is
+ *  optional so existing snapshot-fixture tests need not be touched; absent/null means "not computed", not
+ *  "healthy" -- the caller (runOpsAlerts/computeOpsStats) always populates it today. */
 export interface RepoOutcomeSnapshot {
   repoFullName: string;
   gatePrecision: GatePrecisionReport;
   calibration: OutcomeCalibration;
+  reviewBurst?: { targetKey: string; count: number } | null | undefined;
 }
 
 /**
@@ -93,6 +111,15 @@ export function detectOutcomeAnomalies(snapshot: RepoOutcomeSnapshot): string[]
     );
   }
 
+  // REVIEW BURST (#orb-ci-stuck-repeat / #orb-retry-storm): the same PR published far more review surfaces than
+  // normal iteration ever produces within a short window -- catch a stuck-CI finalize loop or sweep retry storm
+  // within this scan's own next tick instead of requiring a human to notice hours later.
+  if (snapshot.reviewBurst && snapshot.reviewBurst.count >= REVIEW_BURST_THRESHOLD) {
+    out.push(
+      `review burst: ${snapshot.reviewBurst.targetKey} published ${snapshot.reviewBurst.count} review surfaces in the last ${REVIEW_BURST_WINDOW_HOURS}h — likely a stuck-CI finalize loop or retry storm, not normal iteration. Investigate why this PR keeps re-triggering a fresh review.`,
+    );
+  }
+
   return out;
 }
 
@@ -127,25 +154,28 @@ async function opsScanRepos(env: Env): Promise<string[]> {
  */
 export async function runOpsAlerts(env: Env): Promise<Record<string, string[]>> {
   const found: Record<string, string[]> = {};
+  const reviewBurstSinceIso = new Date(Date.now() - REVIEW_BURST_WINDOW_HOURS * 60 * 60 * 1000).toISOString();
   try {
     const repos = await opsScanRepos(env);
     for (const repoFullName of repos) {
       try {
-        const [gatePrecision, calibration] = await Promise.all([
+        const [gatePrecision, calibration, reviewBurst] = await Promise.all([
           loadGatePrecisionReport(env, repoFullName),
           buildRepoOutcomeCalibration(env, repoFullName),
+          findHottestReviewTargetForRepo(env, repoFullName, reviewBurstSinceIso),
         ]);
-        const anomalies = detectOutcomeAnomalies({ repoFullName, gatePrecision, calibration });
+        const anomalies = detectOutcomeAnomalies({ repoFullName, gatePrecision, calibration, reviewBurst });
         if (anomalies.length === 0) continue;
         found[repoFullName] = anomalies;
-        // Structured log = gittensory's notify path (no Discord/operator webhook exists). One line per repo.
-        console.warn(JSON.stringify({ ev: "ops_anomaly", repo: repoFullName, at: nowIso(), anomalies }));
+        // Structured log = gittensory's notify path (no Discord/operator webhook exists) AND the Sentry path
+        // (level:"error" + an `event` field reaches forwardStructuredLogToSentry). One line per repo.
+        console.error(JSON.stringify({ level: "error", event: "ops_anomaly", repo: repoFullName, at: nowIso(), anomalies }));
       } catch (error) {
-        console.warn(JSON.stringify({ ev: "ops_anomaly_repo_error", repo: repoFullName, message: errorMessage(error).slice(0, 200) }));
+        console.error(JSON.stringify({ level: "error", event: "ops_anomaly_repo_error", repo: repoFullName, message: errorMessage(error).slice(0, 200) }));
       }
     }
   } catch (error) {
-    console.warn(JSON.stringify({ ev: "ops_anomaly_error", message: errorMessage(error).slice(0, 200) }));
+    console.error(JSON.stringify({ level: "error", event: "ops_anomaly_error", message: errorMessage(error).slice(0, 200) }));
   }
   return found;
 }
@@ -177,11 +207,13 @@ export interface OpsStatsPayload {
 export async function computeOpsStats(env: Env): Promise<OpsStatsPayload> {
   const repos = await opsScanRepos(env);
   const rows: OpsStatsRepoRow[] = [];
+  const reviewBurstSinceIso = new Date(Date.now() - REVIEW_BURST_WINDOW_HOURS * 60 * 60 * 1000).toISOString();
   for (const repoFullName of repos) {
     try {
-      const [gatePrecision, calibration] = await Promise.all([
+      const [gatePrecision, calibration, reviewBurst] = await Promise.all([
         loadGatePrecisionReport(env, repoFullName),
         buildRepoOutcomeCalibration(env, repoFullName),
+        findHottestReviewTargetForRepo(env, repoFullName, reviewBurstSinceIso),
       ]);
       rows.push({
         repoFullName,
@@ -196,7 +228,7 @@ export async function computeOpsStats(env: Env): Promise<OpsStatsPayload> {
           discriminates: calibration.slop.discriminates,
         },
         recommendations: calibration.recommendations,
-        anomalies: detectOutcomeAnomalies({ repoFullName, gatePrecision, calibration }),
+        anomalies: detectOutcomeAnomalies({ repoFullName, gatePrecision, calibration, reviewBurst }),
       });
     } catch {
       /* a per-repo failure must not blank the whole feed */

test/unit/db-parsers.test.ts

@@ -4,6 +4,7 @@ import {
   countRecentDeadLetters,
   countRecentDeadLettersByType,
   countRecentAuditEventsForActorAndTarget,
+  findHottestReviewTargetForRepo,
   hasAuditEventForDelivery,
   getLatestScorePreview,
   getRepoAuthorPullRequestHistory,
@@ -556,6 +557,33 @@ describe("database row parser hardening", () => {
     expect(await countRecentAuditEventsForActorAndTarget(env, "chatty", "github_app.review_nag_ping", "owner/repo#1", "2026-06-24T13:00:00.000Z")).toBe(0); // none after the cutoff → count(*) returns 0
   });
 
+  it("findHottestReviewTargetForRepo returns the PR with the most published surfaces in the window, scoped to ONE repo (#orb-ci-stuck-repeat)", async () => {
+    const env = createTestEnv();
+    const publish = (targetKey: string, createdAt: string) =>
+      recordAuditEvent(env, { eventType: "github_app.pr_public_surface_published", actor: "contributor", targetKey, outcome: "completed", createdAt });
+    // owner/repo#1: 3 publishes in-window -- the hottest target for this repo.
+    await publish("owner/repo#1", "2026-06-24T10:00:00.000Z");
+    await publish("owner/repo#1", "2026-06-24T10:05:00.000Z");
+    await publish("owner/repo#1", "2026-06-24T10:10:00.000Z");
+    // owner/repo#2: only 1 publish -- must not win over #1.
+    await publish("owner/repo#2", "2026-06-24T10:00:00.000Z");
+    // A DIFFERENT event type on the SAME PR must not count (the eventType filter).
+    await recordAuditEvent(env, { eventType: "github_app.ai_review_cache_hit", actor: "contributor", targetKey: "owner/repo#1", outcome: "completed", createdAt: "2026-06-24T10:07:00.000Z" });
+    // A DIFFERENT repo with an overlapping numeric suffix must not leak into this repo's count (the LIKE scope).
+    await publish("owner/repo-fork#1", "2026-06-24T10:00:00.000Z");
+    await publish("owner/repo-fork#1", "2026-06-24T10:05:00.000Z");
+    await publish("owner/repo-fork#1", "2026-06-24T10:06:00.000Z");
+    await publish("owner/repo-fork#1", "2026-06-24T10:07:00.000Z");
+
+    const hottest = await findHottestReviewTargetForRepo(env, "owner/repo", "2026-06-24T09:00:00.000Z");
+    expect(hottest).toEqual({ targetKey: "owner/repo#1", count: 3 });
+
+    // A cutoff AFTER all the recorded publishes must find nothing.
+    expect(await findHottestReviewTargetForRepo(env, "owner/repo", "2026-06-24T11:00:00.000Z")).toBeNull();
+    // An unregistered/unpublished repo must find nothing.
+    expect(await findHottestReviewTargetForRepo(env, "owner/nothing-here", "2026-06-24T09:00:00.000Z")).toBeNull();
+  });
+
   it("hasAuditEventForDelivery finds a matching deliveryId inside metadata_json, scoped to actor+eventType+targetKey (#2560)", async () => {
     const env = createTestEnv();
     await recordAuditEvent(env, {