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

"""
server-26#297 — config-driven STT provider with OpenAI as the fallback.

The default must stay OpenAI until a Groq key exists (so prod does not break
before the key is provisioned), a failing primary must fall back to whisper-1
without losing the call, and the vocabulary prompt must never carry a
ten-code series (the hallucination documented in transcription.py).
"""
from unittest.mock import AsyncMock, patch

import pytest

from app.internal import transcription as tr
from app.config import settings


def _settings(**over):
    base = dict(stt_provider="auto", groq_api_key=None, openai_api_key="sk-openai",
                stt_api_key=None, stt_base_url=None, stt_fallback_to_openai=True,
                stt_model="whisper-1", stt_compat_model="whisper-large-v3-turbo")
    base.update(over)
    return [patch.object(settings, k, v) for k, v in base.items()]


class _patched:
    def __init__(self, **over):
        self.ps = _settings(**over)

    def __enter__(self):
        for p in self.ps:
            p.start()

    def __exit__(self, *a):
        for p in self.ps:
            p.stop()


# ── provider resolution ─────────────────────────────────────────────────────

def test_default_without_groq_key_is_openai_only():
    with _patched():
        targets = tr.stt_targets()
    assert [(t.provider, t.model, t.tier) for t in targets] == [("openai", "whisper-1", "transcription")]
    assert targets[0].vocab_prompt is False, "the OpenAI path keeps today's fixed prompt"


def test_groq_key_makes_groq_primary_with_openai_fallback():
    with _patched(groq_api_key="gsk"):
        targets = tr.stt_targets()
    assert [(t.provider, t.model) for t in targets] == [
        ("groq", "whisper-large-v3-turbo"), ("openai", "whisper-1")]
    assert targets[0].base_url == "https://api.groq.com/openai/v1"
    assert targets[0].api_key == "gsk"
    assert targets[1].tier == "transcription_fallback"


def test_fallback_can_be_switched_off():
    with _patched(groq_api_key="gsk", stt_fallback_to_openai=False):
        assert [t.provider for t in tr.stt_targets()] == ["groq"]


def test_no_openai_key_means_no_fallback():
    with _patched(groq_api_key="gsk", openai_api_key=None):
        assert [t.provider for t in tr.stt_targets()] == ["groq"]


def test_compat_server_from_base_url():
    with _patched(stt_base_url="http://whisper.local/v1", stt_compat_model="large-v3"):
        targets = tr.stt_targets()
    assert targets[0].provider == "compat" and targets[0].base_url == "http://whisper.local/v1"
    assert targets[0].model == "large-v3"


def test_explicit_groq_without_key_degrades_to_openai():
    with _patched(stt_provider="groq"):
        assert [t.provider for t in tr.stt_targets()] == ["openai"]


def test_nothing_configured_is_empty():
    with _patched(openai_api_key=None):
        assert tr.stt_targets() == []


# ── fallback in transcribe_call ─────────────────────────────────────────────

def _ok(text="Engine 5 responding to a structure fire on Main Street"):
    return {"text": text, "segments": [{"start": 0.0, "end": 3.0, "text": text}],
            "degenerate": False, "duration": 3.0, "confidence": 0.81}


async def _transcribe(sync_impl, **over):
    saved: dict = {}

    async def _doc_set(coll, doc_id, data, merge=True):
        saved.update(data)

    flags = {"transcript_correction_enabled": False}
    with _patched(**over), \
         patch.object(tr, "_download", lambda uri: "/nonexistent.flac"), \
         patch.object(tr, "_sync_transcribe_file", sync_impl), \
         patch.object(tr.transcript_correction, "resolve_context", AsyncMock(return_value={})), \
         patch.object(tr.fstore, "doc_set", _doc_set), \
         patch.object(tr, "ai_health") as health, \
         patch("app.internal.feature_flags.resolve_flags",
               AsyncMock(return_value=(flags, lambda k: flags.get(k, True)))):
        health.report_healthy = AsyncMock()
        health.report_degraded = AsyncMock()
        health.classify = lambda text: "transient"
        result = await tr.transcribe_call("call-1", "gs://b/call-1.flac", "PD Dispatch", system_id="sys-1")
    return result, saved, health


@pytest.mark.asyncio
async def test_primary_failure_falls_back_to_openai():
    seen = []

    def sync(path, target, prompt, glossary=None):
        seen.append(target.provider)
        if target.provider == "groq":
            raise RuntimeError("503 Service Unavailable")
        return _ok()

    (text, segments), saved, health = await _transcribe(sync, groq_api_key="gsk")
    assert seen == ["groq", "openai"]
    assert text.startswith("Engine 5")
    assert saved["stt_provider"] == "openai" and saved["stt_model"] == "whisper-1"
    # The primary stays red; only the fallback tier is marked healthy.
    assert health.report_degraded.await_args.args[:3] == ("transcription", "groq", "whisper-large-v3-turbo")
    health.report_healthy.assert_awaited_once_with("transcription_fallback")


@pytest.mark.asyncio
async def test_primary_success_records_provider_and_cost_inputs():
    (text, _), saved, health = await _transcribe(lambda *a, **k: _ok(), groq_api_key="gsk")
    assert saved["stt_provider"] == "groq"
    assert saved["stt_model"] == "whisper-large-v3-turbo"
    assert saved["stt_audio_seconds"] == 3.0
    assert saved["stt_confidence"] == 0.81
    health.report_healthy.assert_awaited_once_with("transcription")


@pytest.mark.asyncio
async def test_every_target_failing_returns_nothing():
    def sync(*a, **k):
        raise RuntimeError("boom")

    (text, segments), saved, health = await _transcribe(sync, groq_api_key="gsk")
    assert (text, segments) == (None, [])
    assert health.report_degraded.await_count == 2
    health.report_healthy.assert_not_awaited()


@pytest.mark.asyncio
async def test_default_config_never_calls_groq():
    seen = []

    def sync(path, target, prompt, glossary=None):
        seen.append((target.provider, target.base_url))
        return _ok()

    await _transcribe(sync)
    assert seen == [("openai", None)]


# ── prompt ──────────────────────────────────────────────────────────────────

CONTEXT = {
    "area": {"municipality": "Ossining", "county": "Westchester", "state": "New York",
             "local_knowledge": [{"term": "Snowden Avenue"}, {"term": "Sing Sing", "meaning": "prison"}]},
    "vocabulary": ["10-13", "5-David", "Croton Point", "Snowden Avenue"],
    "ten_codes": {"10-4": "ack", "10-13": "officer needs assistance"},
}


def test_no_context_is_the_legacy_prompt():
    prompt, glossary = tr.build_stt_prompt("PD Dispatch", None)
    assert prompt == "Talkgroup: PD Dispatch. " + tr._WHISPER_PROMPT
    assert glossary == []


def test_vocabulary_prompt_carries_places_but_never_ten_codes():
    prompt, glossary = tr.build_stt_prompt("PD Dispatch", CONTEXT)
    assert "Ossining, Westchester, New York" in prompt
    assert glossary == ["Snowden Avenue", "Sing Sing", "5-David", "Croton Point"]
    assert "10-13" not in prompt, "a code-shaped vocabulary term must stay out"
    assert "officer needs assistance" not in prompt
    assert prompt.endswith(tr._WHISPER_PROMPT)


def test_prompt_stays_inside_the_token_budget():
    big = {"area": {}, "vocabulary": [f"Placename{i} Boulevard" for i in range(200)]}
    with patch.object(settings, "stt_prompt_max_tokens", 200):
        prompt, glossary = tr.build_stt_prompt("PD Dispatch", big)
    assert tr._estimate_tokens(prompt) <= 200
    assert 0 < len(glossary) < 200


def test_token_estimate_is_conservative_for_radio_text():
    # GPT-2 tokenises this at ~25 tokens; the estimate must not be lower.
    assert tr._estimate_tokens("Engine 5 responding to a structure fire on Main Street, 10-4.") >= 20


# ── degenerate output the vocabulary prompt could add ───────────────────────

def test_prompt_echo_is_discarded():
    glossary = ["Snowden Avenue", "Sing Sing", "Croton Point", "Main Street"]
    assert tr._is_degenerate("Snowden Avenue, Sing Sing, Croton Point, Main Street.", [], glossary)
    assert tr._is_degenerate("Local names: Snowden Avenue, Sing Sing.", [], glossary)


def test_one_real_street_in_a_sentence_is_not_an_echo():
    glossary = ["Snowden Avenue", "Sing Sing", "Croton Point"]
    real = "Car 4 respond to Snowden Avenue for a report of a disabled vehicle."
    assert not tr._is_degenerate(real, [], glossary)


def test_compression_ratio_catches_an_alternating_loop():
    loop = "Dispatch, copy. Unit 4, go ahead. " * 12
    assert not tr._phrase_loop(loop), "the old checks miss this shape"
    assert tr._is_degenerate(loop, [])


def test_real_long_traffic_passes_the_compression_check():
    real = (
        "Headquarters to 5 David, respond to 12 Snowden Avenue for a report of a "
        "two-car accident with injuries, caller states one vehicle is overturned. "
        "5 David copies, en route from Route 9. Be advised EMS has been notified "
        "and Ossining Fire is responding as well. 5 David on scene, two vehicles, "
        "one patient complaining of neck pain, requesting a tow for one."
    )
    assert len(real) >= tr._MIN_CHARS_FOR_COMPRESSION
    assert not tr._is_degenerate(real, [])


def test_sweep_hold_back_counts_the_fallback_attempt():
    from app.internal import recorrelation_sweep
    with _patched():
        single = recorrelation_sweep.pipeline_worst_case_seconds()
    with _patched(groq_api_key="gsk"):
        double = recorrelation_sweep.pipeline_worst_case_seconds()
    stt_one = (2 if settings.stt_retry_on_degenerate else 1) * (1 + settings.openai_max_retries) \
        * settings.stt_timeout_seconds
    assert double - single == stt_one
    assert double / 60 < settings.recorrelation_scan_minutes, "the sweep must still be able to recover the call"


def test_location_readback_is_not_an_echo():
    """Review of #297: intersection/address readbacks are where pins come from."""
    glossary = ["Route 9", "Croton Avenue", "North Highland Place", "Main Street", "Ossining"]
    for real in ("Route 9 and Croton Avenue, North Highland Place.",
                 "Main Street, Croton Avenue, Route 9, Ossining."):
        assert not tr._is_degenerate(real, [], glossary), real


def test_long_busy_call_is_judged_per_window():
    seg_text = "Engine 5 responding, Ladder 2 responding, Medic 1 en route to Main Street. "
    segments = [{"start": float(i * 5), "end": float(i * 5 + 4), "text": seg_text} for i in range(60)]
    text = " ".join(s["text"] for s in segments)
    # One repeated real update for 5 minutes still loops within every window,
    # so this is degenerate — but a varied call with one looping window out of three is not.
    assert tr._is_degenerate(text, segments)
    real = [
        "Headquarters to 5 David, respond to 12 Snowden Avenue for a two-car accident.",
        "5 David copies, en route from Route 9, ETA four minutes.",
        "Be advised EMS has been notified and Ossining Fire is responding as well.",
        "5 David on scene, two vehicles, one patient complaining of neck pain.",
        "Requesting a tow for the blue sedan, it is blocking the northbound lane.",
        "Copy, tow is being contacted, any further units needed at this time?",
        "Negative, traffic is backing up toward Croton Point, send one for control.",
        "6 Henry, can you clear the larceny and assist 5 David on Snowden?",
        "6 Henry clearing, show me en route, coming down from North Highland.",
        "Medic 1 on location, patient is being packaged for transport to Phelps.",
        "Fire is standing by for fluids, no entrapment, both drivers out of the cars.",
        "Received, all units be advised the roadway is now closed southbound.",
    ]
    loop = "Dispatch, copy. Unit 4, go ahead. " * 12
    varied = [{"start": float(i * 5), "end": float(i * 5 + 4), "text": t} for i, t in enumerate(real)]
    varied.append({"start": 70.0, "end": 98.0, "text": loop})  # its own window
    assert not tr._is_degenerate(" ".join(s["text"] for s in varied), varied)