From cc88c6d9cb828575f22fbd8948dba7c05c79967f Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 6 Jul 2026 01:07:25 -0700 Subject: [PATCH] fix(observability): forward ops-alerts anomalies to Sentry, add a review-burst rule 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. --- src/db/repositories.ts | 36 ++++++++++++++++++++- src/review/ops-wire.ts | 60 ++++++++++++++++++++++++++-------- test/unit/db-parsers.test.ts | 28 ++++++++++++++++ test/unit/ops-wire.test.ts | 62 +++++++++++++++++++++++++++++------- test/unit/queue.test.ts | 6 ++-- 5 files changed, 162 insertions(+), 30 deletions(-) diff --git a/src/db/repositories.ts b/src/db/repositories.ts index 735fd8dcec..7fe6d345a4 100644 --- a/src/db/repositories.ts +++ b/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`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 diff --git a/src/review/ops-wire.ts b/src/review/ops-wire.ts index c9ff7d5d5b..2f8e35c7c6 100644 --- a/src/review/ops-wire.ts +++ b/src/review/ops-wire.ts @@ -14,9 +14,15 @@ // 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 @@ -24,7 +30,7 @@ // `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 { */ export async function runOpsAlerts(env: Env): Promise> { const found: Record = {}; + 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 { 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 { 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 */ diff --git a/test/unit/db-parsers.test.ts b/test/unit/db-parsers.test.ts index 3610d2a987..136a830a49 100644 --- a/test/unit/db-parsers.test.ts +++ b/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, { diff --git a/test/unit/ops-wire.test.ts b/test/unit/ops-wire.test.ts index 420a846d7e..3f47b1f9eb 100644 --- a/test/unit/ops-wire.test.ts +++ b/test/unit/ops-wire.test.ts @@ -1,6 +1,6 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { createApp } from "../../src/api/routes"; -import { recordGateBlockOutcome, upsertPullRequestFromGitHub } from "../../src/db/repositories"; +import { recordAuditEvent, recordGateBlockOutcome, upsertPullRequestFromGitHub } from "../../src/db/repositories"; import { computeOpsStats, detectOutcomeAnomalies, @@ -127,6 +127,22 @@ describe("detectOutcomeAnomalies — over gittensory's own outcome data", () => }; expect(detectOutcomeAnomalies(snap).length).toBe(3); }); + + it("flags a review burst — the same PR published far more review surfaces than normal iteration in the window (#orb-ci-stuck-repeat)", () => { + const snap: RepoOutcomeSnapshot = { ...healthySnapshot, reviewBurst: { targetKey: "owner/repo#42", count: 9 } }; + const out = detectOutcomeAnomalies(snap); + expect(out.some((a) => /review burst/.test(a) && /owner\/repo#42/.test(a) && /9 review surfaces/.test(a))).toBe(true); + }); + + it("does NOT flag a review burst below the threshold", () => { + const snap: RepoOutcomeSnapshot = { ...healthySnapshot, reviewBurst: { targetKey: "owner/repo#42", count: 2 } }; + expect(detectOutcomeAnomalies(snap).some((a) => /review burst/.test(a))).toBe(false); + }); + + it("does NOT flag a review burst when none was computed (absent/null)", () => { + expect(detectOutcomeAnomalies({ ...healthySnapshot, reviewBurst: null }).some((a) => /review burst/.test(a))).toBe(false); + expect(detectOutcomeAnomalies(healthySnapshot).some((a) => /review burst/.test(a))).toBe(false); // field omitted entirely + }); }); // ── DB-backed cron + endpoint integration over the real migrated schema ───────────────────────────────────── @@ -157,19 +173,20 @@ async function seedGateFalsePositiveAnomaly(env: Env, repoFullName: string): Pro describe("runOpsAlerts — cron path over gittensory's outcome data", () => { afterEach(() => vi.restoreAllMocks()); - it("emits a structured ops_anomaly log naming the repo + drift on a seeded anomaly", async () => { + it("emits a structured ops_anomaly log naming the repo + drift on a seeded anomaly, at error level (#orb-ci-stuck-repeat -- so it reaches Sentry)", async () => { const env = createTestEnv(); await seedRegisteredRepo(env, "owner/repo"); await seedGateFalsePositiveAnomaly(env, "owner/repo"); - const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const errors = vi.spyOn(console, "error").mockImplementation(() => {}); const found = await runOpsAlerts(env); expect(found["owner/repo"]?.some((a) => /gate false-positive spike/.test(a))).toBe(true); - const logged = warn.mock.calls.map((c) => String(c[0])).find((line) => line.includes("ops_anomaly") && line.includes("owner/repo")); + const logged = errors.mock.calls.map((c) => String(c[0])).find((line) => line.includes("ops_anomaly") && line.includes("owner/repo")); expect(logged).toBeDefined(); - const parsed = JSON.parse(logged!) as { ev: string; repo: string; anomalies: string[] }; - expect(parsed.ev).toBe("ops_anomaly"); + const parsed = JSON.parse(logged!) as { level: string; event: string; repo: string; anomalies: string[] }; + expect(parsed.level).toBe("error"); + expect(parsed.event).toBe("ops_anomaly"); expect(parsed.repo).toBe("owner/repo"); expect(parsed.anomalies.some((a) => /missing_linked_issue/.test(a))).toBe(true); }); @@ -182,12 +199,33 @@ describe("runOpsAlerts — cron path over gittensory's outcome data", () => { await recordGateBlockOutcome(env, { repoFullName: "owner/clean", pullNumber: i, blockerCodes: ["slop_risk"] }); await upsertPullRequestFromGitHub(env, "owner/clean", { number: i, title: `PR ${i}`, state: "closed", merged_at: null } as never); } - const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const errors = vi.spyOn(console, "error").mockImplementation(() => {}); const found = await runOpsAlerts(env); expect(found["owner/clean"]).toBeUndefined(); - expect(warn.mock.calls.map((c) => String(c[0])).some((line) => line.includes("ops_anomaly\""))).toBe(false); + expect(errors.mock.calls.map((c) => String(c[0])).some((line) => line.includes("ops_anomaly\""))).toBe(false); + }); + + it("detects and reports a review burst end-to-end (a PR published far more review surfaces than normal in the window)", async () => { + const env = createTestEnv(); + await seedRegisteredRepo(env, "owner/repo"); + for (let i = 0; i < 7; i += 1) { + await recordAuditEvent(env, { + eventType: "github_app.pr_public_surface_published", + actor: "contributor", + targetKey: "owner/repo#99", + outcome: "completed", + }); + } + const errors = vi.spyOn(console, "error").mockImplementation(() => {}); + + const found = await runOpsAlerts(env); + + expect(found["owner/repo"]?.some((a) => /review burst/.test(a) && /owner\/repo#99/.test(a))).toBe(true); + const stats = await computeOpsStats(env); + const row = stats.repos.find((r) => r.repoFullName === "owner/repo"); + expect(row?.anomalies.some((a) => /review burst/.test(a))).toBe(true); }); it("fails safe per-repo: a load error on one repo is logged and the scan continues (ops_anomaly_repo_error)", async () => { @@ -196,12 +234,12 @@ describe("runOpsAlerts — cron path over gittensory's outcome data", () => { // The repo is scanned, but the per-repo precision load throws → caught at the inner catch. // gate-precision reads pull_requests (Drizzle, quoted table name) per repo. poisonDbPrepare(env, /"pull_requests"/i); - const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const errors = vi.spyOn(console, "error").mockImplementation(() => {}); const found = await runOpsAlerts(env); // resolves (never throws) expect(found).toEqual({}); - expect(warn.mock.calls.map((c) => String(c[0])).some((line) => line.includes("ops_anomaly_repo_error") && line.includes("owner/repo"))).toBe(true); + expect(errors.mock.calls.map((c) => String(c[0])).some((line) => line.includes("ops_anomaly_repo_error") && line.includes("owner/repo"))).toBe(true); }); it("fails safe at the top level: a repo-scan error is swallowed (ops_anomaly_error), returns {}", async () => { @@ -209,12 +247,12 @@ describe("runOpsAlerts — cron path over gittensory's outcome data", () => { // opsScanRepos → listRepositories reads the repositories table (Drizzle, quoted); poison it so the // outer try throws. poisonDbPrepare(env, /"repositories"/i); - const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const errors = vi.spyOn(console, "error").mockImplementation(() => {}); const found = await runOpsAlerts(env); // resolves (never throws) expect(found).toEqual({}); - expect(warn.mock.calls.map((c) => String(c[0])).some((line) => line.includes("ops_anomaly_error"))).toBe(true); + expect(errors.mock.calls.map((c) => String(c[0])).some((line) => line.includes("ops_anomaly_error"))).toBe(true); }); }); diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index f3b00f04cc..088218c047 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -20990,10 +20990,10 @@ describe("queue processors", () => { await recordGateBlockOutcome(env, { repoFullName: "owner/repo", pullNumber: i, blockerCodes: ["missing_linked_issue"] }); await upsertPullRequestFromGitHub(env, "owner/repo", { number: i, title: `PR ${i}`, state: "closed", merged_at: i <= 4 ? "2026-06-01T00:00:00.000Z" : null } as never); } - const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const errors = vi.spyOn(console, "error").mockImplementation(() => {}); await processJob(env, { type: "ops-alerts", requestedBy: "test" }); - expect(warn.mock.calls.map((c) => String(c[0])).some((line) => line.includes("ops_anomaly") && line.includes("owner/repo"))).toBe(true); - warn.mockRestore(); + expect(errors.mock.calls.map((c) => String(c[0])).some((line) => line.includes("ops_anomaly") && line.includes("owner/repo"))).toBe(true); + errors.mockRestore(); }); describe("type label decoupling (#label-decoupling)", () => {