server-26/drb-c2-core/tests/test_ai_timeouts.py

"""
server-26#153 — no OpenAI or Gemini request ran with an explicit timeout.

OpenAI's SDK waits 600s per attempt and retries twice (timeouts included);
google-genai waits forever. A hung provider call therefore had no upper bound,
and recorrelation_sweep's MIN_MINUTES_SINCE_PIPELINE_START was a guess against
that tail. These tests pin:
  - every client is built with the configured timeout / retry budget,
  - a hung request really does give up (real SDK, local socket that never answers),
  - a timeout takes the same fallback path as any other provider error,
  - the sweep hold-back is derived from the timeouts, not hand-picked.
"""
import pathlib
import re
import socket
import threading
import time
from unittest.mock import AsyncMock, patch

import httpx
import openai
import pytest

from app.config import settings
from app.internal import ai_clients, ai_health, gemini, llm_correlator, recorrelation_sweep
from app.internal import intelligence, summarizer, transcript_correction

APP_DIR = pathlib.Path(__file__).resolve().parent.parent / "app"


def _timeout_error() -> openai.APITimeoutError:
    return openai.APITimeoutError(request=httpx.Request("POST", "https://api.openai.com/v1/x"))


@pytest.fixture(autouse=True)
def _reset_ai_health():
    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}


@pytest.fixture
def openai_key():
    with patch.object(settings, "openai_api_key", "sk-test"):
        yield


# ── Every client carries the configured limits ──────────────────────────────

def test_openai_client_uses_configured_timeout_and_retries(openai_key):
    with patch.object(settings, "openai_timeout_seconds", 42.0), \
         patch.object(settings, "openai_max_retries", 1):
        client = ai_clients.openai_client()
    assert client.timeout == 42.0
    assert client.max_retries == 1


def test_openai_client_timeout_override_for_stt(openai_key):
    client = ai_clients.openai_client(timeout=settings.stt_timeout_seconds)
    assert client.timeout == settings.stt_timeout_seconds


def test_async_openai_client_uses_configured_limits(openai_key):
    with patch.object(settings, "openai_timeout_seconds", 7.0), \
         patch.object(settings, "openai_max_retries", 0):
        client = ai_clients.async_openai_client()
    assert client.timeout == 7.0
    assert client.max_retries == 0


def test_defaults_are_not_the_sdk_defaults():
    # SDK defaults: OpenAI 600s x 3 attempts, Gemini unbounded.
    assert settings.stt_timeout_seconds < 600
    assert settings.openai_timeout_seconds < 600
    assert settings.openai_max_retries < openai.DEFAULT_MAX_RETRIES
    assert settings.gemini_timeout_seconds > 0


def test_gemini_client_sets_timeout_in_milliseconds():
    saved = gemini._client
    gemini._client = None
    try:
        with patch.object(settings, "gemini_api_key", "test-key"), \
             patch.object(settings, "gemini_timeout_seconds", 45.0):
            client = gemini._get_client()
        assert client._api_client._http_options.timeout == 45_000
    finally:
        gemini._client = saved


def test_no_openai_client_is_built_outside_ai_clients():
    """Regression guard: a new `OpenAI(...)` elsewhere would silently run on
    the 600s SDK default again."""
    offenders = []
    for path in APP_DIR.rglob("*.py"):
        if path.name == "ai_clients.py":
            continue
        for n, line in enumerate(path.read_text().splitlines(), 1):
            if re.search(r"\b(Async)?OpenAI\(", line) or re.search(r"genai\.Client\(", line):
                if path.name == "gemini.py" and "genai.Client(" in line:
                    continue
                offenders.append(f"{path.relative_to(APP_DIR)}:{n}: {line.strip()}")
    assert offenders == []


# ── A hung request actually gives up ────────────────────────────────────────

@pytest.fixture
def hung_server():
    """A TCP server that accepts connections and never sends a byte."""
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.bind(("127.0.0.1", 0))
    srv.listen(8)
    conns = []
    stop = threading.Event()

    def accept():
        srv.settimeout(0.1)
        while not stop.is_set():
            try:
                conns.append(srv.accept()[0])
            except OSError:
                continue

    t = threading.Thread(target=accept, daemon=True)
    t.start()
    yield f"http://127.0.0.1:{srv.getsockname()[1]}/v1"
    stop.set()
    t.join(timeout=2)
    for c in conns:
        c.close()
    srv.close()


def test_hung_openai_request_times_out_instead_of_waiting_600s(openai_key, hung_server, monkeypatch):
    monkeypatch.setenv("OPENAI_BASE_URL", hung_server)
    with patch.object(settings, "openai_timeout_seconds", 0.3), \
         patch.object(settings, "openai_max_retries", 1):
        client = ai_clients.openai_client()
        t0 = time.monotonic()
        with pytest.raises(openai.APITimeoutError):
            client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": "x"}])
        elapsed = time.monotonic() - t0
    # 2 attempts x 0.3s + SDK backoff (<~1s); far below the 600s default.
    assert elapsed < 5


# ── Timeouts degrade exactly like other provider errors ────────────────────

def test_timeout_messages_classify_as_transient():
    assert ai_health.classify(str(_timeout_error())) == "transient"
    assert ai_health.classify(str(httpx.ReadTimeout("The read operation timed out"))) == "transient"


def _corr_ctx():
    return {"call_id": "c1", "is_thin_call": False, "recent": [{"incident_id": "i1"}],
            "all_active": [], "call_units": ["12"], "call_cleared": [], "tags": ["mva"]}


@pytest.mark.asyncio
async def test_gemini_timeout_in_decide_falls_back_to_rules():
    def hang(model, prompt):
        raise httpx.ReadTimeout("The read operation timed out")

    with patch.object(settings, "gemini_api_key", "test-key"), \
         patch.object(llm_correlator, "_build_decide_prompt", lambda ctx: "p"), \
         patch.object(llm_correlator, "_sync_gemini", hang):
        assert await llm_correlator.decide("c1", _corr_ctx()) is None  # None == rules_only
    state = ai_health.snapshot()["correlation_cheap"]
    assert state["consecutive_failures"] == 1
    assert state["permanent"] is False


@pytest.mark.asyncio
async def test_gemini_timeout_in_tiebreak_keeps_the_rules_decision():
    def hang(model, prompt):
        raise httpx.ReadTimeout("The read operation timed out")

    rules = {"action": "new", "matched_incident": None, "corr_debug": {}}
    with patch.object(llm_correlator, "_build_tiebreak_prompt", lambda r, l, c: "p"), \
         patch.object(llm_correlator, "_sync_gemini", hang):
        out = await llm_correlator.tiebreak(rules, {"action": "orphan"}, _corr_ctx())
    assert out is rules
    assert ai_health.snapshot()["correlation_smart"]["permanent"] is False


@pytest.mark.asyncio
async def test_gemini_timeout_in_transcript_correction_keeps_raw_text():
    def hang(model, prompt):
        raise httpx.ReadTimeout("The read operation timed out")

    text = "engine 5 respond to a structure fire at 12 main street smoke showing"
    with patch.object(settings, "gemini_api_key", "test-key"), \
         patch.object(transcript_correction, "resolve_context", AsyncMock(return_value={})), \
         patch.object(transcript_correction, "_sync_gemini", hang):
        assert await transcript_correction.correct("c1", text) == (None, None, False)


class _TimingOutOpenAI:
    """Stands in for the client ai_clients builds; every endpoint times out."""
    def __init__(self, *a, **k):
        self.chat = self.embeddings = self
        self.completions = self

    def create(self, **kwargs):
        raise _timeout_error()


def test_extraction_timeout_propagates_to_the_ai_health_path(openai_key):
    # _sync_extract re-raises API failures so extract_scenes reports them to
    # ai_health (tier "extraction"); a timeout must not be swallowed as "no scenes".
    with patch.object(ai_clients, "openai_client", _TimingOutOpenAI):
        with pytest.raises(openai.APITimeoutError):
            intelligence._sync_extract("unit 12 on scene", "PD Dispatch", 1, "sys", None)


def test_embedding_timeout_returns_none(openai_key):
    with patch.object(ai_clients, "openai_client", _TimingOutOpenAI):
        assert intelligence._sync_embed("unit 12 on scene") is None


def test_summary_timeout_returns_none(openai_key):
    with patch.object(ai_clients, "openai_client", _TimingOutOpenAI):
        assert summarizer._sync_summarize({"type": "fire"}, ["engine 5 on scene"]) is None


# ── The sweep hold-back is derived from the timeouts ────────────────────────

def test_worst_case_bound_follows_the_settings():
    with patch.object(settings, "stt_timeout_seconds", 100.0), \
         patch.object(settings, "openai_timeout_seconds", 10.0), \
         patch.object(settings, "openai_max_retries", 1), \
         patch.object(settings, "gemini_timeout_seconds", 20.0), \
         patch.object(settings, "stt_retry_on_degenerate", True):
        bound = recorrelation_sweep.pipeline_worst_case_seconds(n_scenes=2)
    # stt 2x2x100 + correction 20 + extraction 2x10 + embeds 2x2x10 + decide/tiebreak 2x20 + slack
    assert bound == 400 + 20 + 20 + 40 + 40 + recorrelation_sweep._NON_AI_SLACK_SECONDS


def test_hold_back_covers_the_worst_case_and_never_drops_below_old_floor():
    bound_min = recorrelation_sweep.pipeline_worst_case_seconds() / 60
    assert recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START >= bound_min
    assert recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START >= 15
    # ...and stays inside the scan window, or the sweep could never recover a
    # call whose pipeline died after writing intelligence_started_at.
    assert recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START < settings.recorrelation_scan_minutes