diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index bfdd844..712043e 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -101,6 +101,12 @@ class Settings(BaseSettings): # Defaults to "*" for local development only. cors_origins: list[str] = ["*"] + # Discord webhook URL that app/internal/ai_health.py posts to when an AI + # tier (transcription/correlation) transitions into or out of degraded + # state. Empty disables the POST entirely — not every self-hosted + # deployment will set this up, and skipping it must be silent. + ai_alert_webhook_url: str = "" + class Config: env_file = ".env" diff --git a/drb-c2-core/app/internal/ai_health.py b/drb-c2-core/app/internal/ai_health.py new file mode 100644 index 0000000..f14af5a --- /dev/null +++ b/drb-c2-core/app/internal/ai_health.py @@ -0,0 +1,192 @@ +""" +Shared AI-provider degradation registry. + +On the night of 2026-08-18 three independent AI dependency failures (a +retired Gemini model ID, a depleted Gemini balance, an unpayable OpenAI +account) each surfaced only as a single ERROR log line -- and nobody reads +container logs continuously. This module is the fix: every AI call site +reports its outcome here instead of (or in addition to) just logging, so the +current state of every AI tier can be read back over HTTP (see +app/main.py's /health/ai) and pushed out to Discord on state changes. + +Tiers are tracked independently and in memory only (module-level singleton, +no Firestore/DI -- consistent with the rest of this codebase). State is lost +on restart, which is fine: a fresh process should re-derive degradation from +the next few calls rather than resurrect a possibly-stale alert. + +The load-bearing distinction, from the incident this module exists to +prevent: a PERMANENT condition (retired model, dead billing account, bad API +key) will never clear on its own and must alert on the very first +occurrence. A TRANSIENT condition (rate limit, network blip) clears by +itself constantly and must NOT page anyone for the first failure -- only if +it persists. classify() is the one place that tells the two apart from a +provider error message, because both this module's callers (llm_correlator.py, +transcription.py) need the exact same judgment call and must not each grow +their own slightly-different copy that drifts. +""" +import asyncio +from datetime import datetime, timezone +from typing import Optional + +from app.internal.logger import logger +from app.config import settings + +TIERS = ("transcription", "correlation_cheap", "correlation_smart", "extraction") + +# Consecutive failures a TRANSIENT condition must reach before it alerts. +# Permanent conditions skip this entirely and alert on failure #1. +TRANSIENT_ALERT_THRESHOLD = 5 + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _default_state() -> dict: + return { + "degraded": False, + "permanent": False, + "provider": None, + "model": None, + "problem": None, + "fix": None, + "first_seen": None, + "last_seen": None, + "consecutive_failures": 0, + "alerted": False, + } + + +_state: dict[str, dict] = {t: _default_state() for t in TIERS} + + +def classify(text: str) -> str: + """ + Classify a provider failure message body. + + Returns "dead_model", "billing", or "transient". + + A depleted balance and an ordinary rate limit both arrive as HTTP 429 -- + the status code can't tell them apart, only the message body can. This + logic previously lived independently in llm_correlator.py and (in a + slightly different shape) transcription.py; it now lives here once, and + both call in rather than re-matching the text themselves. + """ + low = text.lower() + + if "404" in text or "not found" in low or "no longer available" in low: + return "dead_model" + + if ( + "credits are depleted" in low + or "prepayment" in low + or "billing" in low + or "insufficient_quota" in low + or "credit" in low + or "exceeded your current quota" in low + ): + return "billing" + + return "transient" + + +async def report_degraded( + tier: str, + provider: str, + model: str, + problem: str, + fix: str, + permanent: bool = False, +) -> None: + """ + Record a failure for `tier`. Call this from a failure path, once per + failure (it does its own once-per-episode alert suppression -- do not + gate the call site on that yourself). + + permanent=True (dead model, unpayable account, bad key) alerts on this + very call. permanent=False (rate limit, network blip) only alerts once + TRANSIENT_ALERT_THRESHOLD consecutive failures have been reported for + this tier, so an ordinary blip never pages anyone. + """ + if tier not in _state: + _state[tier] = _default_state() + entry = _state[tier] + + now = _now() + if entry["consecutive_failures"] == 0: + entry["first_seen"] = now + entry["last_seen"] = now + entry["consecutive_failures"] += 1 + entry["provider"] = provider + entry["model"] = model + entry["problem"] = problem + entry["fix"] = fix + entry["permanent"] = permanent + + should_alert_now = permanent or entry["consecutive_failures"] >= TRANSIENT_ALERT_THRESHOLD + + if should_alert_now and not entry["degraded"]: + entry["degraded"] = True + + if should_alert_now and not entry["alerted"]: + entry["alerted"] = True + await _post_webhook( + f"**AI tier degraded: {tier}**\n" + f"Provider: {provider} ({model})\n" + f"Problem: {problem}\n" + f"Fix: {fix}\n" + f"Kind: {'permanent' if permanent else 'transient, persisted ' + str(entry['consecutive_failures']) + ' calls'}" + ) + + +async def report_healthy(tier: str) -> None: + """ + Record a successful call for `tier`. Call this on every success, not + just after a failure -- it is what lets a degraded tier recover on its + own instead of staying red forever after one transient blip. + """ + if tier not in _state: + _state[tier] = _default_state() + entry = _state[tier] + + was_alerted = entry["alerted"] + was_degraded = entry["degraded"] + provider, model = entry["provider"], entry["model"] + + _state[tier] = _default_state() + # Keep the last-known provider/model around for the recovery message + # and for a quick glance at snapshot() even when healthy. + _state[tier]["provider"] = provider + _state[tier]["model"] = model + + if was_alerted: + await _post_webhook(f"**AI tier recovered: {tier}**\nProvider: {provider} ({model})") + elif was_degraded: + # Reached "degraded" internally but never crossed the alert + # threshold before recovering -- nothing was ever posted, so + # nothing needs un-posting. Nothing to do. + pass + + +def snapshot() -> dict: + """Current state of every tier, for /health/ai.""" + return {tier: dict(entry) for tier, entry in _state.items()} + + +async def _post_webhook(content: str) -> None: + """ + POST a message to the AI-alert Discord webhook, if one is configured. + + Same httpx pattern as app/internal/alerter.py's _post_webhook: short + timeout, never raises. Self-hosted deployments that don't set + ai_alert_webhook_url just skip this silently. + """ + url = settings.ai_alert_webhook_url + if not url: + return + try: + import httpx + async with httpx.AsyncClient(timeout=5.0) as client: + await client.post(url, json={"content": content}) + except Exception as e: + logger.warning(f"ai_health: Discord webhook POST failed: {e}") diff --git a/drb-c2-core/app/internal/llm_correlator.py b/drb-c2-core/app/internal/llm_correlator.py index e51f85b..345373c 100644 --- a/drb-c2-core/app/internal/llm_correlator.py +++ b/drb-c2-core/app/internal/llm_correlator.py @@ -24,6 +24,7 @@ import json from datetime import datetime, timezone from typing import Optional from app.internal.logger import logger +from app.internal import ai_health from app.config import settings @@ -239,18 +240,21 @@ async def decide(call_id: str, ctx: dict) -> Optional[dict]: f"action={decision['action']} incident={_id} " f"reasoning={decision['reasoning']!r}" ) + await ai_health.report_healthy("correlation_cheap") return decision except Exception as e: - _log_llm_failure("LLM correlator", call_id, settings.corr_cheap_model, e) + await _log_llm_failure("LLM correlator", "correlation_cheap", call_id, settings.corr_cheap_model, e) return None _dead_models: set[str] = set() -def _log_llm_failure(where: str, call_id: str, model: str, exc: Exception) -> None: +async def _log_llm_failure(where: str, tier: str, call_id: str, model: str, exc: Exception) -> None: """ - Log an LLM failure, escalating a dead model ID to ERROR once per model. + Log an LLM failure, escalating a dead model ID to ERROR once per model, + and report it to the shared app.internal.ai_health registry either way + (which is what drives /health/ai and the Discord degradation alert). A per-call WARNING was the only signal that gemini-2.0-flash had been shut down, and since every failure falls back to the rules decision the pipeline @@ -260,27 +264,32 @@ def _log_llm_failure(where: str, call_id: str, model: str, exc: Exception) -> No that will never fix itself, so it gets ERROR and says what to do. """ text = str(exc) - low = text.lower() + kind = ai_health.classify(text) - if "404" in text or "not found" in low or "no longer available" in low: - _log_tier_down(where, model, "model is unavailable", - "Update CORR_CHEAP_MODEL/CORR_SMART_MODEL in config.py", text) + if kind == "dead_model": + await _log_tier_down(where, tier, model, "model is unavailable", + "Update CORR_CHEAP_MODEL/CORR_SMART_MODEL in config.py", text) return # A depleted balance reads as 429, the same status as an ordinary rate limit, # but it is the opposite kind of problem: a rate limit clears on its own and a - # dead account never does. Matching on the billing wording keeps a burst of - # rate limits at WARNING while an empty account escalates like a bad model ID. - if "credits are depleted" in low or "prepayment" in low or "billing" in low: - _log_tier_down(where, model, "the Gemini account is out of credit", - "Top up billing at https://ai.studio/projects", text) + # dead account never does. ai_health.classify() keeps a burst of rate limits + # at WARNING while an empty account escalates like a bad model ID. + if kind == "billing": + await _log_tier_down(where, tier, model, "the Gemini account is out of credit", + "Top up billing at https://ai.studio/projects", text) return logger.warning(f"{where} failed for call {call_id}: {text}") + await ai_health.report_degraded( + tier, "gemini", model, "transient API error", + "no action needed unless this persists", permanent=False, + ) -def _log_tier_down(where: str, model: str, problem: str, fix: str, text: str) -> None: +async def _log_tier_down(where: str, tier: str, model: str, problem: str, fix: str, text: str) -> None: """ERROR once per model, not once per call — this runs at radio-traffic volume.""" + await ai_health.report_degraded(tier, "gemini", model, problem, fix, permanent=True) if model in _dead_models: return _dead_models.add(model) @@ -306,9 +315,10 @@ async def tiebreak(rules_decision: dict, llm_decision: dict, ctx: dict) -> dict: f"action={decision['action']} incident={_id} " f"reasoning={decision['reasoning']!r}" ) + await ai_health.report_healthy("correlation_smart") return decision except Exception as e: - _log_llm_failure("LLM tiebreak", call_id, settings.corr_smart_model, e) + await _log_llm_failure("LLM tiebreak", "correlation_smart", call_id, settings.corr_smart_model, e) return rules_decision diff --git a/drb-c2-core/app/internal/transcription.py b/drb-c2-core/app/internal/transcription.py index 313ffed..b71ed5f 100644 --- a/drb-c2-core/app/internal/transcription.py +++ b/drb-c2-core/app/internal/transcription.py @@ -11,6 +11,8 @@ import os from typing import Optional from app.internal.logger import logger from app.internal import firestore as fstore +from app.internal import ai_health +from app.config import settings # Whisper treats `prompt` as preceding transcript text, not instructions. # Writing it as actual radio speech primes the vocabulary toward P25 codes @@ -93,12 +95,12 @@ def _is_degenerate(text: str, segments: list[dict]) -> bool: return False -_billing_reported = False - - -def _log_transcribe_failure(call_id: str, exc: Exception) -> None: +async def _log_transcribe_failure(call_id: str, exc: Exception) -> None: """ - Log a transcription failure, escalating an unpayable account to ERROR once. + Log a transcription failure, escalating a permanent condition to ERROR + once (via app.internal.ai_health, which also drives the /health/ai + endpoint and the Discord degradation alert) and reporting it to the + shared registry either way. Transcription failing returns None and the pipeline carries on by design, so a per-call WARNING is invisible: no transcript means no extraction, which @@ -110,23 +112,41 @@ def _log_transcribe_failure(call_id: str, exc: Exception) -> None: The same failure mode already bit the Gemini correlator twice (a retired model ID, then a depleted balance), which is why this is worth the code. """ - global _billing_reported text = str(exc) - low = text.lower() + kind = ai_health.classify(text) - if ("insufficient_quota" in low or "billing" in low - or "credit" in low or "exceeded your current quota" in low): - if not _billing_reported: - _billing_reported = True - logger.error( - "Transcription: the OpenAI account cannot be billed -- EVERY call is " - "now stored with no transcript, so extraction, correlation and " - "incidents are all dead downstream. Top up at " - f"https://platform.openai.com/settings/organization/billing. API said: {text}" - ) + if kind == "billing": + problem = "the OpenAI account cannot be billed" + fix = "top up at https://platform.openai.com/settings/organization/billing" + logger.error( + "Transcription: the OpenAI account cannot be billed -- EVERY call is " + "now stored with no transcript, so extraction, correlation and " + "incidents are all dead downstream. Top up at " + f"https://platform.openai.com/settings/organization/billing. API said: {text}" + ) + await ai_health.report_degraded( + "transcription", "openai", settings.stt_model, problem, fix, permanent=True + ) + return + + if kind == "dead_model": + problem = "the STT model is unavailable" + fix = "update STT_MODEL in config.py" + logger.error( + f"Transcription: the configured model ({settings.stt_model!r}) is unavailable " + "-- EVERY call is now stored with no transcript, so extraction, correlation " + f"and incidents are all dead downstream. Update STT_MODEL in config.py. API said: {text}" + ) + await ai_health.report_degraded( + "transcription", "openai", settings.stt_model, problem, fix, permanent=True + ) return logger.warning(f"Transcription failed for call {call_id}: {text}") + await ai_health.report_degraded( + "transcription", "openai", settings.stt_model, + "transient API error", "no action needed unless this persists", permanent=False, + ) async def transcribe_call( @@ -150,9 +170,15 @@ async def transcribe_call( _sync_transcribe, gcs_uri, talkgroup_name ) except Exception as e: - _log_transcribe_failure(call_id, e) + await _log_transcribe_failure(call_id, e) return None, [] + # No exception means the provider call itself succeeded (this also + # covers transcripts discarded as degenerate/hallucinated output — + # that's a filtering decision, not a provider failure), so the + # transcription tier is healthy and any prior degradation clears. + await ai_health.report_healthy("transcription") + if transcript: updates: dict = {"transcript": transcript} if segments: diff --git a/drb-c2-core/app/main.py b/drb-c2-core/app/main.py index 1f0868f..6df1a55 100644 --- a/drb-c2-core/app/main.py +++ b/drb-c2-core/app/main.py @@ -8,6 +8,7 @@ from app.internal.node_sweeper import sweeper_loop from app.internal.summarizer import summarizer_loop from app.internal.vocabulary_learner import vocabulary_induction_loop from app.internal.recorrelation_sweep import recorrelation_loop +from app.internal import ai_health from app.config import settings from app.internal.auth import ( require_firebase_token, @@ -120,3 +121,12 @@ app.include_router(media.router) @app.get("/health") async def health(): return {"ok": True, "mqtt_connected": mqtt_handler.is_connected} + + +# Deliberately unauthenticated, same as /health above: the CI deploy step +# curls /health with no credentials, and this is diagnostic state (which AI +# tier is degraded and why), not a secret — no API keys or tokens appear in +# it. Keeping it auth-free means an external uptime check can watch it too. +@app.get("/health/ai") +async def health_ai(): + return {"tiers": ai_health.snapshot()} diff --git a/drb-c2-core/tests/test_ai_health.py b/drb-c2-core/tests/test_ai_health.py new file mode 100644 index 0000000..d75e419 --- /dev/null +++ b/drb-c2-core/tests/test_ai_health.py @@ -0,0 +1,201 @@ +""" +Unit tests for app.internal.ai_health — the shared AI-provider degradation +registry added for logan/server-26#14 (no alerting when a provider account +runs dry or a model is retired). + +Covers: + * classify() telling a permanent condition (dead model, depleted billing — + both of which can arrive as the same HTTP status a rate limit uses) apart + from a transient one. + * report_degraded() alerting immediately for a permanent condition but only + after TRANSIENT_ALERT_THRESHOLD consecutive failures for a transient one. + * Alerting exactly once per episode, not once per call, and again exactly + once on recovery. + * report_healthy() clearing degraded state so a later re-degradation can + alert again (a fresh episode, not a continuation of the old one). + +The Discord webhook is patched at ai_health._post_webhook so no real HTTP is +made; settings.ai_alert_webhook_url is irrelevant to these tests since +_post_webhook itself is replaced. +""" +import pytest +from unittest.mock import AsyncMock, patch + +from app.internal import ai_health + + +@pytest.fixture(autouse=True) +def _reset_state(): + """Every test gets a clean registry — module-level state persists otherwise.""" + ai_health._state = {t: ai_health._default_state() for t in ai_health.TIERS} + yield + ai_health._state = {t: ai_health._default_state() for t in ai_health.TIERS} + + +# --------------------------------------------------------------------------- +# classify() +# --------------------------------------------------------------------------- + +def test_classify_dead_model_404(): + assert ai_health.classify("404 models/gemini-2.0-flash is not found") == "dead_model" + + +def test_classify_dead_model_no_longer_available(): + assert ai_health.classify("this model is no longer available") == "dead_model" + + +def test_classify_billing_depleted_credits(): + # The exact wording that bit the Gemini correlator on 2026-08-18. + assert ai_health.classify("429 prepayment credits are depleted") == "billing" + + +def test_classify_billing_openai_insufficient_quota(): + assert ai_health.classify("Error: insufficient_quota — exceeded your current quota") == "billing" + + +def test_classify_ordinary_rate_limit_is_transient(): + # Same HTTP status (429) as the depleted-balance case, but no billing + # wording — this must NOT be classified as billing. + assert ai_health.classify("429 Too Many Requests, please retry later") == "transient" + + +def test_classify_network_error_is_transient(): + assert ai_health.classify("Connection reset by peer") == "transient" + + +# --------------------------------------------------------------------------- +# report_degraded — permanent alerts immediately +# --------------------------------------------------------------------------- + +async def test_permanent_failure_alerts_on_first_occurrence(): + with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: + await ai_health.report_degraded( + "correlation_cheap", "gemini", "gemini-2.0-flash", + "model is unavailable", "update the model ID", permanent=True, + ) + webhook.assert_awaited_once() + state = ai_health.snapshot()["correlation_cheap"] + assert state["degraded"] is True + assert state["alerted"] is True + assert state["permanent"] is True + + +async def test_permanent_failure_alerts_only_once_per_episode(): + with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: + for _ in range(5): + await ai_health.report_degraded( + "correlation_cheap", "gemini", "gemini-2.0-flash", + "model is unavailable", "update the model ID", permanent=True, + ) + # Once per episode, not once per call — this runs at radio-traffic volume. + webhook.assert_awaited_once() + assert ai_health.snapshot()["correlation_cheap"]["consecutive_failures"] == 5 + + +# --------------------------------------------------------------------------- +# report_degraded — transient only alerts once it persists +# --------------------------------------------------------------------------- + +async def test_transient_failure_does_not_alert_below_threshold(): + with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: + for _ in range(ai_health.TRANSIENT_ALERT_THRESHOLD - 1): + await ai_health.report_degraded( + "transcription", "openai", "whisper-1", + "transient API error", "no action needed unless this persists", + permanent=False, + ) + webhook.assert_not_awaited() + state = ai_health.snapshot()["transcription"] + assert state["degraded"] is False + assert state["alerted"] is False + + +async def test_transient_failure_alerts_once_threshold_crossed(): + with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: + for _ in range(ai_health.TRANSIENT_ALERT_THRESHOLD): + await ai_health.report_degraded( + "transcription", "openai", "whisper-1", + "transient API error", "no action needed unless this persists", + permanent=False, + ) + webhook.assert_awaited_once() + + # Further failures in the same episode must not re-alert. + await ai_health.report_degraded( + "transcription", "openai", "whisper-1", + "transient API error", "no action needed unless this persists", + permanent=False, + ) + webhook.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# report_healthy — recovery +# --------------------------------------------------------------------------- + +async def test_recovery_alerts_once_after_an_alerted_episode(): + with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: + await ai_health.report_degraded( + "correlation_smart", "gemini", "gemini-1.5-pro", + "the Gemini account is out of credit", "top up billing", permanent=True, + ) + webhook.reset_mock() + + await ai_health.report_healthy("correlation_smart") + + webhook.assert_awaited_once() + state = ai_health.snapshot()["correlation_smart"] + assert state["degraded"] is False + assert state["alerted"] is False + assert state["consecutive_failures"] == 0 + + +async def test_recovery_from_never_alerted_transient_state_is_silent(): + with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: + # Two failures — below the transient threshold, never alerted. + await ai_health.report_degraded( + "extraction", "gemini", "gemini-3.6-flash", + "transient API error", "no action needed unless this persists", + permanent=False, + ) + await ai_health.report_degraded( + "extraction", "gemini", "gemini-3.6-flash", + "transient API error", "no action needed unless this persists", + permanent=False, + ) + webhook.reset_mock() + + await ai_health.report_healthy("extraction") + + # Nothing was ever posted for this episode, so recovery posts nothing either. + webhook.assert_not_awaited() + + +async def test_recovery_then_re_degradation_alerts_again_as_a_new_episode(): + with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: + await ai_health.report_degraded( + "correlation_cheap", "gemini", "gemini-2.0-flash", + "model is unavailable", "update the model ID", permanent=True, + ) + await ai_health.report_healthy("correlation_cheap") + webhook.reset_mock() + + # A second, later episode must alert on its own first occurrence. + await ai_health.report_degraded( + "correlation_cheap", "gemini", "gemini-2.0-flash", + "model is unavailable", "update the model ID", permanent=True, + ) + + webhook.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# snapshot() +# --------------------------------------------------------------------------- + +async def test_snapshot_reports_all_tiers_healthy_by_default(): + state = ai_health.snapshot() + assert set(state.keys()) == set(ai_health.TIERS) + for tier_state in state.values(): + assert tier_state["degraded"] is False + assert tier_state["consecutive_failures"] == 0