Transcription: config-driven OpenAI-compatible STT with OpenAI fallback (server-26#297, server-26#83)

STT_PROVIDER=auto picks Groq whisper-large-v3-turbo only when GROQ_API_KEY is
set (else OpenAI, unchanged); compat covers any OpenAI-compatible server via
STT_BASE_URL/STT_API_KEY. A failed primary falls back to OpenAI STT_MODEL on
the same downloaded audio; the fallback reports to its own ai_health tier so
a dead primary stays red. The non-OpenAI provider gets a token-budgeted
glossary prompt (area places, local_knowledge, vocabulary; never ten-codes),
guarded by a narrow prompt-echo check. A per-30s-window zlib compression
check (Whisper's 2.4) adds a noise gate that survives correction being off
(#83, partial). Call docs record stt_provider/model/audio_seconds/confidence.
The sweep hold-back counts the fallback attempt. gate_chatter added to
chatter_classifier (used by transcription to skip correction when
CHATTER_GATE_ENABLED). GROQ_API_KEY templated with default ''.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

drb-c2-core/app/config.py

@@ -28,8 +28,39 @@ class Settings(BaseSettings):
 
     # OpenAI (STT + intelligence)
     openai_api_key: Optional[str] = None
+    # The OpenAI transcription model. Used when OpenAI is the STT provider AND
+    # as the fallback when another provider fails (server-26#297).
     stt_model: str = "whisper-1"  # whisper-1 | gpt-4o-mini-transcribe | gpt-4o-transcribe
 
+    # ── STT provider (server-26#297) ──────────────────────────────────────────
+    # Any OpenAI-compatible /audio/transcriptions endpoint. Groq serves
+    # whisper-large-v3-turbo at $0.04/audio-hour against whisper-1's $0.36, same
+    # API (min billed length 10s per request, prompt capped at 224 tokens --
+    # console.groq.com/docs/speech-to-text, checked 2026-10-27).
+    #   auto   -- groq when GROQ_API_KEY is set, compat when STT_BASE_URL is
+    #             set, else openai. The default, so a deploy without either
+    #             behaves exactly as before.
+    #   openai -- OpenAI only (stt_model).
+    #   groq   -- Groq (stt_compat_model), OpenAI fallback.
+    #   compat -- any other OpenAI-compatible server: stt_base_url + stt_api_key
+    #             + stt_compat_model (e.g. a self-hosted faster-whisper, #296).
+    stt_provider: str = "auto"
+    stt_base_url: Optional[str] = None   # None -> the provider's own default
+    stt_api_key: Optional[str] = None    # None -> GROQ_API_KEY / OPENAI_API_KEY
+    groq_api_key: Optional[str] = None
+    stt_compat_model: str = "whisper-large-v3-turbo"  # model on a non-OpenAI provider
+    # Retry a failed non-OpenAI request once on OpenAI stt_model. Needs
+    # OPENAI_API_KEY; without it a primary failure just means no transcript.
+    stt_fallback_to_openai: bool = True
+    # Domain vocabulary in the STT prompt (area places, local_knowledge terms,
+    # vocabulary). Applied to the non-OpenAI provider only: whisper-1 echoed
+    # vocabulary into silence before (see transcription.py), so the OpenAI
+    # path -- including the fallback -- keeps today's fixed prompt.
+    stt_vocab_prompt_enabled: bool = True
+    # Whisper keeps only the LAST 224 prompt tokens and silently drops the
+    # rest. The prompt is built against this estimated budget, with headroom.
+    stt_prompt_max_tokens: int = 200
+
     # AI request timeouts (server-26#153). Without these the OpenAI SDK waits
     # up to 600s per attempt and retries twice, and the google-genai client
     # waits forever, so a hung provider call had no upper bound at all — and

drb-c2-core/app/internal/ai_clients.py

@@ -34,6 +34,24 @@ def openai_client(timeout: Optional[float] = None):
     )
 
 
+def stt_client(api_key: Optional[str], base_url: Optional[str] = None):
+    """
+    Synchronous client for an OpenAI-compatible transcription endpoint
+    (server-26#297) -- OpenAI itself, Groq, or a self-hosted server. Same
+    timeout/retry limits as every other client here, with STT's own timeout.
+    """
+    from openai import OpenAI
+    kwargs = {}
+    if base_url:
+        kwargs["base_url"] = base_url
+    return OpenAI(
+        api_key=api_key,
+        timeout=settings.stt_timeout_seconds,
+        max_retries=settings.openai_max_retries,
+        **kwargs,
+    )
+
+
 def async_openai_client(timeout: Optional[float] = None):
     """Async OpenAI client, same limits as openai_client()."""
     from openai import AsyncOpenAI

drb-c2-core/app/internal/chatter_classifier.py

@@ -14,13 +14,17 @@ This module is that classifier. It is a PURE function of the transcript text —
 no Firestore, no LLM call, no side effects — so it is cheap to run on every
 transcript and cheap to test against real dumps offline.
 
-SHADOW MODE ONLY. As of this module's introduction, nothing skips scene
+LIVE AS A COST GATE SINCE server-26#298, through `gate_chatter`, not
-extraction based on this verdict. `intelligence.extract_scenes` calls
+`classify_chatter`. `classify_chatter` is still the shadow verdict recorded on
-`classify_chatter` purely to record the verdict on the call doc
+every call doc (`chatter_classifier_verdict` / `chatter_classifier_reason`),
-(`chatter_classifier_verdict` / `chatter_classifier_reason`) so it becomes
+unchanged, so the `/admin` tallies keep measuring the same thing. The gate is
-observable in the next `/admin` correlation-debug dump, exactly like
+stricter on the one shape the #127 review flagged as a live false-positive
-`corr_gate_veto` (server-26#115 / PR #126). See the TODO at that call site for
+risk: `roll_call` matches "roll call" anywhere, so "Delayed for roll call, then
-what has to be true before this flips live.
+respond to a structure fire at 12 Main Street" is a roll_call verdict. The gate
+only skips a roll call whose remaining words are callsigns and roll-call
+phrasing. Gated transcripts skip transcript correction and scene extraction
+(and with it embedding and geocoding); the free rule paths (clearance parse,
+self-initiated/plate backstop) still run.
 
 Precision over recall, deliberately. A false positive here — flagging a REAL
 event as chatter — would, once live, silently mean that event never gets a
@@ -133,3 +137,36 @@ def classify_chatter(transcript: Optional[str]) -> tuple[bool, Optional[str]]:
         return False, None
 
     return True, "bare_acknowledgement"
+
+
+# Words a roll call is made of, beyond the filler/designator/phonetic sets.
+_ROLL_CALL_WORDS = frozenset({
+    "roll", "call", "all", "cars", "car", "patrol", "patrols", "troop", "staff",
+    "concludes", "time", "came", "because", "volunteer", "nyc", "kl", "henry",
+    "here", "present", "s", "m", "it",
+})
+# A roll call may name a place or two ("Post 4, Ossining"); more than this many
+# leftover words means something else is being said in the same transmission.
+_ROLL_CALL_MAX_LEFTOVER = 2
+
+
+def gate_chatter(transcript: Optional[str]) -> tuple[bool, Optional[str]]:
+    """
+    Whether to skip paid AI for this transcript (server-26#298).
+
+    `classify_chatter`'s verdict, with roll_call narrowed: the transcript must
+    be roll-call content end to end, not a roll call mentioned on the way to
+    something real. Precision over recall still — a false positive here means
+    a real event never gets a scene.
+    """
+    is_chatter, reason = classify_chatter(transcript)
+    if not is_chatter:
+        return False, None
+    if reason == "roll_call":
+        leftover = [
+            t for t in _tokenize(transcript or "")
+            if not _is_filler_token(t) and t not in _ROLL_CALL_WORDS
+        ]
+        if len(leftover) > _ROLL_CALL_MAX_LEFTOVER:
+            return False, None
+    return True, reason

drb-c2-core/app/internal/recorrelation_sweep.py

@@ -65,7 +65,11 @@ def pipeline_worst_case_seconds(n_scenes: int = _BOUND_SCENES) -> float:
     _run_intelligence_pipeline."""
     openai_attempts = 1 + max(0, settings.openai_max_retries)
     stt_rounds = 2 if settings.stt_retry_on_degenerate else 1
-    stt = stt_rounds * openai_attempts * settings.stt_timeout_seconds
+    # A failed primary STT provider falls back to OpenAI (server-26#297), so
+    # the worst case is every configured target timing out in turn.
+    from app.internal.transcription import stt_targets
+    stt_providers = max(1, len(stt_targets()))
+    stt = stt_providers * stt_rounds * openai_attempts * settings.stt_timeout_seconds
     # Transcript correction (inside transcribe_call). gemini.generate_json only
     # re-issues a request when the model rejects a thinking level — a fast
     # 400, not a timeout — and google-genai itself does not retry.

drb-c2-core/app/internal/replay.py

@@ -73,6 +73,9 @@ _DERIVED = {
     "vehicles", "cleared_units", "severity", "incident_type", "type",
     "embedding", "skip_reason", "intelligence_started_at", "reassignment",
     "resolved", "has_updates", "audio_url",
+    # server-26#297: which STT produced the transcript, and its confidence —
+    # a re-transcription that comes back empty must not keep the live values.
+    "stt_provider", "stt_model", "stt_audio_seconds", "stt_confidence",
 }
 _DERIVED_PREFIXES = ("corr_", "chatter_classifier_", "eval_")
 _TRANSCRIPT_FIELDS = (
@@ -406,6 +409,7 @@ async def _prepare(call: dict, mode: str, source_scenes: dict[str, dict]) -> dic
             transcript, segments = await transcription.transcribe_call(
                 call_id, call["audio_gcs_uri"], talkgroup_name,
                 system_id=call.get("system_id"), talkgroup_id=call.get("talkgroup_id"),
+                node_id=call.get("node_id"),
             )
     else:
         transcript, segments = _stored_input(call)

drb-c2-core/app/internal/transcription.py

@@ -1,18 +1,25 @@
 """
-Speech-to-text transcription for recorded calls using OpenAI Whisper.
+Speech-to-text transcription for recorded calls.
 
-Audio is downloaded from GCS then sent to the Whisper API. Falls back to
+Audio is downloaded from GCS once, then sent to an OpenAI-compatible
-returning None on any failure so the intelligence pipeline can still run.
+/audio/transcriptions endpoint: OpenAI whisper-1 by default, or a cheaper
+provider (Groq whisper-large-v3-turbo, a self-hosted server) when configured,
+with OpenAI as the fallback when that provider fails (server-26#297). Returns
+None on any failure so the intelligence pipeline can still run.
 """
 import asyncio
+import math
 import re
 import tempfile
 import os
+from dataclasses import dataclass
 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.internal import transcript_correction
+from app.internal import area_context
+from app.internal.chatter_classifier import gate_chatter
 from app.config import settings
 
 # Whisper treats `prompt` as preceding transcript text, not instructions.
@@ -43,6 +50,19 @@ _RUN_RATIO = 0.7             # share of consecutive pairs that must step by +1
 _MIN_SEGMENTS_FOR_REPEAT = 6  # segments needed before repetition is considered
 _UNIQUE_RATIO = 0.25         # unique/total segment texts at or below this is degenerate
 _MAX_PHRASE_REPEATS = 8      # identical consecutive phrase repeats allowed in one blob
+# Whisper's own failure test (openai/whisper transcribe.py, compression_ratio_
+# threshold=2.4): text that gzips this well is a loop, whatever its shape. The
+# two checks above only know two loop shapes; this catches the rest (two
+# phrases alternating, a list re-emitted) without a model call, so a noise
+# gate survives with transcript correction switched off (server-26#83). Only
+# on longer text -- a short string has too little to compress to judge.
+_MAX_COMPRESSION_RATIO = 2.4
+_MIN_CHARS_FOR_COMPRESSION = 200
+# Prompt echo (server-26#297): on silence Whisper can emit its own prompt.
+# With a vocabulary glossary in the prompt that reads as a list of local
+# street names. See _prompt_echo for why this is kept narrow.
+_ECHO_MIN_TERMS = 3
+_ECHO_WORD_RATIO = 0.6
 
 
 def _ten_code_run(text: str) -> bool:
@@ -77,7 +97,74 @@ def _phrase_loop(text: str) -> bool:
     return False
 
 
-def _is_degenerate(text: str, segments: list[dict]) -> bool:
+def _compression_ratio(text: str) -> float:
+    import zlib
+    raw = text.encode("utf-8")
+    return len(raw) / max(1, len(zlib.compress(raw)))
+
+
+def _prompt_echo(text: str, glossary: list[str]) -> bool:
+    """
+    True if the transcript is the prompt's own glossary read back.
+
+    Deliberately narrow (review of server-26#297): a real location readback
+    ("Main Street, Croton Avenue, Route 9") names glossary terms too, and
+    that is where map pins come from. So either the lead-in itself came back
+    ("Local names: ..."), or nearly the whole glossary did, in prompt order,
+    with little else around it.
+    """
+    if not text or not glossary:
+        return False
+    low = text.lower()
+    if "local names" in low:
+        return True
+    if len(glossary) < _ECHO_MIN_TERMS:
+        return False
+    positions = []
+    for t in glossary:
+        m = re.search(rf"\b{re.escape(t.lower())}\b", low)
+        if m:
+            positions.append(m.start())
+    if len(positions) < max(_ECHO_MIN_TERMS, math.ceil(0.8 * len(glossary))):
+        return False
+    if positions != sorted(positions):
+        return False
+    words = re.findall(r"[\w'-]+", low)
+    term_words = {w for t in glossary for w in re.findall(r"[\w'-]+", t.lower())}
+    return bool(words) and sum(1 for w in words if w in term_words) / len(words) >= _ECHO_WORD_RATIO
+
+
+def _compression_loop(text: str, segments: list[dict]) -> bool:
+    """
+    Whisper applies its 2.4 threshold per 30s decode window, and zlib's ratio
+    grows with length, so judging a whole 5-minute call at once would drop
+    busy real traffic. Judge ~30s windows of segments (or ~500-char chunks of
+    a segment-less transcript) and call it a loop only when at least half of
+    the windows long enough to judge are loops.
+    """
+    windows: list[str] = []
+    if segments:
+        cur, start = [], None
+        for s in segments:
+            st = float(s.get("start") or 0.0)
+            if start is None:
+                start = st
+            if st - start > 30.0 and cur:
+                windows.append(" ".join(cur))
+                cur, start = [], st
+            cur.append(s.get("text") or "")
+        if cur:
+            windows.append(" ".join(cur))
+    else:
+        windows = [text[i:i + 500] for i in range(0, len(text), 500)]
+    judged = [w for w in windows if len(w) >= _MIN_CHARS_FOR_COMPRESSION]
+    if not judged:
+        return False
+    loops = sum(1 for w in judged if _compression_ratio(w) > _MAX_COMPRESSION_RATIO)
+    return loops * 2 >= len(judged)
+
+
+def _is_degenerate(text: str, segments: list[dict], glossary: Optional[list[str]] = None) -> bool:
     """True if a transcript looks like Whisper output rather than radio traffic.
 
     Applied AFTER the per-segment no_speech_prob filter, which does not catch
@@ -88,6 +175,10 @@ def _is_degenerate(text: str, segments: list[dict]) -> bool:
         return False
     if _ten_code_run(text) or _phrase_loop(text):
         return True
+    if _compression_loop(text, segments):
+        return True
+    if _prompt_echo(text, glossary or []):
+        return True
     # Near-identical segments repeated across the whole recording.
     if len(segments) >= _MIN_SEGMENTS_FOR_REPEAT:
         normalised = {s["text"].strip().lower() for s in segments}
@@ -96,7 +187,155 @@ def _is_degenerate(text: str, segments: list[dict]) -> bool:
     return False
 
 
-async def _log_transcribe_failure(call_id: str, exc: Exception) -> None:
+
+
+# ── Providers (server-26#297) ───────────────────────────────────────────────
+
+_GROQ_BASE_URL = "https://api.groq.com/openai/v1"
+
+
+@dataclass(frozen=True)
+class SttTarget:
+    """One OpenAI-compatible transcription endpoint to try."""
+    provider: str            # "openai" | "groq" | "compat"
+    model: str
+    api_key: Optional[str]
+    base_url: Optional[str]  # None -> the OpenAI SDK default
+    vocab_prompt: bool       # whether the domain-vocabulary prompt is sent
+    tier: str                # ai_health tier this target reports to
+
+
+def stt_targets() -> list[SttTarget]:
+    """
+    The endpoints to try, in order: the configured provider, then OpenAI as
+    fallback. Empty when nothing usable is configured.
+
+    `stt_provider="auto"` resolves to Groq only when GROQ_API_KEY is set, so a
+    deploy that has not provisioned the key yet keeps transcribing on OpenAI
+    exactly as before.
+    """
+    provider = (settings.stt_provider or "auto").strip().lower()
+    if provider == "auto":
+        if settings.groq_api_key:
+            provider = "groq"
+        elif settings.stt_base_url:
+            provider = "compat"  # an explicit endpoint is a compat server
+        else:
+            provider = "openai"
+
+    openai_target = SttTarget(
+        provider="openai", model=settings.stt_model, api_key=settings.openai_api_key,
+        base_url=None, vocab_prompt=False, tier="transcription",
+    )
+    if provider not in ("groq", "compat"):
+        if provider != "openai":
+            logger.warning(f"Unknown STT_PROVIDER {settings.stt_provider!r} — using OpenAI")
+        return [openai_target] if settings.openai_api_key else []
+
+    if provider == "groq":
+        key = settings.stt_api_key or settings.groq_api_key
+        base = settings.stt_base_url or _GROQ_BASE_URL
+    else:
+        # A self-hosted server may not check the key, but the SDK insists on one.
+        key = settings.stt_api_key or "unused"
+        base = settings.stt_base_url
+
+    targets: list[SttTarget] = []
+    if key and base:
+        targets.append(SttTarget(
+            provider=provider, model=settings.stt_compat_model, api_key=key, base_url=base,
+            vocab_prompt=settings.stt_vocab_prompt_enabled, tier="transcription",
+        ))
+    else:
+        logger.warning(f"STT provider {provider!r} has no API key/base URL — using OpenAI only")
+    if settings.openai_api_key and (not targets or settings.stt_fallback_to_openai):
+        tier = "transcription_fallback" if targets else "transcription"
+        targets.append(SttTarget(**{**openai_target.__dict__, "tier": tier}))
+    return targets
+
+
+# ── Prompt (server-26#297) ──────────────────────────────────────────────────
+
+_CODE_SHAPED = re.compile(r"^\s*\d{1,3}(?:-\d{1,3})+\s*$")
+_MAX_TERM_CHARS = 40
+
+
+def _estimate_tokens(text: str) -> int:
+    """
+    Conservative token estimate for Whisper's GPT-2 tokenizer, without the
+    tokenizer: one per punctuation mark and per digit, ~4 letters per token
+    for words. Over-counts real English slightly, which is the safe side —
+    Whisper drops prompt tokens past 224 from the FRONT, silently.
+    """
+    n = 0
+    for piece in re.findall(r"[A-Za-z]+|\d|[^\sA-Za-z\d]", text or ""):
+        n += max(1, math.ceil(len(piece) / 4)) if piece[0].isalpha() else 1
+    return n
+
+
+def _glossary_terms(context: Optional[dict]) -> list[str]:
+    """
+    Local names worth spelling for the recogniser, most specific first:
+    the talkgroup/system local_knowledge terms, then vocabulary.
+
+    Never ten-codes. An enumerated run of codes in the prompt is what made
+    Whisper count "10-4. 10-5. 10-6." over silence (see _WHISPER_PROMPT), so
+    the ten_codes reference and any code-shaped vocabulary term stay out.
+    """
+    if not context:
+        return []
+    area = context.get("area") or {}
+    terms = [e.get("term") or "" for e in (area.get("local_knowledge") or []) if isinstance(e, dict)]
+    terms += [t for t in (context.get("vocabulary") or []) if isinstance(t, str)]
+    out: list[str] = []
+    seen: set[str] = set()
+    for t in terms:
+        t = t.strip()
+        key = t.lower()
+        if not t or key in seen or len(t) > _MAX_TERM_CHARS or _CODE_SHAPED.match(t):
+            continue
+        seen.add(key)
+        out.append(t)
+    return out
+
+
+def build_stt_prompt(
+    talkgroup_name: Optional[str], context: Optional[dict] = None,
+) -> tuple[str, list[str]]:
+    """
+    The STT prompt, and the glossary terms that made it in.
+
+    Without `context` this is exactly the prompt OpenAI has always been sent.
+    With it, the channel's area and local names go in front of the fixed
+    radio-speech priming, cut to settings.stt_prompt_max_tokens. A glossary
+    is how Whisper's prompting guide says to fix the spelling of proper
+    nouns; the priming text stays last because it is the part that sets style.
+    """
+    tg_prefix = f"Talkgroup: {talkgroup_name}. " if talkgroup_name else ""
+    if not context:
+        return tg_prefix + _WHISPER_PROMPT, []
+
+    area = context.get("area") or {}
+    place = ", ".join(str(area[f]) for f in area_context.PLACE_FIELDS if area.get(f))
+    head = tg_prefix + (f"{place}. " if place else "")
+    budget = settings.stt_prompt_max_tokens - _estimate_tokens(head + _WHISPER_PROMPT)
+
+    chosen: list[str] = []
+    for term in _glossary_terms(context):
+        trial = "Local names: " + ", ".join(chosen + [term]) + ". "
+        if _estimate_tokens(trial) > budget:
+            continue  # a shorter later term may still fit
+        chosen.append(term)
+    glossary = ("Local names: " + ", ".join(chosen) + ". ") if chosen else ""
+    prompt = head + glossary + _WHISPER_PROMPT
+    if _estimate_tokens(prompt) > settings.stt_prompt_max_tokens:
+        return tg_prefix + _WHISPER_PROMPT, []
+    return prompt, chosen
+
+
+# ── Failure reporting ───────────────────────────────────────────────────────
+
+async def _log_transcribe_failure(call_id: str, exc: Exception, target: SttTarget) -> None:
     """
     Log a transcription failure, escalating a permanent condition to ERROR
     once (via app.internal.ai_health, which also drives the /health/ai
@@ -115,47 +354,71 @@ async def _log_transcribe_failure(call_id: str, exc: Exception) -> None:
     """
     text = str(exc)
     kind = ai_health.classify(text)
+    who = f"{target.provider} ({target.model})"
+    billing_url = {
+        "openai": "https://platform.openai.com/settings/organization/billing",
+        "groq": "https://console.groq.com/settings/billing",
+    }.get(target.provider, "the provider's billing page")
 
     if kind == "billing":
-        problem = "the OpenAI account cannot be billed"
+        problem = f"the {target.provider} account cannot be billed"
-        fix = "top up at https://platform.openai.com/settings/organization/billing"
+        fix = f"top up at {billing_url}"
         logger.error(
-            "Transcription: the OpenAI account cannot be billed -- EVERY call is "
+            f"Transcription: the {target.provider} account cannot be billed -- calls on this "
-            "now stored with no transcript, so extraction, correlation and "
+            f"provider get no transcript (OpenAI fallback, if configured, is carrying them). "
-            "incidents are all dead downstream. Top up at "
+            f"Top up at {billing_url}. API said: {text}"
-            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
+            target.tier, target.provider, target.model, problem, fix, permanent=True
         )
         return
 
     if kind == "dead_model":
         problem = "the STT model is unavailable"
-        fix = "update STT_MODEL in config.py"
+        fix = "update STT_MODEL / STT_COMPAT_MODEL"
         logger.error(
-            f"Transcription: the configured model ({settings.stt_model!r}) is unavailable "
+            f"Transcription: the configured model {who} is unavailable -- calls on this provider "
-            "-- EVERY call is now stored with no transcript, so extraction, correlation "
+            f"get no transcript. Update STT_MODEL / STT_COMPAT_MODEL. API said: {text}"
-            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
+            target.tier, target.provider, target.model, problem, fix, permanent=True
         )
         return
 
-    logger.warning(f"Transcription failed for call {call_id}: {text}")
+    logger.warning(f"Transcription via {who} failed for call {call_id}: {text}")
     await ai_health.report_degraded(
-        "transcription", "openai", settings.stt_model,
+        target.tier, target.provider, target.model,
         "transient API error", "no action needed unless this persists", permanent=False,
     )
 
 
+# ── Pipeline entry point ────────────────────────────────────────────────────
+
+async def _transcribe_with(
+    call_id: str, tmp_path: str, target: SttTarget, prompt: str, glossary: list[str],
+) -> dict:
+    """One target, with the one retry on degenerate output. Raises on API failure."""
+    result = await asyncio.to_thread(_sync_transcribe_file, tmp_path, target, prompt, glossary)
+    # A hallucination is a coin-flip, not a property of the clip: call
+    # e49ea32c produced a 56-word ten-code counting run on one attempt and
+    # ordinary speech on the next, same audio and temperature=0. Discarding
+    # on the first bad roll threw away a recoverable transcript, so spend
+    # one more request before giving up.
+    if result["degenerate"] and settings.stt_retry_on_degenerate:
+        logger.info(f"Retrying transcription for call {call_id} after degenerate output")
+        result = await asyncio.to_thread(_sync_transcribe_file, tmp_path, target, prompt, glossary)
+        if result["degenerate"]:
+            logger.warning(f"Transcription for call {call_id} was degenerate twice — giving up")
+    return result
+
+
 async def transcribe_call(
     call_id: str,
     gcs_uri: str,
     talkgroup_name: Optional[str] = None,
     system_id: Optional[str] = None,
     talkgroup_id: Optional[int] = None,
+    node_id: Optional[str] = None,
 ) -> tuple[Optional[str], list[dict]]:
     """
     Transcribe audio at the given GCS URI and store the result in Firestore.
@@ -167,36 +430,78 @@ async def transcribe_call(
     if not gcs_uri or not gcs_uri.startswith("gs://"):
         return None, []
 
+    targets = stt_targets()
+    if not targets:
+        logger.warning("No STT provider configured (OPENAI_API_KEY / GROQ_API_KEY unset) — transcription disabled.")
+        return None, []
+
+    context: Optional[dict] = None
+    if any(t.vocab_prompt for t in targets):
         try:
-        transcript, segments, degenerate = await asyncio.to_thread(
+            context = await transcript_correction.resolve_context(system_id, talkgroup_id)
-            _sync_transcribe, gcs_uri, talkgroup_name
-        )
-        # A hallucination is a coin-flip, not a property of the clip: call
-        # e49ea32c produced a 56-word ten-code counting run on one attempt and
-        # ordinary speech on the next, same audio and temperature=0. Discarding
-        # on the first bad roll threw away a recoverable transcript, so spend
-        # one more request before giving up.
-        if degenerate and settings.stt_retry_on_degenerate:
-            logger.info(f"Retrying transcription for call {call_id} after degenerate output")
-            transcript, segments, degenerate = await asyncio.to_thread(
-                _sync_transcribe, gcs_uri, talkgroup_name
-            )
-            if degenerate:
-                logger.warning(
-                    f"Transcription for call {call_id} was degenerate twice — giving up"
-                )
         except Exception as e:
-        await _log_transcribe_failure(call_id, e)
+            logger.warning(f"Could not resolve STT vocabulary for call {call_id}: {e}")
+
+    try:
+        tmp_path = await asyncio.to_thread(_download, gcs_uri)
+    except Exception as e:
+        # Not a provider failure: classify() would read a GCS "not found" as a
+        # retired model and page someone to change STT_MODEL.
+        logger.warning(f"Could not download audio for call {call_id} ({gcs_uri}): {e}")
+        return None, []
+
+    result: Optional[dict] = None
+    used: Optional[SttTarget] = None
+    try:
+        for i, target in enumerate(targets):
+            prompt, glossary = build_stt_prompt(talkgroup_name, context if target.vocab_prompt else None)
+            try:
+                result = await _transcribe_with(call_id, tmp_path, target, prompt, glossary)
+            except Exception as e:
+                await _log_transcribe_failure(call_id, e, target)
+                if i + 1 < len(targets):
+                    logger.warning(
+                        f"Transcription for call {call_id}: {target.provider} failed, "
+                        f"falling back to {targets[i + 1].provider} ({targets[i + 1].model})"
+                    )
+                continue
+            used = target
+            break
+    finally:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+
+    if used is None or result is None:
         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
+    # that's a filtering decision, not a provider failure), so that tier is
-    # transcription tier is healthy and any prior degradation clears.
+    # healthy and any prior degradation clears. A fallback success clears
-    await ai_health.report_healthy("transcription")
+    # only the fallback tier: the primary is still broken and must stay red.
+    await ai_health.report_healthy(used.tier)
+
+    transcript: Optional[str] = result["text"]
+    segments: list[dict] = result["segments"]
 
     if transcript:
-        updates: dict = {"transcript": transcript}
+        updates: dict = {
+            "transcript": transcript,
+            # Which endpoint produced this text, and how much audio it billed,
+            # so cost (server-26#45) and an A/B of providers can be read off
+            # the call docs instead of guessed.
+            "stt_provider": used.provider,
+            "stt_model": used.model,
+        }
+        if result.get("duration") is not None:
+            updates["stt_audio_seconds"] = result["duration"]
+        if result.get("confidence") is not None:
+            # Mean per-token probability, exp(avg_logprob) weighted by segment
+            # length. The only STT-side confidence there is; the alert gate
+            # reads it (server-26#300).
+            updates["stt_confidence"] = result["confidence"]
         if segments:
             updates["segments"] = segments
 
@@ -212,17 +517,30 @@ async def transcribe_call(
         from app.internal.feature_flags import resolve_flags
         _, _ai_flag = await resolve_flags(system_id)
 
+        # server-26#298: radio housekeeping (roll call, a bare "Post 4, 10-8,
+        # copy") has no proper noun to correct, and extraction skips it too
+        # (intelligence.extract_scenes), so a correction pass here is pure
+        # spend. Same pure classifier both places, so they cannot disagree.
+        is_chatter, chatter_reason = (
+            gate_chatter(transcript) if settings.chatter_gate_enabled else (False, None)
+        )
+
         corrected, corrected_segments, not_speech = (None, None, False)
-        if _ai_flag("transcript_correction_enabled"):
+        if not _ai_flag("transcript_correction_enabled"):
+            logger.info(
+                f"Transcript correction disabled — saving raw transcript for call {call_id}"
+            )
+        elif is_chatter:
+            logger.info(
+                f"Transcript correction skipped for call {call_id}: chatter ({chatter_reason})"
+            )
+        else:
             corrected, corrected_segments, not_speech = await transcript_correction.correct(
                 call_id, transcript, segments,
                 system_id=system_id,
                 talkgroup_id=talkgroup_id,
                 talkgroup_name=talkgroup_name,
-            )
+                node_id=node_id,
-        else:
-            logger.info(
-                f"Transcript correction disabled — saving raw transcript for call {call_id}"
             )
         if corrected_segments:
             # Raw stays as evidence; the corrected copy is what extraction reads.
@@ -241,7 +559,7 @@ async def transcribe_call(
         try:
             await fstore.doc_set("calls", call_id, updates)
             logger.info(
-                f"Transcript saved for call {call_id} "
+                f"Transcript saved for call {call_id} via {used.provider}/{used.model} "
                 f"({len(transcript)} chars, {len(segments)} segment(s)"
                 f"{', corrected' if corrected and not not_speech else ''})"
             )
@@ -259,27 +577,10 @@ async def transcribe_call(
     return transcript, segments
 
 
-def _sync_transcribe(
+def _download(gcs_uri: str) -> str:
-    gcs_uri: str,
+    """Download the recording to a temp file once; the caller deletes it."""
-    talkgroup_name: Optional[str] = None,
-) -> tuple[Optional[str], list[dict], bool]:
-    """Download audio from GCS and transcribe with OpenAI Whisper.
-
-    Third element is True when output was DISCARDED as degenerate, which the
-    caller distinguishes from ordinary silence so it can retry — the same clip
-    can hallucinate on one attempt and transcribe on the next.
-    """
     from google.cloud import storage as gcs
     from google.oauth2 import service_account
-    from app.config import settings
-    from app.internal.ai_clients import openai_client
-
-    if not settings.openai_api_key:
-        logger.warning("OPENAI_API_KEY not set — transcription disabled.")
-        # Tuple, not a bare None: the caller unpacks two values, so returning
-        # None here raised a TypeError that surfaced as a misleading
-        # "Transcription failed" instead of the real missing-key warning.
-        return None, [], False
 
     without_scheme = gcs_uri[len("gs://"):]
     bucket_name, blob_path = without_scheme.split("/", 1)
@@ -293,34 +594,56 @@ def _sync_transcribe(
     else:
         gcs_client = gcs.Client()
 
-    bucket = gcs_client.bucket(bucket_name)
+    blob = gcs_client.bucket(bucket_name).blob(blob_path)
-    blob = bucket.blob(blob_path)
-
     suffix = os.path.splitext(blob_path)[1] or ".mp3"
     with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
         tmp_path = tmp.name
-
     try:
         blob.download_to_filename(tmp_path)
+    except Exception:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
+    return tmp_path
 
-        tg_prefix = f"Talkgroup: {talkgroup_name}. " if talkgroup_name else ""
-        # Vocabulary is intentionally excluded from the Whisper prompt.
-        # whisper-1 treats the prompt as a transcription prior and echoes
-        # vocabulary terms into noise/silence, polluting downstream extraction.
-        # Vocabulary context is applied in the GPT extraction step instead,
-        # where it is used as reference rather than a transcription prior.
-        prompt = tg_prefix + _WHISPER_PROMPT
 
-        # Only whisper-1 supports verbose_json (per-segment timestamps + no_speech_prob).
+def _segment_confidence(raw_segments: list) -> Optional[float]:
-        # gpt-4o-transcribe and gpt-4o-mini-transcribe only support json/text.
+    """exp(avg_logprob) per segment, weighted by segment length; None if absent."""
-        use_verbose = settings.stt_model == "whisper-1"
+    total, weight = 0.0, 0.0
+    for s in raw_segments:
+        lp = getattr(s, "avg_logprob", None)
+        if lp is None:
+            continue
+        w = max(0.01, float(getattr(s, "end", 0) or 0) - float(getattr(s, "start", 0) or 0))
+        total += math.exp(float(lp)) * w
+        weight += w
+    return round(total / weight, 3) if weight else None
 
-        # Own timeout, not openai_timeout_seconds: this request uploads the
+
-        # whole recording (server-26#153).
+def _sync_transcribe_file(
-        stt_client = openai_client(timeout=settings.stt_timeout_seconds)
+    tmp_path: str, target: SttTarget, prompt: str, glossary: Optional[list[str]] = None,
+) -> dict:
+    """Transcribe one downloaded file on one target.
+
+    Returns {"text", "segments", "degenerate", "duration", "confidence"}.
+    `degenerate` is True when output was DISCARDED as degenerate, which the
+    caller distinguishes from ordinary silence so it can retry — the same clip
+    can hallucinate on one attempt and transcribe on the next. Raises on any
+    API failure, so the caller can fall back to the next target.
+    """
+    from app.internal.ai_clients import stt_client
+
+    # verbose_json (per-segment timestamps + no_speech_prob) is a Whisper
+    # feature: whisper-1 on OpenAI, every whisper-* model on Groq. OpenAI's
+    # gpt-4o-transcribe / gpt-4o-mini-transcribe only support json/text.
+    use_verbose = target.model.startswith("whisper")
+
+    client = stt_client(target.api_key, target.base_url)
     with open(tmp_path, "rb") as f:
-            response = stt_client.audio.transcriptions.create(
+        response = client.audio.transcriptions.create(
-                model=settings.stt_model,
+            model=target.model,
             file=f,
             language="en",
             prompt=prompt,
@@ -338,32 +661,49 @@ def _sync_transcribe(
         #    start before the first PTT press.  Whisper sometimes hallucinates filler
         #    words or codes over this silence.  Detected via no_speech_prob > 0.8
         #    (Whisper's own confidence that a segment contains no real speech).
-            audio_duration: float = getattr(response, "duration", None) or float("inf")
+        duration = getattr(response, "duration", None)
+        audio_duration: float = duration or float("inf")
+        kept = [
+            s for s in (response.segments or [])
+            if (_seg(s, "text") or "").strip()
+            and (_seg(s, "start") or 0.0) < audio_duration
+            and (_seg(s, "no_speech_prob") or 0.0) < 0.8
+        ]
         segments = [
-                {"start": round(s.start, 2), "end": round(s.end, 2), "text": s.text.strip()}
+            {"start": round(_seg(s, "start") or 0.0, 2), "end": round(_seg(s, "end") or 0.0, 2),
-                for s in (response.segments or [])
+             "text": _seg(s, "text").strip()}
-                if s.text.strip()
+            for s in kept
-                and s.start < audio_duration
-                and getattr(s, "no_speech_prob", 0.0) < 0.8
         ]
         # Reconstruct text from non-hallucinated segments only so the two stay
         # in sync.  If every segment was filtered, text becomes None which prevents
         # the intelligence pipeline from running on hallucinated content.
         text = " ".join(s["text"] for s in segments) or None
-            if _is_degenerate(text or "", segments):
+        base = {"duration": duration, "confidence": _segment_confidence([_SegView(s) for s in kept])}
-                logger.info(f"Discarded hallucinated transcript for {gcs_uri}: {(text or '')[:80]!r}")
+        if _is_degenerate(text or "", segments, glossary):
-                return None, [], True
+            logger.info(f"Discarded hallucinated transcript from {target.provider}: {(text or '')[:80]!r}")
-            return text, segments, False
+            return {"text": None, "segments": [], "degenerate": True, **base}
-        else:
+        return {"text": text, "segments": segments, "degenerate": False, **base}
+
     # json format returns just {"text": "..."} — no segments or timestamps.
     # Intelligence extraction falls back to treating the whole transcript as one block.
     text = (response.text or "").strip() or None
-            if _is_degenerate(text or "", []):
+    if _is_degenerate(text or "", [], glossary):
-                logger.info(f"Discarded hallucinated transcript for {gcs_uri}: {(text or '')[:80]!r}")
+        logger.info(f"Discarded hallucinated transcript from {target.provider}: {(text or '')[:80]!r}")
-                return None, [], True
+        return {"text": None, "segments": [], "degenerate": True, "duration": None, "confidence": None}
-            return text, [], False
+    return {"text": text, "segments": [], "degenerate": False, "duration": None, "confidence": None}
-    finally:
+
-        try:
+
-            os.unlink(tmp_path)
+def _seg(s, field: str):
-        except OSError:
+    """Segment field from an SDK object or a plain dict (OpenAI-compatible
-            pass
+    servers do not all come back through the same pydantic model)."""
+    if isinstance(s, dict):
+        return s.get(field)
+    return getattr(s, field, None)
+
+
+class _SegView:
+    """Attribute view over a segment, for _segment_confidence."""
+    def __init__(self, s):
+        self.avg_logprob = _seg(s, "avg_logprob")
+        self.start = _seg(s, "start")
+        self.end = _seg(s, "end")

drb-c2-core/app/routers/upload.py

@@ -586,7 +586,7 @@ async def _run_intelligence_pipeline(
         if _flag("stt_enabled"):
             transcript, segments = await transcription.transcribe_call(
                 call_id, gcs_uri, talkgroup_name,
-                system_id=system_id, talkgroup_id=talkgroup_id,
+                system_id=system_id, talkgroup_id=talkgroup_id, node_id=node_id,
             )
         else:
             scope = "globally" if not flags["stt_enabled"] else f"system {system_id}"

drb-c2-core/tests/test_ai_flag_gating.py

@@ -219,15 +219,21 @@ async def test_transcript_correction_is_skipped_when_its_flag_is_off():
          patch.object(transcription, "fstore") as fs, \
          patch.object(transcription, "ai_health") as health, \
          patch.object(transcription, "transcript_correction") as tc, \
-         patch("asyncio.to_thread", AsyncMock(return_value=("units respond", [], False))):
+         patch.object(transcription.settings, "openai_api_key", "k"), \
+         patch.object(transcription.settings, "groq_api_key", None), \
+         patch.object(transcription, "_download", lambda uri: "/nonexistent.mp3"), \
+         patch.object(transcription, "_sync_transcribe_file", lambda *a, **k: {
+             "text": "units respond to the fire on Main Street", "segments": [],
+             "degenerate": False, "duration": 3.0, "confidence": None}):
         fs.doc_set = AsyncMock()
         health.report_healthy = AsyncMock()
         health.report_failure = AsyncMock()
         tc.correct = AsyncMock()
-        await transcription.transcribe_call(
+        text, _ = await transcription.transcribe_call(
             "call-1", "gs://bucket/call-1.mp3", "PD Dispatch", system_id="sys-1"
         )
 
+    assert text == "units respond to the fire on Main Street", "the transcript must still be produced"
     tc.correct.assert_not_awaited()
 
 

drb-c2-core/tests/test_stt_provider.py

@@ -0,0 +1,279 @@
+"""
+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)

infra/ansible/roles/deploy/templates/c2-core.env.j2

@@ -26,6 +26,10 @@ PUBLIC_API_URL=https://api.{{ domain }}
 OPENAI_API_KEY={{ vault_openai_api_key }}
 GOOGLE_MAPS_API_KEY={{ vault_google_maps_api_key }}
 GEMINI_API_KEY={{ vault_gemini_api_key }}
+# STT provider (server-26#297). With this set, STT_PROVIDER=auto moves
+# transcription to Groq whisper-large-v3-turbo with OpenAI as the fallback;
+# default('') keeps a vault that predates the key on OpenAI, unchanged.
+GROQ_API_KEY={{ vault_groq_api_key | default('') }}
 
 SERVICE_KEY={{ vault_service_key }}
 ENROLLMENT_TOKEN={{ vault_enrollment_token }}