server-26/drb-c2-core/app/internal/ai_health.py

"""
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 contextvars import ContextVar
from datetime import datetime, timezone
from typing import Optional

from app.internal.logger import logger
from app.config import settings

TIERS = ("transcription", "correlation_cheap", "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}

# Set by a replay run to a list it owns; report_degraded appends there instead
# of touching _state while inside a sandbox (see app/internal/replay.py).
_sandbox_failures: ContextVar[Optional[list]] = ContextVar("drb_ai_sandbox_failures", default=None)


def collect_sandbox_failures(sink: Optional[list]):
    return _sandbox_failures.set(sink)


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.
    """
    from app.internal import firestore as fstore
    if fstore.in_sandbox():
        # A replay's rate limits are not a live outage, and must never page
        # the AI-alert webhook or flip /health/ai (app/internal/replay.py).
        # They are the run's own problem, so they go to the run instead.
        sink = _sandbox_failures.get()
        if sink is not None:
            sink.append({"tier": tier, "provider": provider, "model": model,
                         "problem": problem, "permanent": permanent})
        return
    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.
    """
    from app.internal import firestore as fstore
    if fstore.in_sandbox():
        return  # nor may a replay's success "recover" a real live outage
    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}")