Logan Cusano
a250c29e3c
Three AI dependency failures in one night (retired Gemini model IDs, depleted Gemini balance, unpayable OpenAI account) each surfaced only as a single ERROR log line that nobody was watching. Add app/internal/ai_health.py, a shared in-memory registry that transcription.py and llm_correlator.py report into on every call (success and failure), distinguishing permanent conditions (dead model, dead billing) which alert immediately from transient ones (rate limits, network blips) which only alert after they persist. Alerts POST once per degradation episode and once on recovery to an optional Discord webhook (AI_ALERT_WEBHOOK_URL), reusing alerter.py's httpx pattern. State is exposed unauthenticated at GET /health/ai alongside the existing /health. Closes logan/server-26#14
202 lines
8.2 KiB
Python
202 lines
8.2 KiB
Python
"""
|
|
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
|