diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index 0ca00d714c..bea3639208 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -25,6 +25,7 @@ import { performRepoDocRefresh } from "../github/repo-doc-refresh-runner"; import { executeAgentRun } from "../services/agent-orchestrator"; import { deliverNotification, evaluateNotificationEvent } from "../notifications/service"; import { isOpsEnabled, resolveOpsManifestOverride, runOpsAlerts } from "../review/ops-wire"; +import { runAnomalyAlertsWired } from "../review/alerts-wire"; import { isSweepWatchdogEnabled, resolveSweepWatchdogManifestOverride, runSweepLivenessWatchdog } from "../review/sweep-watchdog"; import { isLoopEscalationSweepEnabled, resolveLoopEscalationManifestOverride, runLoopEscalationSweep } from "../review/loop-escalation-wire"; import { isPrReconciliationEnabled, resolvePrReconciliationManifestOverride, runOpenPrReconciliation } from "../review/pr-reconciliation"; @@ -336,7 +337,12 @@ export async function processJob(env: Env, message: JobMessage): Promise { // (env OR manifest) must still no-op, so disabled does zero work here too. Read-only telemetry — never // throws into the queue. const opsManifestOverride = await resolveOpsManifestOverride(env); - if (isOpsEnabled(env, opsManifestOverride)) await runOpsAlerts(env); + if (isOpsEnabled(env, opsManifestOverride)) { + await runOpsAlerts(env); + // Same tick, second channel: the ported Discord anomaly-alerter (#8905). Self-gated on a configured + // DISCORD_WEBHOOK_URL and fail-safe, so it is a no-op until an operator wires the channel. + await runAnomalyAlertsWired(env); + } return; } case "sweep-liveness-watchdog": diff --git a/src/review/alerts-wire.ts b/src/review/alerts-wire.ts new file mode 100644 index 0000000000..385634a474 --- /dev/null +++ b/src/review/alerts-wire.ts @@ -0,0 +1,40 @@ +// Wire the ported Discord anomaly-alerter (src/review/alerts.ts) into the host cron. alerts.ts is a +// self-contained, dependency-injected port: it computes no snapshots itself, so the host builds the +// operator agent-config from env and injects the native ops-port health/calibration snapshots +// (src/review/ops.ts — the same pair operator-dashboard already reads). Runs on the SAME cron tick as +// runOpsAlerts so the Discord channel fires alongside the structured-log + PagerDuty path (#8905). +// +// FAILS SAFE: an alerting side-channel must never break the cron. Unlike runOpsAlerts' pure reads, +// runAnomalyAlerts writes throttle-claim rows, so a storage error is possible; a top-level error is +// swallowed to a structured log, mirroring runOpsAlerts' own "never throws into the queue" contract. +import { runAnomalyAlerts, type AlertAgentConfig, type AnomalyAlertDeps } from "./alerts"; +import { computeAgentHealth, computeCalibration } from "./ops"; + +/** + * Build the operator agent-config, inject the native health/calibration snapshots, and fire the Discord + * anomaly alert. No-op unless a valid DISCORD_WEBHOOK_URL is configured — runAnomalyAlerts self-gates on + * the webhook before any storage write, so an operator who has not wired a channel pays nothing. + * + * Caller invokes this only from the flag-ON ops cron path (alongside runOpsAlerts), so flag-OFF it is + * never reached and the cron does zero new work. + */ +export async function runAnomalyAlertsWired(env: Env): Promise { + // The same slug fallback operator-dashboard / api routes use to key this deployment's review_targets rows. + const slug = env.GITHUB_APP_SLUG?.trim() || "loopover"; + const config: AlertAgentConfig = { + slug, + // discordNotify is always ON here; the REAL gate is the webhook — resolveWebhook returns "" when + // DISCORD_WEBHOOK_URL is unset and runAnomalyAlerts then returns before touching storage. + features: { discordNotify: true }, + secrets: { discordWebhook: "DISCORD_WEBHOOK_URL" }, + }; + const deps: AnomalyAlertDeps = { + computeAgentHealth: (alertEnv, alertConfig) => computeAgentHealth(alertEnv, { slug: alertConfig.slug, secrets: {} }), + computeCalibration: (alertEnv, alertConfig) => computeCalibration(alertEnv, { slug: alertConfig.slug, secrets: {} }), + }; + try { + await runAnomalyAlerts(env, config, deps); + } catch (error) { + console.error(JSON.stringify({ level: "error", event: "anomaly_alert_wire_error", message: String(error).slice(0, 200) })); + } +} diff --git a/test/unit/anomaly-alerts-wire.test.ts b/test/unit/anomaly-alerts-wire.test.ts new file mode 100644 index 0000000000..198a81ecf1 --- /dev/null +++ b/test/unit/anomaly-alerts-wire.test.ts @@ -0,0 +1,93 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { runAnomalyAlertsWired } from "../../src/review/alerts-wire"; +import { runAnomalyAlerts, type AlertAgentConfig, type AnomalyAlertDeps } from "../../src/review/alerts"; +import { computeAgentHealth, computeCalibration } from "../../src/review/ops"; + +vi.mock("../../src/review/alerts", () => ({ runAnomalyAlerts: vi.fn() })); +vi.mock("../../src/review/ops", () => ({ computeAgentHealth: vi.fn(), computeCalibration: vi.fn() })); + +const runAnomalyAlertsMock = vi.mocked(runAnomalyAlerts); +const computeAgentHealthMock = vi.mocked(computeAgentHealth); +const computeCalibrationMock = vi.mocked(computeCalibration); + +function lastConfig(): AlertAgentConfig { + return runAnomalyAlertsMock.mock.calls.at(-1)?.[1] as AlertAgentConfig; +} +function lastDeps(): AnomalyAlertDeps { + return runAnomalyAlertsMock.mock.calls.at(-1)?.[2] as AnomalyAlertDeps; +} + +describe("runAnomalyAlertsWired", () => { + afterEach(() => { + vi.restoreAllMocks(); + vi.clearAllMocks(); + }); + + it("fires runAnomalyAlerts with the configured GITHUB_APP_SLUG and the DISCORD_WEBHOOK_URL secret ref", async () => { + runAnomalyAlertsMock.mockResolvedValue(undefined); + const env = { GITHUB_APP_SLUG: " my-agent " } as unknown as Env; + + await runAnomalyAlertsWired(env); + + expect(runAnomalyAlertsMock).toHaveBeenCalledTimes(1); + expect(runAnomalyAlertsMock).toHaveBeenCalledWith(env, expect.anything(), expect.anything()); + expect(lastConfig()).toEqual({ + slug: "my-agent", // trimmed + features: { discordNotify: true }, + secrets: { discordWebhook: "DISCORD_WEBHOOK_URL" }, + }); + }); + + it("falls back to the 'loopover' slug when GITHUB_APP_SLUG is unset", async () => { + runAnomalyAlertsMock.mockResolvedValue(undefined); + const env = {} as unknown as Env; + + await runAnomalyAlertsWired(env); + + expect(lastConfig().slug).toBe("loopover"); + }); + + it("falls back to 'loopover' when GITHUB_APP_SLUG is blank/whitespace-only", async () => { + runAnomalyAlertsMock.mockResolvedValue(undefined); + const env = { GITHUB_APP_SLUG: " " } as unknown as Env; + + await runAnomalyAlertsWired(env); + + expect(lastConfig().slug).toBe("loopover"); + }); + + it("injects health/calibration deps that delegate to the native ops port keyed by the config slug", async () => { + runAnomalyAlertsMock.mockResolvedValue(undefined); + const health = { manualRate: 0 } as unknown as Awaited>; + const calibration = { currentFloor: 0 } as unknown as Awaited>; + computeAgentHealthMock.mockResolvedValue(health); + computeCalibrationMock.mockResolvedValue(calibration); + const env = { GITHUB_APP_SLUG: "agent-x" } as unknown as Env; + + await runAnomalyAlertsWired(env); + + const deps = lastDeps(); + const config = lastConfig(); + await expect(deps.computeAgentHealth(env, config)).resolves.toBe(health); + await expect(deps.computeCalibration(env, config)).resolves.toBe(calibration); + expect(computeAgentHealthMock).toHaveBeenCalledWith(env, { slug: "agent-x", secrets: {} }); + expect(computeCalibrationMock).toHaveBeenCalledWith(env, { slug: "agent-x", secrets: {} }); + }); + + it("fails safe: swallows a runAnomalyAlerts error and logs a structured wire-error line", async () => { + runAnomalyAlertsMock.mockRejectedValue(new Error("no such column: project")); + const errors: string[] = []; + vi.spyOn(console, "error").mockImplementation((...args: unknown[]) => { + errors.push(String(args[0])); + }); + const env = { GITHUB_APP_SLUG: "agent-y" } as unknown as Env; + + await expect(runAnomalyAlertsWired(env)).resolves.toBeUndefined(); + + expect(errors).toHaveLength(1); + const log = JSON.parse(errors[0] ?? "{}") as Record; + expect(log).toMatchObject({ level: "error", event: "anomaly_alert_wire_error" }); + expect(String(log.message)).toContain("no such column: project"); + }); +}); diff --git a/test/unit/ops-alerts-cron-wire.test.ts b/test/unit/ops-alerts-cron-wire.test.ts new file mode 100644 index 0000000000..50b8fcf917 --- /dev/null +++ b/test/unit/ops-alerts-cron-wire.test.ts @@ -0,0 +1,60 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { processJob } from "../../src/queue/job-dispatch"; +import { isOpsEnabled, runOpsAlerts } from "../../src/review/ops-wire"; +import { runAnomalyAlerts } from "../../src/review/alerts"; +import type { JobMessage } from "../../src/types"; + +vi.mock("../../src/review/ops-wire", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + runOpsAlerts: vi.fn().mockResolvedValue({}), + isOpsEnabled: vi.fn().mockReturnValue(true), + resolveOpsManifestOverride: vi.fn().mockResolvedValue(null), + }; +}); +// Mock only the leaf alerter (the real alerts-wire adapter still runs, so its config/dep wiring is exercised +// end-to-end) — its D1 throttle-claim writes are what we do NOT want to hit here. +vi.mock("../../src/review/alerts", () => ({ runAnomalyAlerts: vi.fn().mockResolvedValue(undefined) })); + +const runOpsAlertsMock = vi.mocked(runOpsAlerts); +const runAnomalyAlertsMock = vi.mocked(runAnomalyAlerts); +const isOpsEnabledMock = vi.mocked(isOpsEnabled); + +const opsAlertsJob = { type: "ops-alerts", requestedBy: "schedule" } as unknown as JobMessage; + +describe("processJob ops-alerts cron wiring (#8905)", () => { + beforeEach(() => { + isOpsEnabledMock.mockReturnValue(true); + }); + afterEach(() => { + vi.clearAllMocks(); + }); + + it("fires BOTH runOpsAlerts (PagerDuty/log) and runAnomalyAlerts (Discord) on the enabled cron tick", async () => { + const env = { GITHUB_APP_SLUG: "loopover" } as unknown as Env; + + await processJob(env, opsAlertsJob); + + expect(runOpsAlertsMock).toHaveBeenCalledTimes(1); + expect(runOpsAlertsMock).toHaveBeenCalledWith(env); + // The Discord channel fires through the real alerts-wire adapter on the same tick. + expect(runAnomalyAlertsMock).toHaveBeenCalledTimes(1); + expect(runAnomalyAlertsMock).toHaveBeenCalledWith( + env, + expect.objectContaining({ slug: "loopover", features: { discordNotify: true } }), + expect.objectContaining({ computeAgentHealth: expect.any(Function), computeCalibration: expect.any(Function) }), + ); + }); + + it("does NEITHER when ops is disabled (a stale in-flight job after a flag-flip no-ops)", async () => { + isOpsEnabledMock.mockReturnValue(false); + const env = {} as unknown as Env; + + await processJob(env, opsAlertsJob); + + expect(runOpsAlertsMock).not.toHaveBeenCalled(); + expect(runAnomalyAlertsMock).not.toHaveBeenCalled(); + }); +});