From 7ecfca76482ead5bfabef3580209fb3fcf757c1f Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:28:29 -0400 Subject: [PATCH 1/5] Correction: reject lost numbers and growth; place verifier anchors from the talkgroup name (server-26#162, server-26#160) transcript_correction now also discards a correction that drops or changes a bare number the recogniser heard (the live "7" -> "ShotSpotter" case) or grows the transcript past max(4 words, 25%). place_verifier.verify() derives an anchor from the talkgroup's own name plus the receiving node's state when no area is configured; no state means no anchor, and a configured place without an anchor keeps the load-bearing skip. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/internal/place_verifier.py | 61 +++++++++++++++++-- .../app/internal/transcript_correction.py | 48 +++++++++++++++ .../tests/test_transcript_correction.py | 32 +++++++++- 3 files changed, 136 insertions(+), 5 deletions(-) diff --git a/drb-c2-core/app/internal/place_verifier.py b/drb-c2-core/app/internal/place_verifier.py index d83394b..ebbd191 100644 --- a/drb-c2-core/app/internal/place_verifier.py +++ b/drb-c2-core/app/internal/place_verifier.py @@ -23,7 +23,8 @@ permanently improves the reference data for that channel. NO ANCHOR MEANS SKIP, NOT ACCEPT. An anchor too wide to discriminate is not stored at all (see `area_context`), and without one this module returns -immediately. A statewide radius would confirm anything inside it, which is worse +immediately. A channel nobody described may still get one from its own +talkgroup name plus the node's state (server-26#160, `_talkgroup_anchor`). A statewide radius would confirm anything inside it, which is worse than not checking — it looks like verification and is not. THE FREE TIER RUNS FIRST. A sound-alike among the terms the operator already @@ -169,6 +170,50 @@ def _known_soundalike(heard: str, area: dict) -> Optional[dict]: return best +async def _talkgroup_anchor( + talkgroup_name: Optional[str], node_id: Optional[str], +) -> tuple[Optional[dict], str]: + """ + An anchor derived from the talkgroup's own name plus the receiving node's + state, or (None, ""). Same width rule as an operator anchor + (area_context.resolve_anchor drops anything wider than + area_anchor_max_radius_km, and caches per place for the process). + + A STATE IS REQUIRED. "Springfield PD" alone geocodes to whichever + Springfield Google prefers, and an anchor in the wrong half of the + country would let Maps rewrite real local streets into sound-alikes from + there (the failure server-26#36 added `state` to prevent). The node's + reverse-geocoded state is coarse enough to be right even when the node + sits far from the area the channel covers. + """ + from app.internal import firestore as fstore + from app.internal.intelligence import _get_node_state, _municipality_from_tg + + muni = _municipality_from_tg(talkgroup_name) + if not muni or not node_id: + return None, "" + node = await fstore.doc_get_cached("nodes", node_id) or {} + lat, lon = node.get("lat"), node.get("lon") + if lat is None or lon is None: + return None, "" + state = await _get_node_state(node_id, lat, lon) + if not state: + return None, "" + place = {"municipality": muni, "state": state} + resolved = await area_context.resolve_anchor(place) + if not resolved: + return None, "" + try: + anchor = { + "lat": float(resolved["center"]["lat"]), + "lng": float(resolved["center"]["lng"]), + "radius_km": float(resolved["radius_km"]), + } + except (KeyError, TypeError, ValueError): + return None, "" + return anchor, f"{muni}, {state}" + + # -- Public -------------------------------------------------------------------- def _substitute(text: str, swaps: list[tuple[str, str]]) -> str: @@ -186,6 +231,8 @@ async def verify( tg_area: Optional[dict], system_id: Optional[str] = None, talkgroup_id: Optional[Any] = None, + talkgroup_name: Optional[str] = None, + node_id: Optional[str] = None, ) -> tuple[Optional[str], Optional[list[dict]]]: """ Check the corrector's location nouns against the talkgroup's anchor. @@ -197,12 +244,18 @@ async def verify( if not settings.place_verification_enabled or not locations: return None, None + area = area_context.effective(system_area, tg_area) anchor = area_context.anchor_for(system_area, tg_area) + suffix = _place_suffix(area) + if not anchor and not area_context.has_place(area): + # server-26#160: nobody described this channel, but its own name may + # (e.g. "New York City - NYPD Citywide 2 Patch") — the same signal + # #159 trusted in intelligence.py's geocoder. Only when an operator set + # NOTHING: a place that is set but has no anchor was too wide or is + # stale, and that skip stays load-bearing. + anchor, suffix = await _talkgroup_anchor(talkgroup_name, node_id) if not anchor: return None, None # load-bearing: no anchor means skip, never accept - - area = area_context.effective(system_area, tg_area) - suffix = _place_suffix(area) swaps: list[tuple[str, str]] = [] proposals: list[dict] = [] diff --git a/drb-c2-core/app/internal/transcript_correction.py b/drb-c2-core/app/internal/transcript_correction.py index e1a25bb..2f09fd4 100644 --- a/drb-c2-core/app/internal/transcript_correction.py +++ b/drb-c2-core/app/internal/transcript_correction.py @@ -116,11 +116,37 @@ def _dedupe(items: list[str]) -> list[str]: # mishearing this pass exists to fix (server-26#162). _CODE_TOKEN_RE = re.compile(r"\b\d{1,3}(?:-\d{1,3})+\b") +# Bare numbers (the rest of server-26#162): the same call had a bare "7" +# turned into "ShotSpotter". A unit number, a post, a house number or a count +# the recogniser DID hear must survive, in order. A number may still be +# added, because a sound-alike can legitimately be one ("back to Rose" -> +# "back to Route 9"). "4th" is not bare (it is a word). +_BARE_NUMBER_RE = re.compile(r"(? "Home Street and Forest Ave"); growing +# by more than this is the model writing, not correcting. +_MAX_ADDED_WORDS = 4 +_MAX_GROWTH_RATIO = 0.25 + def _code_tokens(text: str) -> list[str]: return _CODE_TOKEN_RE.findall(text or "") +def _lost_numbers(original: str, corrected: str) -> bool: + """True unless every bare number of `original` is still in `corrected`, in order.""" + remaining = iter(_BARE_NUMBER_RE.findall(corrected or "")) + return not all(any(n == m for m in remaining) for n in _BARE_NUMBER_RE.findall(original or "")) + + +def _grew(original: str, corrected: str) -> bool: + n_orig = len((original or "").split()) + n_new = len((corrected or "").split()) + return n_new - n_orig > max(_MAX_ADDED_WORDS, int(n_orig * _MAX_GROWTH_RATIO)) + + def _talkgroup_entry(system_doc: dict, talkgroup_id: Optional[int]) -> dict: """The config.talkgroups[] entry for this talkgroup, or {}.""" if talkgroup_id is None: @@ -241,6 +267,7 @@ async def correct( system_id: Optional[str] = None, talkgroup_id: Optional[int] = None, talkgroup_name: Optional[str] = None, + node_id: Optional[str] = None, ) -> tuple[Optional[str], Optional[list[dict]], bool]: """ Second-opinion pass over a transcript. @@ -330,6 +357,8 @@ async def correct( context.get("tg_area"), system_id=system_id, talkgroup_id=talkgroup_id, + talkgroup_name=talkgroup_name, + node_id=node_id, ) except Exception as e: logger.warning(f"Place verification failed for call {call_id}: {e}") @@ -354,6 +383,18 @@ async def correct( f"discarding the joined correction" ) corrected = None + if corrected is not None and _lost_numbers(text, corrected): + logger.warning( + f"Transcript correction for call {call_id} dropped or changed a number — " + f"discarding the joined correction" + ) + corrected = None + if corrected is not None and _grew(text, corrected): + logger.warning( + f"Transcript correction for call {call_id} grew the transcript " + f"({len(text.split())} -> {len(corrected.split())} words) — discarding the joined correction" + ) + corrected = None if corrected_segments is not None: for seg, orig in zip(corrected_segments, segments or []): if _code_tokens(seg["text"]) != _code_tokens(orig.get("text", "")): @@ -363,6 +404,13 @@ async def correct( ) corrected_segments = None break + if _lost_numbers(orig.get("text", ""), seg["text"]) or _grew(orig.get("text", ""), seg["text"]): + logger.warning( + f"Transcript correction for call {call_id} grew a segment or dropped " + f"a number in one — discarding segment corrections" + ) + corrected_segments = None + break if corrected or corrected_segments or not_speech: changed = raw.get("changed") or [] diff --git a/drb-c2-core/tests/test_transcript_correction.py b/drb-c2-core/tests/test_transcript_correction.py index 060328e..dabd279 100644 --- a/drb-c2-core/tests/test_transcript_correction.py +++ b/drb-c2-core/tests/test_transcript_correction.py @@ -260,7 +260,10 @@ async def test_segment_code_change_discards_segments_only(): "segments": ["Headquarters, 10-13.", "Show it out to Ossining.", "360 north, back to Route 9."], } with _system(), _gemini(payload): - text, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1", talkgroup_id=9048) + # A realistic joined original (code-free, like the correction): a + # placeholder like "x y z w" now trips the server-26#162 growth guard. + text, segs, _ = await tc.correct("c1", "Shout out to Optum, back to Rose.", SEGS, + system_id="sys-1", talkgroup_id=9048) assert segs is None assert text == "Show it out to Ossining, back to Route 9." @@ -279,3 +282,30 @@ async def test_reference_data_reaches_the_prompt(): assert "Sing Sing — state prison" in p, "a term without its meaning is half the information" assert "Ossining, Westchester, New York" in p, "state must reach the prompt (server-26#36)" assert "10-13=officer needs assistance" in p + + +# ── server-26#162, the rest: bare numbers and invented words ──────────────── + +@pytest.mark.asyncio +async def test_a_bare_number_turned_into_a_word_is_rejected(): + """The live case: a bare "7" came back as "ShotSpotter".""" + payload = {"corrected": "Car ShotSpotter on Home Street and Forest Ave."} + with _system(), _gemini(payload): + text, _, _ = await tc.correct("c1", "Car 7 on Holmes Street and 4th and Rowe.", None, system_id="sys-1") + assert text is None + + +@pytest.mark.asyncio +async def test_a_place_fix_that_keeps_its_numbers_still_passes(): + payload = {"corrected": "Car 7 on Home Street and Forest Ave."} + with _system(), _gemini(payload): + text, _, _ = await tc.correct("c1", "Car 7 on Holmes Street and 4th and Rowe.", None, system_id="sys-1") + assert text == "Car 7 on Home Street and Forest Ave." + + +@pytest.mark.asyncio +async def test_a_correction_that_writes_new_sentences_is_rejected(): + payload = {"corrected": "Car 7 on Main Street. Suspect is armed and fled northbound in a red sedan."} + with _system(), _gemini(payload): + text, _, _ = await tc.correct("c1", "Car 7 on Mane Street.", None, system_id="sys-1") + assert text is None From d485eae41ab3c5f3ca76c75608b87f9d336adfe6 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:28:30 -0400 Subject: [PATCH 2/5] 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 --- drb-c2-core/app/config.py | 31 + drb-c2-core/app/internal/ai_clients.py | 18 + .../app/internal/chatter_classifier.py | 51 +- .../app/internal/recorrelation_sweep.py | 6 +- drb-c2-core/app/internal/replay.py | 4 + drb-c2-core/app/internal/transcription.py | 602 ++++++++++++++---- drb-c2-core/app/routers/upload.py | 2 +- drb-c2-core/tests/test_ai_flag_gating.py | 10 +- drb-c2-core/tests/test_stt_provider.py | 279 ++++++++ .../roles/deploy/templates/c2-core.env.j2 | 4 + 10 files changed, 865 insertions(+), 142 deletions(-) create mode 100644 drb-c2-core/tests/test_stt_provider.py diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 922aeed..c9e338f 100644 --- a/drb-c2-core/app/config.py +++ b/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 diff --git a/drb-c2-core/app/internal/ai_clients.py b/drb-c2-core/app/internal/ai_clients.py index 544ce20..149e5e7 100644 --- a/drb-c2-core/app/internal/ai_clients.py +++ b/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 diff --git a/drb-c2-core/app/internal/chatter_classifier.py b/drb-c2-core/app/internal/chatter_classifier.py index 81e244d..dcb595b 100644 --- a/drb-c2-core/app/internal/chatter_classifier.py +++ b/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 -extraction based on this verdict. `intelligence.extract_scenes` calls -`classify_chatter` purely to record the verdict on the call doc -(`chatter_classifier_verdict` / `chatter_classifier_reason`) so it becomes -observable in the next `/admin` correlation-debug dump, exactly like -`corr_gate_veto` (server-26#115 / PR #126). See the TODO at that call site for -what has to be true before this flips live. +LIVE AS A COST GATE SINCE server-26#298, through `gate_chatter`, not +`classify_chatter`. `classify_chatter` is still the shadow verdict recorded on +every call doc (`chatter_classifier_verdict` / `chatter_classifier_reason`), +unchanged, so the `/admin` tallies keep measuring the same thing. The gate is +stricter on the one shape the #127 review flagged as a live false-positive +risk: `roll_call` matches "roll call" anywhere, so "Delayed for roll call, then +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 diff --git a/drb-c2-core/app/internal/recorrelation_sweep.py b/drb-c2-core/app/internal/recorrelation_sweep.py index a4d9fed..abff1d9 100644 --- a/drb-c2-core/app/internal/recorrelation_sweep.py +++ b/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. diff --git a/drb-c2-core/app/internal/replay.py b/drb-c2-core/app/internal/replay.py index ee7b3a8..eed3a00 100644 --- a/drb-c2-core/app/internal/replay.py +++ b/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) diff --git a/drb-c2-core/app/internal/transcription.py b/drb-c2-core/app/internal/transcription.py index 7218bb6..ac88cdd 100644 --- a/drb-c2-core/app/internal/transcription.py +++ b/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 -returning None on any failure so the intelligence pipeline can still run. +Audio is downloaded from GCS once, then sent to an OpenAI-compatible +/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" - fix = "top up at https://platform.openai.com/settings/organization/billing" + problem = f"the {target.provider} account cannot be billed" + fix = f"top up at {billing_url}" logger.error( - "Transcription: the OpenAI account cannot be billed -- EVERY call is " - "now stored with no transcript, so extraction, correlation and " - "incidents are all dead downstream. Top up at " - f"https://platform.openai.com/settings/organization/billing. API said: {text}" + f"Transcription: the {target.provider} account cannot be billed -- calls on this " + f"provider get no transcript (OpenAI fallback, if configured, is carrying them). " + f"Top up at {billing_url}. 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 " - "-- EVERY call is now stored with no transcript, so extraction, correlation " - f"and incidents are all dead downstream. Update STT_MODEL in config.py. API said: {text}" + f"Transcription: the configured model {who} is unavailable -- calls on this provider " + f"get no transcript. Update STT_MODEL / STT_COMPAT_MODEL. 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: + context = await transcript_correction.resolve_context(system_id, talkgroup_id) + except Exception as e: + logger.warning(f"Could not resolve STT vocabulary for call {call_id}: {e}") + try: - transcript, segments, degenerate = await asyncio.to_thread( - _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" - ) + tmp_path = await asyncio.to_thread(_download, gcs_uri) except Exception as e: - await _log_transcribe_failure(call_id, 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 - # transcription tier is healthy and any prior degradation clears. - await ai_health.report_healthy("transcription") + # that's a filtering decision, not a provider failure), so that tier is + # healthy and any prior degradation clears. A fallback success clears + # 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, - ) - else: - logger.info( - f"Transcript correction disabled — saving raw transcript for call {call_id}" + node_id=node_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( - gcs_uri: str, - 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. - """ +def _download(gcs_uri: str) -> str: + """Download the recording to a temp file once; the caller deletes it.""" 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,77 +594,116 @@ def _sync_transcribe( else: gcs_client = gcs.Client() - bucket = gcs_client.bucket(bucket_name) - blob = bucket.blob(blob_path) - + blob = gcs_client.bucket(bucket_name).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) - - 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). - # gpt-4o-transcribe and gpt-4o-mini-transcribe only support json/text. - use_verbose = settings.stt_model == "whisper-1" - - # Own timeout, not openai_timeout_seconds: this request uploads the - # whole recording (server-26#153). - stt_client = openai_client(timeout=settings.stt_timeout_seconds) - with open(tmp_path, "rb") as f: - response = stt_client.audio.transcriptions.create( - model=settings.stt_model, - file=f, - language="en", - prompt=prompt, - response_format="verbose_json" if use_verbose else "json", - temperature=0, - ) - - if use_verbose: - # Filter hallucinated segments. Two sources of hallucination in P25 recordings: - # - # 1. Trailing silence / static — Whisper fills silence past real content with - # sequential radio codes (10-4, 10-5...). Clamped by audio duration. - # - # 2. Leading silence — OP25 recordings typically have a short silence at the - # 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") - segments = [ - {"start": round(s.start, 2), "end": round(s.end, 2), "text": s.text.strip()} - for s in (response.segments or []) - if s.text.strip() - 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): - logger.info(f"Discarded hallucinated transcript for {gcs_uri}: {(text or '')[:80]!r}") - return None, [], True - return text, segments, False - else: - # 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 "", []): - logger.info(f"Discarded hallucinated transcript for {gcs_uri}: {(text or '')[:80]!r}") - return None, [], True - return text, [], False - finally: + except Exception: try: os.unlink(tmp_path) except OSError: pass + raise + return tmp_path + + +def _segment_confidence(raw_segments: list) -> Optional[float]: + """exp(avg_logprob) per segment, weighted by segment length; None if absent.""" + 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 + + +def _sync_transcribe_file( + 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 = client.audio.transcriptions.create( + model=target.model, + file=f, + language="en", + prompt=prompt, + response_format="verbose_json" if use_verbose else "json", + temperature=0, + ) + + if use_verbose: + # Filter hallucinated segments. Two sources of hallucination in P25 recordings: + # + # 1. Trailing silence / static — Whisper fills silence past real content with + # sequential radio codes (10-4, 10-5...). Clamped by audio duration. + # + # 2. Leading silence — OP25 recordings typically have a short silence at the + # 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). + 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(_seg(s, "start") or 0.0, 2), "end": round(_seg(s, "end") or 0.0, 2), + "text": _seg(s, "text").strip()} + for s in kept + ] + # 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 + base = {"duration": duration, "confidence": _segment_confidence([_SegView(s) for s in kept])} + if _is_degenerate(text or "", segments, glossary): + logger.info(f"Discarded hallucinated transcript from {target.provider}: {(text or '')[:80]!r}") + return {"text": None, "segments": [], "degenerate": True, **base} + 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 "", [], glossary): + logger.info(f"Discarded hallucinated transcript from {target.provider}: {(text or '')[:80]!r}") + return {"text": None, "segments": [], "degenerate": True, "duration": None, "confidence": None} + return {"text": text, "segments": [], "degenerate": False, "duration": None, "confidence": None} + + +def _seg(s, field: str): + """Segment field from an SDK object or a plain dict (OpenAI-compatible + 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") diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index 43b72b5..c446469 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/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}" diff --git a/drb-c2-core/tests/test_ai_flag_gating.py b/drb-c2-core/tests/test_ai_flag_gating.py index b7524c0..f2c1122 100644 --- a/drb-c2-core/tests/test_ai_flag_gating.py +++ b/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() diff --git a/drb-c2-core/tests/test_stt_provider.py b/drb-c2-core/tests/test_stt_provider.py new file mode 100644 index 0000000..2fe61e6 --- /dev/null +++ b/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) diff --git a/infra/ansible/roles/deploy/templates/c2-core.env.j2 b/infra/ansible/roles/deploy/templates/c2-core.env.j2 index 8291115..2050ade 100644 --- a/infra/ansible/roles/deploy/templates/c2-core.env.j2 +++ b/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 }} From 240bbee5a4facc5cf520720fa90a0dd9e614c09c Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:28:46 -0400 Subject: [PATCH 3/5] Extraction: model in config, chatter gate, geocode cache (server-26#298, server-26#104, server-26#88) EXTRACTION_MODEL / EMBEDDING_MODEL replace the gpt-4o-mini and text-embedding-3-small literals. gate_chatter (classify_chatter with roll_call narrowed to whole-transcript roll calls) skips correction, extraction, embedding and geocoding for housekeeping, writes skip_reason=non_event_chatter and still runs the free clearance/plate backstops; upload.py's no-scene fallback admits it like transcript_too_short. Ships OFF (CHATTER_GATE_ENABLED =false): a gated call has no units, so it takes the no-fit-check thin attach, and #127 requires a clean live shadow window first. In-process TTL cache (24h, negatives 1h, 5000 entries, LRU) in front of the Geocoding API, keyed by address+bounds; holds coordinates only, inside the Maps 30-day lat/lng caching limit; errors and key/quota statuses not cached. #104: the no-scene fallback passes the call doc's severity. #88: resolved/reassignment parsed by _llm_bool, not bool(). Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/config.py | 23 ++ drb-c2-core/app/internal/intelligence.py | 266 ++++++++++++------ drb-c2-core/app/routers/upload.py | 10 +- drb-c2-core/tests/conftest.py | 10 + drb-c2-core/tests/test_chatter_gate.py | 194 +++++++++++++ drb-c2-core/tests/test_geocode_cache.py | 162 +++++++++++ .../tests/test_intelligence_llm_bool.py | 14 + 7 files changed, 597 insertions(+), 82 deletions(-) create mode 100644 drb-c2-core/tests/test_chatter_gate.py create mode 100644 drb-c2-core/tests/test_geocode_cache.py create mode 100644 drb-c2-core/tests/test_intelligence_llm_bool.py diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index c9e338f..66ff4fb 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -79,8 +79,31 @@ class Settings(BaseSettings): # google-genai does not retry unless told to, so this is the whole bound. gemini_timeout_seconds: float = 60.0 + # Scene extraction + embeddings (server-26#298). Were literals in + # intelligence.py; any OpenAI chat model that supports JSON mode works. + extraction_model: str = "gpt-4o-mini" + embedding_model: str = "text-embedding-3-small" + # Skip correction + extraction (+ embedding, geocoding) for transcripts + # chatter_classifier.gate_chatter calls radio housekeeping. OFF by default: + # a gated call carries no units, so it is thin and takes the no-fit-check + # thin attach (mechanism A) instead of the fit-tested path or the #115 + # orphan gate, and server-26#127 set "0 false positives in a live shadow + # window" as the bar for going live. Read chatter_classifier_verdict off a + # shadow window first, then set CHATTER_GATE_ENABLED=true. + chatter_gate_enabled: bool = False + # Google Maps (geocoding) google_maps_api_key: Optional[str] = None + # In-process cache of Geocoding API answers (server-26#298). Dispatch + # repeats the same address on every follow-up call of a job, and the + # place verifier re-geocodes the same nouns. In-process rather than + # Firestore: a Firestore read costs about what it saves, adds latency on + # every miss, and would persist Maps content -- the Maps terms allow + # caching lat/lng for at most 30 days, and an in-memory TTL can never + # outlive that. Lost on restart, which only costs a few repeat lookups. + geocode_cache_ttl_seconds: int = 24 * 60 * 60 + geocode_cache_negative_ttl_seconds: int = 60 * 60 # ZERO_RESULTS etc. + geocode_cache_max_entries: int = 5000 # 511NY developer key (server-26#183/#198). Issued with the NYSDOT Developer # Access Agreement grant; the keyless legacy API is gone since the 511x cutover. ny511_api_key: Optional[str] = None diff --git a/drb-c2-core/app/internal/intelligence.py b/drb-c2-core/app/internal/intelligence.py index 6e8705f..c3effea 100644 --- a/drb-c2-core/app/internal/intelligence.py +++ b/drb-c2-core/app/internal/intelligence.py @@ -1,7 +1,8 @@ """ -GPT-4o-mini intelligence extraction from call transcripts. +Intelligence extraction from call transcripts. -Sends the transcript to GPT-4o-mini with a structured prompt that detects +Sends the transcript to the extraction model (settings.extraction_model, +gpt-4o-mini by default) with a structured prompt that detects whether the recording contains one or multiple distinct scenes (back-to-back dispatch conversations on a busy channel). Returns a list of scene dicts — one per detected incident. Most calls produce a single scene. @@ -12,12 +13,13 @@ import asyncio import json import math import re +from collections import OrderedDict 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 area_context -from app.internal.chatter_classifier import classify_chatter +from app.internal.chatter_classifier import classify_chatter, gate_chatter # Location validity is defined once, by the module that owns the incident's # location/pin invariant. incident_correlator does not import this module, so # this is not a cycle. @@ -147,6 +149,23 @@ def _is_garbage_transcript(transcript: str) -> bool: return False +def _llm_bool(value) -> bool: + """ + A model's yes/no field as a real bool (server-26#88). True only for a + JSON true or the strings "true"/"yes" (any case); everything else — + "false", "no", null, a number, garbage — is False, because both fields + this reads (resolved, reassignment) change incident state when True. + + Local stand-in for incident_correlator.parse_llm_bool, which lives on the + correlation branch and is not on main yet; fold into it after that merges. + """ + if isinstance(value, bool): + return value + if isinstance(value, str): + return value.strip().lower() in ("true", "yes") + return False + + def _build_ten_codes_block(ten_codes: dict[str, str]) -> str: if not ten_codes: return "" @@ -218,11 +237,12 @@ async def extract_scenes( pass return [] - # server-26#127 — SHADOW MODE ONLY. Computes whether this transcript looks + # server-26#127 — the shadow verdict. Computes whether this transcript looks # like non-event radio housekeeping (roll call, bare 10-4/10-8/98 - # acknowledgements, unit check-ins) and records the verdict on the call - # doc, but does NOT skip extraction anywhere below — every path runs - # exactly as it did before this landed. Deliberately ahead of the ≤5-word + # acknowledgements, unit check-ins) and records it on the call doc. The + # live cost gate (server-26#298, gate_chatter below) is a stricter view of + # the same classifier; this verdict is recorded unchanged so the /admin + # tallies keep measuring the same thing. Deliberately ahead of the ≤5-word # skip: most bare acknowledgements ARE ≤5 words, and the first pass of # this feature put the classifier after that return, so it never saw the # bulk of its own target population — a review backtest against three @@ -231,13 +251,9 @@ async def extract_scenes( # roughly a fifth of the real catch rate. Computing it once, here, and # folding the result into whichever skip/continue path runs below fixes # that without adding a second Firestore write. - # TODO(server-26#127): flip this from shadow to live (skip extraction and - # write skip_reason="non_event_chatter" instead of just recording the - # verdict) once a live shadow-mode window confirms 0 false positives on - # real production traffic — pay particular attention to whole-transcript - # vs contains-anywhere matching for "roll call" and to digit-hyphen street - # addresses (e.g. "72-Holland"), both flagged as classifier risks that the - # dump backtest could not surface on its own. + # Residual risk of the live gate, unchanged from the #127 review: a + # digit-hyphen street address ("72-Holland") alone in a transmission reads + # as a unit ID. CHATTER_GATE_ENABLED=false restores shadow mode. chatter_is_chatter, chatter_reason = classify_chatter(transcript) # Transcripts with ≤5 words carry no extractable intelligence — GPT hallucinates @@ -277,6 +293,40 @@ async def extract_scenes( "tags": tags, "incident_type": typ, "severity": sev}] return [] + # server-26#298 — the gate is live. Radio housekeeping has no incident + # content for the extraction model to find, so it no longer pays for an + # extraction, an embedding per scene and a geocode to come back as a thin + # routine scene. gate_chatter is classify_chatter with roll_call narrowed + # to whole-transcript roll calls (the #127 review's false-positive shape). + # The free rule paths still run, exactly as on the <=5-word path above. + from app.config import settings + gated, gate_reason = gate_chatter(transcript) if settings.chatter_gate_enabled else (False, None) + if gated: + logger.info( + f"Intelligence: call {call_id} — non-event chatter ({gate_reason}), skipping extraction" + ) + cleared_unit = _short_clearance_unit(transcript) + tags, typ, sev = _self_initiated_backstop(transcript, [], None, "routine", talkgroup_name) + updates = { + "skip_reason": "non_event_chatter", + "severity": sev if tags else "routine", + "chatter_classifier_verdict": chatter_is_chatter, + "chatter_classifier_reason": chatter_reason, + } + if cleared_unit: + updates["units"] = [cleared_unit] + updates["cleared_units"] = [cleared_unit] + try: + await fstore.doc_set("calls", call_id, updates) + except Exception: + pass + if cleared_unit: + return [_clearance_scene(transcript, cleared_unit)] + if tags: + return [{**_clearance_scene(transcript, ""), "units": [], "cleared_units": [], + "tags": tags, "incident_type": typ, "severity": sev}] + return [] + try: await fstore.doc_set("calls", call_id, { "chatter_classifier_verdict": chatter_is_chatter, @@ -294,13 +344,14 @@ async def extract_scenes( except Exception as e: text = str(e) kind = ai_health.classify(text) - logger.warning(f"GPT-4o-mini extraction failed for call {call_id}: {text}") + from app.config import settings as _s + logger.warning(f"Extraction ({_s.extraction_model}) failed for call {call_id}: {text}") await ai_health.report_degraded( - "extraction", "openai", "gpt-4o-mini", + "extraction", "openai", _s.extraction_model, {"billing": "the OpenAI account is out of credit", "dead_model": "model is unavailable"}.get(kind, f"extraction failed: {text[:200]}"), {"billing": "Top up OpenAI billing", - "dead_model": "Update the extraction model in intelligence.py"}.get(kind, "Usually transient"), + "dead_model": "Update EXTRACTION_MODEL"}.get(kind, "Usually transient"), permanent=kind != "transient", ) return [] @@ -364,8 +415,10 @@ async def extract_scenes( severity: str = scene.get("severity") or "routine" if severity == "unknown": severity = "routine" - resolved: bool = bool(scene.get("resolved", False)) - reassignment: bool = bool(scene.get("reassignment", False)) + # server-26#88: bool("false") is True, which let a model answering + # with a JSON string resolve an incident it merely linked into. + resolved: bool = _llm_bool(scene.get("resolved")) + reassignment: bool = _llm_bool(scene.get("reassignment")) transcript_corrected: Optional[str]= scene.get("transcript_corrected") or None segment_indices: Optional[list] = scene.get("segment_indices") @@ -729,7 +782,6 @@ async def _geocode_location( touch the anchor path at all — an anchor's own radius is always authoritative when one has been resolved. """ - import httpx from app.config import settings if not settings.google_maps_api_key: @@ -755,70 +807,122 @@ async def _geocode_location( f"{ref_lat - delta},{ref_lon - delta}" f"|{ref_lat + delta},{ref_lon + delta}" ) - params = { - "address": location_str, - "bounds": bounds, - "region": "us", - "key": settings.google_maps_api_key, - } try: - async with httpx.AsyncClient(timeout=5.0) as client: - r = await client.get( - "https://maps.googleapis.com/maps/api/geocode/json", - params=params, - ) - r.raise_for_status() - data = r.json() - if data.get("status") != "OK" or not data.get("results"): - return None - result = data["results"][0] - location_type = result.get("geometry", {}).get("location_type", "") - # Reject only APPROXIMATE — a region/city boundary centroid, which is - # what an ungeocodable string degrades to and is genuinely useless. - # - # ROOFTOP-only was too strict and emptied the map: dispatch names - # places the way people speak, and Google returns GEOMETRIC_CENTER for - # exactly those forms — intersections ("Lake Street and Veterans - # Memorial Drive") and named POIs ("Brewster Station"). Both are - # precise enough to plot and to proximity-match; requiring a street - # address threw away nearly every real dispatch location, leaving only - # numbered addresses geocoded. - if location_type not in ("ROOFTOP", "RANGE_INTERPOLATED", "GEOMETRIC_CENTER"): - logger.info( - f"Geocoding rejected '{location_str}' — imprecise result " - f"(location_type={location_type!r}), returning None" - ) - return None - loc = result["geometry"]["location"] - lat, lng = float(loc["lat"]), float(loc["lng"]) - dist_km = _geo_dist_km(ref_lat, ref_lon, lat, lng) - if dist_km > max_km: - # server-26#159: the node-distance bound is a proxy for "is - # this plausible" that only makes sense when the node's own - # position is our best guess at the area — never when the - # query already names a different region on its own terms. - if not (ref_label == "node" and trust_named_region): - logger.warning( - f"Geocoding rejected '{location_str}' → ({lat:.4f}, {lng:.4f}) " - f"— {dist_km:.1f}km from {ref_label} exceeds {max_km:.1f}km" - ) - return None - logger.info( - f"Geocoding '{location_str}' → ({lat:.4f}, {lng:.4f}) is " - f"{dist_km:.1f}km from the receiving node, past {max_km:.1f}km — " - f"accepted anyway: the query names its own region, not the node's" - ) - coords = {"lat": lat, "lng": lng, "location_type": location_type} + result = await _geocode_lookup(location_str, bounds) + if result is None: + return None + location_type = result.get("geometry", {}).get("location_type", "") + # Reject only APPROXIMATE — a region/city boundary centroid, which is + # what an ungeocodable string degrades to and is genuinely useless. + # + # ROOFTOP-only was too strict and emptied the map: dispatch names + # places the way people speak, and Google returns GEOMETRIC_CENTER for + # exactly those forms — intersections ("Lake Street and Veterans + # Memorial Drive") and named POIs ("Brewster Station"). Both are + # precise enough to plot and to proximity-match; requiring a street + # address threw away nearly every real dispatch location, leaving only + # numbered addresses geocoded. + if location_type not in ("ROOFTOP", "RANGE_INTERPOLATED", "GEOMETRIC_CENTER"): logger.info( - f"Geocoded '{location_str}' → {coords} " - f"({dist_km:.1f}km from {ref_label}) [{location_type}]" + f"Geocoding rejected '{location_str}' — imprecise result " + f"(location_type={location_type!r}), returning None" ) - return coords + return None + loc = result["geometry"]["location"] + lat, lng = float(loc["lat"]), float(loc["lng"]) + dist_km = _geo_dist_km(ref_lat, ref_lon, lat, lng) + if dist_km > max_km: + # server-26#159: the node-distance bound is a proxy for "is + # this plausible" that only makes sense when the node's own + # position is our best guess at the area — never when the + # query already names a different region on its own terms. + if not (ref_label == "node" and trust_named_region): + logger.warning( + f"Geocoding rejected '{location_str}' → ({lat:.4f}, {lng:.4f}) " + f"— {dist_km:.1f}km from {ref_label} exceeds {max_km:.1f}km" + ) + return None + logger.info( + f"Geocoding '{location_str}' → ({lat:.4f}, {lng:.4f}) is " + f"{dist_km:.1f}km from the receiving node, past {max_km:.1f}km — " + f"accepted anyway: the query names its own region, not the node's" + ) + coords = {"lat": lat, "lng": lng, "location_type": location_type} + logger.info( + f"Geocoded '{location_str}' → {coords} " + f"({dist_km:.1f}km from {ref_label}) [{location_type}]" + ) + return coords except Exception as e: logger.warning(f"Geocoding failed for '{location_str}': {e}") return None +# server-26#298: geocode answers, keyed by the exact request (address + +# bounds). Only the first result's geometry is kept — the minimum the checks +# in _geocode_location read — so the cache holds coordinates, not Maps +# content. The distance/precision checks still run per call on a hit, so a +# cached answer is judged against whatever anchor or node asks. +_geocode_cache: "OrderedDict[tuple[str, str], tuple[float, Optional[dict]]]" = OrderedDict() +geocode_cache_stats = {"hits": 0, "misses": 0} + + +def clear_geocode_cache() -> None: + _geocode_cache.clear() + geocode_cache_stats.update(hits=0, misses=0) + + +async def _geocode_lookup(address: str, bounds: str) -> Optional[dict]: + """ + The Geocoding API's first result for `address` inside `bounds`, or None + when it has none. Cached in process (settings.geocode_cache_*). A network + error or non-200 raises and is NOT cached — the next call retries. + """ + import time + import httpx + from app.config import settings + + key = (address.strip().lower(), bounds) + now = time.monotonic() + hit = _geocode_cache.get(key) + if hit is not None: + expires, value = hit + if expires > now: + _geocode_cache.move_to_end(key) + geocode_cache_stats["hits"] += 1 + return value + _geocode_cache.pop(key, None) + geocode_cache_stats["misses"] += 1 + + async with httpx.AsyncClient(timeout=5.0) as client: + r = await client.get( + "https://maps.googleapis.com/maps/api/geocode/json", + params={"address": address, "bounds": bounds, "region": "us", + "key": settings.google_maps_api_key}, + ) + r.raise_for_status() + data = r.json() + + status = data.get("status") + value: Optional[dict] = None + if status == "OK" and data.get("results"): + geometry = data["results"][0].get("geometry") or {} + value = {"geometry": {"location": geometry.get("location") or {}, + "location_type": geometry.get("location_type", "")}} + ttl = settings.geocode_cache_ttl_seconds + elif status == "ZERO_RESULTS": + ttl = settings.geocode_cache_negative_ttl_seconds + else: + # OVER_QUERY_LIMIT / REQUEST_DENIED / UNKNOWN_ERROR: a key or quota + # problem, not an answer about this address. + return None + if ttl > 0 and settings.geocode_cache_max_entries > 0: + _geocode_cache[key] = (now + ttl, value) + while len(_geocode_cache) > settings.geocode_cache_max_entries: + _geocode_cache.popitem(last=False) + return value + + def _municipality_from_tg(tg_name: Optional[str]) -> Optional[str]: """ Extract the municipality name from a talkgroup name. @@ -943,7 +1047,7 @@ def _sync_extract( ten_codes: Optional[dict[str, str]] = None, unit_format_hint: Optional[str] = None, ) -> list[dict]: - """Call GPT-4o-mini and return a list of scene dicts.""" + """Call the extraction model and return a list of scene dicts.""" from app.config import settings from app.internal.ai_clients import openai_client @@ -965,7 +1069,7 @@ def _sync_extract( try: client = openai_client() response = client.chat.completions.create( - model="gpt-4o-mini", + model=settings.extraction_model, messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, ) @@ -980,7 +1084,7 @@ def _sync_extract( return [raw] except json.JSONDecodeError as e: - logger.warning(f"GPT-4o-mini returned non-JSON: {e}") + logger.warning(f"Extraction model returned non-JSON: {e}") return [] # Any other exception is the API call itself failing (no credit, rate # limit, outage) and propagates to extract_scenes, which reports it to @@ -990,7 +1094,7 @@ def _sync_extract( def _sync_embed(text: str) -> Optional[list[float]]: - """Generate a text-embedding-3-small vector for semantic similarity.""" + """Generate an embedding (settings.embedding_model) for semantic similarity.""" from app.config import settings from app.internal.ai_clients import openai_client @@ -999,7 +1103,7 @@ def _sync_embed(text: str) -> Optional[list[float]]: try: client = openai_client() - result = client.embeddings.create(model="text-embedding-3-small", input=text) + result = client.embeddings.create(model=settings.embedding_model, input=text) return result.data[0].embedding except Exception as e: logger.warning(f"Embedding generation failed: {e}") diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index c446469..2f3b172 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/drb-c2-core/app/routers/upload.py @@ -501,7 +501,10 @@ async def _extract_and_correlate( if not scenes: _call_doc = await fstore.doc_get("calls", call_id) skip_reason = (_call_doc or {}).get("skip_reason") - if not skip_reason or skip_reason == "transcript_too_short": + # non_event_chatter (server-26#298) is the same population as + # transcript_too_short, just longer: housekeeping that skipped the + # paid extraction but may still be a same-talkgroup follow-up. + if not skip_reason or skip_reason in ("transcript_too_short", "non_event_chatter"): incident_id = await _correlate_with_consensus( call_id=call_id, node_id=node_id, @@ -512,6 +515,11 @@ async def _extract_and_correlate( incident_type=None, location=None, location_coords=None, + # server-26#104: the call doc's own severity, as + # recorrelation_sweep passes it. None resolved to "routine", + # which turned a reprocessed not-thin call into a thin one + # that attaches with no fit check. + severity=(_call_doc or {}).get("severity"), ) if incident_id: incident_ids.append(incident_id) diff --git a/drb-c2-core/tests/conftest.py b/drb-c2-core/tests/conftest.py index 05a31b1..58f7f1e 100644 --- a/drb-c2-core/tests/conftest.py +++ b/drb-c2-core/tests/conftest.py @@ -96,3 +96,13 @@ def _no_real_discord(monkeypatch): monkeypatch.setattr(discord_rest, "_transport", httpx.MockTransport(_offline)) yield discord_rest.reset() + + +@pytest.fixture(autouse=True) +def _fresh_geocode_cache(): + """The geocode cache (server-26#298) is process-wide; a test that stubs + Maps must never be answered by a previous test's stub.""" + from app.internal import intelligence + intelligence.clear_geocode_cache() + yield + intelligence.clear_geocode_cache() diff --git a/drb-c2-core/tests/test_chatter_gate.py b/drb-c2-core/tests/test_chatter_gate.py new file mode 100644 index 0000000..e76d0a8 --- /dev/null +++ b/drb-c2-core/tests/test_chatter_gate.py @@ -0,0 +1,194 @@ +""" +server-26#298 — the chatter pre-filter gates every paid call. + +Radio housekeeping (roll call, a bare "Post 4, 10-8, 10-4") used to go through +transcript correction, scene extraction, an embedding per scene and a geocode, +only to come back as a thin routine scene. Now it pays for none of them; the +free rule paths (clearance parse, self-initiated/plate backstop) still run. +""" +from unittest.mock import AsyncMock, patch + +import pytest + +from app.config import settings +from app.internal import intelligence, transcription +from app.internal.chatter_classifier import gate_chatter + + +@pytest.fixture(autouse=True) +def _gate_on(): + """The gate ships OFF (shadow) until a live window clears it (#127); + these tests exercise it switched on.""" + with patch.object(settings, "chatter_gate_enabled", True): + yield + + +class _NoPaidCalls: + """Fail the test if any paid extraction/embedding/geocode call happens.""" + + def __enter__(self): + def _boom(*a, **k): + raise AssertionError("a paid AI call was made for chatter") + + async def _aboom(*a, **k): + raise AssertionError("a paid geocode was made for chatter") + + self.ps = [ + patch.object(intelligence, "_sync_extract", _boom), + patch.object(intelligence, "_sync_embed", _boom), + patch.object(intelligence, "_geocode_lookup", _aboom), + ] + for p in self.ps: + p.start() + return self + + def __exit__(self, *a): + for p in self.ps: + p.stop() + + +async def _extract(transcript, talkgroup_name="PD Dispatch"): + saved: dict = {} + + async def _doc_set(coll, doc_id, data, merge=True): + saved.update(data) + + with patch.object(intelligence.fstore, "doc_get_cached", AsyncMock(return_value=None)), \ + patch.object(intelligence.fstore, "doc_set", _doc_set): + scenes = await intelligence.extract_scenes("call-1", transcript, talkgroup_name) + return scenes, saved + + +@pytest.mark.asyncio +async def test_bare_acknowledgement_over_five_words_skips_extraction(): + with _NoPaidCalls(): + scenes, saved = await _extract("10-4, 10-4 Central, 98. 10-4, 98.") + assert scenes == [] + assert saved["skip_reason"] == "non_event_chatter" + assert saved["severity"] == "routine" + assert saved["chatter_classifier_verdict"] is True + + +@pytest.mark.asyncio +async def test_whole_roll_call_skips_extraction(): + roll = ("Car 100, roll call. Henry 1. Henry 1. Sam 1. Sam 1. 45 Baker. 45 Baker. " + "11 Adam. Adam. 11 Baker. Baker. Staff 1. 1. Staff 2.") + with _NoPaidCalls(): + scenes, saved = await _extract(roll) + assert scenes == [] and saved["skip_reason"] == "non_event_chatter" + + +@pytest.mark.asyncio +async def test_roll_call_mentioned_on_the_way_to_a_real_job_still_extracts(): + """The #127 review's false-positive shape must still reach the model.""" + text = "Delayed for roll call, then respond to a structure fire at 12 Main Street" + assert gate_chatter(text) == (False, None) + with patch.object(intelligence, "_sync_extract", return_value=[]) as ext: + await _extract(text) + ext.assert_called_once() + + +@pytest.mark.asyncio +async def test_plate_read_chatter_still_becomes_a_stop_for_free(): + text = "Lincoln Charlie Robert 7-4-0-7, Lincoln Charlie Robert 7-4-0-7." + with _NoPaidCalls(): + scenes, saved = await _extract(text) + assert len(scenes) == 1 and scenes[0]["tags"] == ["traffic-stop"] + assert saved["severity"] == "minor" + + +@pytest.mark.asyncio +async def test_gate_off_restores_shadow_mode(): + with patch.object(settings, "chatter_gate_enabled", False), \ + patch.object(intelligence, "_sync_extract", return_value=[]) as ext: + _, saved = await _extract("10-4, 10-4 Central, 98. 10-4, 98.") + ext.assert_called_once() + assert saved["chatter_classifier_verdict"] is True, "the shadow verdict is still recorded" + + +@pytest.mark.asyncio +async def test_real_event_is_not_gated(): + with patch.object(intelligence, "_sync_extract", return_value=[]) as ext: + await _extract("Shots fired, officer down, requesting backup immediately at the school.") + ext.assert_called_once() + + +@pytest.mark.asyncio +async def test_transcription_skips_correction_for_chatter(): + flags = {"transcript_correction_enabled": True} + with patch.object(settings, "openai_api_key", "k"), \ + patch.object(settings, "groq_api_key", None), \ + patch.object(transcription, "_download", lambda uri: "/nonexistent.flac"), \ + patch.object(transcription, "_sync_transcribe_file", lambda *a, **k: { + "text": "Post 4, 10-8. 10-4. Copy, 98.", "segments": [], + "degenerate": False, "duration": 2.0, "confidence": None}), \ + patch.object(transcription.fstore, "doc_set", AsyncMock()), \ + patch.object(transcription, "ai_health") as health, \ + patch.object(transcription.transcript_correction, "correct", AsyncMock()) as correct, \ + patch("app.internal.feature_flags.resolve_flags", + AsyncMock(return_value=(flags, lambda k: flags.get(k, True)))): + health.report_healthy = AsyncMock() + text, _ = await transcription.transcribe_call("call-1", "gs://b/c.flac", "PD Dispatch") + assert text == "Post 4, 10-8. 10-4. Copy, 98." + correct.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_chatter_still_reaches_the_free_thin_correlation_path(): + """non_event_chatter joins transcript_too_short in upload.py's no-scene + fallback (rules-only for a thin call — no LLM).""" + from app.routers import upload + + with patch.object(upload, "fstore") as fs, \ + patch.object(upload, "_correlate_with_consensus", AsyncMock(return_value=None)) as corr: + fs.doc_get = AsyncMock(return_value={"skip_reason": "non_event_chatter", "severity": "routine"}) + fs.doc_set = AsyncMock() + await upload._extract_and_correlate( + call_id="c1", node_id="n1", system_id="s1", talkgroup_id=1, + talkgroup_name="PD Dispatch", transcript="10-4 Central, 98.", scenes=[], + ) + corr.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_no_scene_fallback_passes_the_call_doc_severity(): + """server-26#104: a reprocessed call whose doc says 'major' must not be + re-correlated as 'routine' (thin, no fit check).""" + from app.routers import upload + + with patch.object(upload, "fstore") as fs, \ + patch.object(upload, "_correlate_with_consensus", AsyncMock(return_value=None)) as corr: + fs.doc_get = AsyncMock(return_value={"severity": "major"}) + fs.doc_set = AsyncMock() + await upload._extract_and_correlate( + call_id="c1", node_id="n1", system_id="s1", talkgroup_id=1, + talkgroup_name="PD Dispatch", transcript="whatever", scenes=[], + ) + assert corr.await_args.kwargs["severity"] == "major" + + +@pytest.mark.asyncio +async def test_gate_off_still_corrects_chatter(): + """With the gate off, transcription must not skip correction either.""" + flags = {"transcript_correction_enabled": True} + with patch.object(settings, "chatter_gate_enabled", False), \ + patch.object(settings, "openai_api_key", "k"), \ + patch.object(settings, "groq_api_key", None), \ + patch.object(transcription, "_download", lambda uri: "/nonexistent.flac"), \ + patch.object(transcription, "_sync_transcribe_file", lambda *a, **k: { + "text": "Post 4, 10-8. 10-4. Copy, 98.", "segments": [], + "degenerate": False, "duration": 2.0, "confidence": None}), \ + patch.object(transcription.fstore, "doc_set", AsyncMock()), \ + patch.object(transcription, "ai_health") as health, \ + patch.object(transcription.transcript_correction, "correct", + AsyncMock(return_value=(None, None, False))) as correct, \ + patch("app.internal.feature_flags.resolve_flags", + AsyncMock(return_value=(flags, lambda k: flags.get(k, True)))): + health.report_healthy = AsyncMock() + await transcription.transcribe_call("call-1", "gs://b/c.flac", "PD Dispatch") + correct.assert_awaited_once() + + +def test_gate_ships_off(): + from app.config import Settings + assert Settings.model_fields["chatter_gate_enabled"].default is False diff --git a/drb-c2-core/tests/test_geocode_cache.py b/drb-c2-core/tests/test_geocode_cache.py new file mode 100644 index 0000000..5d0d95a --- /dev/null +++ b/drb-c2-core/tests/test_geocode_cache.py @@ -0,0 +1,162 @@ +""" +server-26#298 — geocode result cache in front of the Google Geocoding API, +and server-26#160 — place verification gets an anchor from the talkgroup's +own name when nobody configured one. +""" +from unittest.mock import AsyncMock, patch + +import pytest + +import app.internal.intelligence as intel +from app.config import settings +from app.internal import place_verifier as pv + +ANCHOR = {"lat": 41.16, "lng": -73.86, "radius_km": 10.0} + + +class _Counter: + def __init__(self, payload=None, exc=None): + self.calls = 0 + self.payload = payload + self.exc = exc + + def patch(self): + outer = self + + class _Resp: + def raise_for_status(self): + if outer.exc: + raise outer.exc + + def json(self): + return outer.payload + + class _Client: + async def __aenter__(self): return self + async def __aexit__(self, *a): return False + + async def get(self, *a, **k): + outer.calls += 1 + return _Resp() + + return patch("httpx.AsyncClient", lambda *a, **k: _Client()) + + +def _ok(lat=41.161, lng=-73.861): + return {"status": "OK", "results": [{"geometry": { + "location": {"lat": lat, "lng": lng}, "location_type": "ROOFTOP"}}]} + + +@pytest.fixture(autouse=True) +def _key(): + with patch.object(settings, "google_maps_api_key", "test-key"): + yield + + +@pytest.mark.asyncio +async def test_repeat_address_is_one_request(): + c = _Counter(_ok()) + with c.patch(): + a = await intel._geocode_location("12 Main Street, Ossining", anchor=ANCHOR) + b = await intel._geocode_location("12 main street, ossining ", anchor=ANCHOR) + assert a == b == {"lat": 41.161, "lng": -73.861, "location_type": "ROOFTOP"} + assert c.calls == 1 + assert intel.geocode_cache_stats == {"hits": 1, "misses": 1} + + +@pytest.mark.asyncio +async def test_cached_answer_is_still_judged_against_the_caller_anchor(): + c = _Counter(_ok()) + with c.patch(): + assert await intel._geocode_location("12 Main Street", anchor=ANCHOR) + far = {"lat": 0.0, "lng": 0.0, "radius_km": 10.0} + # Different bounds -> different key, so a separate request — never a + # cached answer leaking past a different area's distance check. + assert await intel._geocode_location("12 Main Street", anchor=far) is None + assert c.calls == 2 + + +@pytest.mark.asyncio +async def test_zero_results_is_cached_but_errors_are_not(): + c = _Counter({"status": "ZERO_RESULTS", "results": []}) + with c.patch(): + await intel._geocode_location("Cool Parts, Illinois", anchor=ANCHOR) + await intel._geocode_location("Cool Parts, Illinois", anchor=ANCHOR) + assert c.calls == 1 + + denied = _Counter({"status": "REQUEST_DENIED"}) + with denied.patch(): + await intel._geocode_location("Snowden Avenue", anchor=ANCHOR) + await intel._geocode_location("Snowden Avenue", anchor=ANCHOR) + assert denied.calls == 2, "a key/quota failure is not an answer about the address" + + boom = _Counter(_ok(), exc=RuntimeError("503")) + with boom.patch(): + await intel._geocode_location("Croton Point", anchor=ANCHOR) + await intel._geocode_location("Croton Point", anchor=ANCHOR) + assert boom.calls == 2 + + +@pytest.mark.asyncio +async def test_ttl_expiry_and_size_bound(): + c = _Counter(_ok()) + with c.patch(), patch.object(settings, "geocode_cache_ttl_seconds", 0): + await intel._geocode_location("12 Main Street", anchor=ANCHOR) + await intel._geocode_location("12 Main Street", anchor=ANCHOR) + assert c.calls == 2, "TTL 0 disables the cache" + + c = _Counter(_ok()) + with c.patch(), patch.object(settings, "geocode_cache_max_entries", 2): + for street in ("A Street", "B Street", "C Street"): + await intel._geocode_location(street, anchor=ANCHOR) + assert len(intel._geocode_cache) == 2 + + +# ── server-26#160 ─────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_unconfigured_channel_gets_an_anchor_from_its_talkgroup_name(): + resolved = {"center": {"lat": 40.71, "lng": -74.0}, "radius_km": 25.0} + seen = {} + + async def _geo(query, anchor): + seen["query"], seen["anchor"] = query, anchor + return {"lat": 40.7, "lng": -74.0} + + with patch.object(pv.settings, "place_verification_enabled", True), \ + patch.object(intel.fstore, "doc_get_cached", AsyncMock(return_value={"lat": 41.2, "lon": -73.8})), \ + patch.object(intel, "_get_node_state", AsyncMock(return_value="New York")), \ + patch.object(pv.area_context, "resolve_anchor", AsyncMock(return_value=resolved)) as res, \ + patch.object(pv, "_geocode_in_anchor", _geo): + await pv.verify("c1", "Respond to Lexington Avenue", None, ["Lexington Avenue"], + {}, {}, talkgroup_name="New York City PD", node_id="node-1") + res.assert_awaited_once_with({"municipality": "New York City", "state": "New York"}) + assert seen["anchor"] == {"lat": 40.71, "lng": -74.0, "radius_km": 25.0} + assert seen["query"] == "Lexington Avenue, New York City, New York" + + +@pytest.mark.asyncio +async def test_no_state_means_no_derived_anchor(): + """'Springfield PD' alone could anchor anywhere — skip, never guess.""" + with patch.object(pv.settings, "place_verification_enabled", True), \ + patch.object(intel.fstore, "doc_get_cached", AsyncMock(return_value={"lat": 41.2, "lon": -73.8})), \ + patch.object(intel, "_get_node_state", AsyncMock(return_value="")), \ + patch.object(pv, "_geocode_in_anchor") as geo: + out = await pv.verify("c1", "Respond to Main Street", None, ["Main Street"], + {}, {}, talkgroup_name="Springfield PD", node_id="node-1") + assert out == (None, None) + geo.assert_not_called() + + +@pytest.mark.asyncio +async def test_configured_place_without_anchor_still_skips(): + """A place that is set but too wide (no anchor stored) keeps the + load-bearing skip; the talkgroup-name fallback is only for unset areas.""" + with patch.object(pv.settings, "place_verification_enabled", True), \ + patch.object(pv, "_talkgroup_anchor", AsyncMock()) as derived, \ + patch.object(pv, "_geocode_in_anchor") as geo: + out = await pv.verify("c1", "Main Street", None, ["Main Street"], + {"state": "Colorado"}, {}, talkgroup_name="Denver PD", node_id="n") + assert out == (None, None) + derived.assert_not_awaited() + geo.assert_not_called() diff --git a/drb-c2-core/tests/test_intelligence_llm_bool.py b/drb-c2-core/tests/test_intelligence_llm_bool.py new file mode 100644 index 0000000..ddeed24 --- /dev/null +++ b/drb-c2-core/tests/test_intelligence_llm_bool.py @@ -0,0 +1,14 @@ +"""server-26#88 — resolved/reassignment are parsed, not bool()-coerced.""" +import pytest + +from app.internal.intelligence import _llm_bool + + +@pytest.mark.parametrize("value,expected", [ + (True, True), (False, False), + ("true", True), ("TRUE", True), (" Yes ", True), ("yes", True), + ("false", False), ("False", False), ("no", False), ("NO", False), + ("", False), ("maybe", False), (None, False), (1, False), (0, False), ([], False), +]) +def test_llm_bool(value, expected): + assert _llm_bool(value) is expected From ff0a89e9f1a8a460c3cfcccc7bff75ebada45844 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:29:01 -0400 Subject: [PATCH 4/5] Alerts: confidence gate before any push, held alerts, release on corroboration (server-26#300) No confidence field exists anywhere in scene/incident data, so the alerter derives one: match weight (talkgroup 1.0, extracted tag 0.8, raw transcript 0.6) scaled down only when stt_confidence is under ALERT_STT_REFERENCE (0.3), zero for not_speech. Below ALERT_MIN_CONFIDENCE, or high severity (term list, or a major incident) without ALERT_CORROBORATION_MIN_CALLS calls or ALERT_HIGH_SEVERITY_MIN_CONFIDENCE, the match is written to alert_holds (no Firestore rule, admin-only GET /alerts/held) and logged instead of pushed. A needs_corroboration hold is released when a later call brings its incident to the call minimum; a rule pushed by that call supersedes its hold. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/config.py | 24 +++ drb-c2-core/app/internal/alerter.py | 209 ++++++++++++++++++++++++++- drb-c2-core/app/routers/alerts.py | 21 +++ drb-c2-core/tests/test_alert_gate.py | 168 +++++++++++++++++++++ 4 files changed, 421 insertions(+), 1 deletion(-) create mode 100644 drb-c2-core/tests/test_alert_gate.py diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 66ff4fb..652872f 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -163,6 +163,30 @@ class Settings(BaseSettings): # else's PII. Both gates must be open before any snippet leaves. alert_transcript_snippet_enabled: bool = False place_verify_max_per_call: int = 3 + # ── Alert confidence gate (server-26#300) ───────────────────────────────── + # Nothing in the pipeline emits a confidence, so the alerter derives one: + # how the rule matched (talkgroup 1.0 / extracted tag 0.8 / raw transcript + # text only 0.5 -- the CrimeRadar "fire alarm" -> "firearm" failure was a + # transcript match) scaled down when Whisper's own per-token probability + # (calls.stt_confidence) is below alert_stt_reference. Below the floor -> + # held, never pushed. High-severity matches also need the incident to + # have alert_corroboration_min_calls calls, or a confidence at or above + # alert_high_severity_min_confidence; held ones are released when a + # later call corroborates the incident. Starting values, not measured. + alert_min_confidence: float = 0.5 + alert_high_severity_min_confidence: float = 0.85 + alert_corroboration_min_calls: int = 2 + # Low on purpose: exp(avg_logprob) on vocoded radio audio is often well + # under 0.5 for speech that is right, and Groq/OpenAI logprobs are not on + # a shared scale. Only clearly garbled audio should pull the score down. + alert_stt_reference: float = 0.3 + alert_high_severity_terms: list[str] = [ + "shots-fired", "shots fired", "shooting", "shot", "gun", "firearm", "weapon", + "stabbing", "stabbed", "officer-down", "officer down", "10-13", + "officer-needs-assistance", "active-shooter", "hostage", "barricade", + "explosion", "structure-fire", "structure fire", "working-fire", + "entrapment", "cardiac-arrest", "vehicle-pursuit", "pursuit", + ] # How close a candidate has to sound before it may rewrite a transcript. # Below this, Places Text Search will confidently hand back the nearest # business for any garbage string. diff --git a/drb-c2-core/app/internal/alerter.py b/drb-c2-core/app/internal/alerter.py index d4ca3cb..c4c6ab7 100644 --- a/drb-c2-core/app/internal/alerter.py +++ b/drb-c2-core/app/internal/alerter.py @@ -2,9 +2,19 @@ Alert dispatch engine. Loads enabled alert rules from Firestore and checks each one against the call's -talkgroup ID, tags, and transcript. On a match: +talkgroup ID, tags, and transcript. On a match that passes the confidence +gate (server-26#300, `_assess`): 1. Creates an AlertEvent document in Firestore. 2. Optionally POSTs a Discord webhook message if the rule has one configured. +A match that fails the gate is written to `alert_holds` instead -- never +pushed, readable by admins (GET /alerts/held) -- and a hold that was waiting +only on corroboration is released when a later call brings its incident to +alert_corroboration_min_calls calls. + +WHY A GATE: CrimeRadar pushed "shots fired at an elementary school" after STT +heard "fire alarm" as "firearm". A keyword hit in raw transcript text is the +weakest evidence this pipeline has, and a push alert is the one surface that +cannot be taken back. Raw transcript text is withheld from both by default -- see _snippet_allowed and server-26#85. @@ -54,6 +64,8 @@ async def check_and_dispatch( # Loop-invariant: every rule here belongs to the same org, so the opt-in is # resolved once rather than per match. snippet_allowed = await _snippet_allowed(org_id) + evidence = await _evidence(call_doc or {}) + pushed_rules: set[str] = set() for rule in rules: matched_keywords = _match_rule(rule, talkgroup_id, tags, transcript) @@ -63,6 +75,12 @@ async def check_and_dispatch( alert_id = str(uuid.uuid4()) snippet = _snippet(transcript) if snippet_allowed else None now = datetime.now(timezone.utc).isoformat() + verdict = _assess(matched_keywords, tags, evidence) + if not verdict["push"]: + await _hold(alert_id, rule, org_id, call_id, node_id, talkgroup_id, + talkgroup_name, matched_keywords, snippet, now, verdict, evidence) + continue + pushed_rules.add(rule.get("rule_id", "")) event = { "alert_id": alert_id, "org_id": org_id, @@ -76,6 +94,9 @@ async def check_and_dispatch( "transcript_snippet": snippet, "triggered_at": now, "acknowledged": False, + "confidence": verdict["confidence"], + "high_severity": verdict["high_severity"], + "corroborating_calls": evidence["max_calls"], } try: @@ -92,6 +113,192 @@ async def check_and_dispatch( if webhook_url: await _post_webhook(webhook_url, rule.get("name", ""), talkgroup_name, matched_keywords, snippet) + await _release_corroborated(rules, evidence, call_id, pushed_rules) + + +# ── Confidence gate (server-26#300) ───────────────────────────────────────── + +# How a rule matched, as evidence. Talkgroup activity is metadata, not +# recognised speech, so it cannot be misheard. +_W_TALKGROUP = 1.0 +_W_TAG = 0.8 # the extraction model put it in tags +_W_TRANSCRIPT = 0.6 # substring of raw STT text only (above the 0.5 default floor) + + +def _norm_term(term: str) -> str: + return (term or "").strip().lower().replace("-", " ") + + +def _is_high_severity_term(term: str) -> bool: + wanted = {_norm_term(t) for t in settings.alert_high_severity_terms} + return _norm_term(term) in wanted + + +async def _evidence(call_doc: dict) -> dict: + """What the call doc and its incidents say about how much to trust it.""" + incident_ids = list(call_doc.get("incident_ids") or []) + max_calls = 0 + max_severity = call_doc.get("severity") + severity_rank = {"routine": 0, "minor": 1, "moderate": 2, "major": 3} + for iid in incident_ids: + try: + inc = await fstore.doc_get("incidents", iid) or {} + except Exception as e: + logger.warning(f"Alerter: could not read incident {iid}: {e}") + continue + max_calls = max(max_calls, len(inc.get("call_ids") or [])) + if severity_rank.get(inc.get("severity"), -1) > severity_rank.get(max_severity, -1): + max_severity = inc.get("severity") + stt = call_doc.get("stt_confidence") + return { + "incident_ids": incident_ids, + "max_calls": max_calls, + "severity": max_severity, + "stt_confidence": stt if isinstance(stt, (int, float)) else None, + "not_speech": bool(call_doc.get("transcript_not_speech")), + } + + +def _assess(matched: list[str], tags: list[str], evidence: dict) -> dict: + """ + Push or hold, and why. Pure, so the gate is testable without Firestore. + + confidence = match weight x STT factor. The STT factor is 1.0 unless + Whisper's own mean token probability is under alert_stt_reference, and + 1.0 when there is none (a json-only model) rather than guessing. + """ + content = [m for m in matched if not m.startswith("talkgroup:")] + if not content: + weight = _W_TALKGROUP + elif any(m in (tags or []) for m in content): + weight = _W_TAG + else: + weight = _W_TRANSCRIPT + + stt = evidence.get("stt_confidence") + stt_factor = 1.0 + if content and stt is not None and settings.alert_stt_reference > 0: + stt_factor = min(1.0, max(0.0, stt / settings.alert_stt_reference)) + if content and evidence.get("not_speech"): + stt_factor = 0.0 # the corrector called this recogniser noise + confidence = round(weight * stt_factor, 3) + + high = bool(content) and ( + any(_is_high_severity_term(m) for m in content) + or any(_is_high_severity_term(t) for t in (tags or [])) + or evidence.get("severity") == "major" + ) + corroborated = evidence.get("max_calls", 0) >= settings.alert_corroboration_min_calls + + if confidence < settings.alert_min_confidence: + return {"push": False, "confidence": confidence, "high_severity": high, + "hold_reason": "low_confidence"} + if high and not corroborated and confidence < settings.alert_high_severity_min_confidence: + return {"push": False, "confidence": confidence, "high_severity": high, + "hold_reason": "needs_corroboration"} + return {"push": True, "confidence": confidence, "high_severity": high, "hold_reason": None} + + +async def _hold(alert_id, rule, org_id, call_id, node_id, talkgroup_id, talkgroup_name, + matched_keywords, snippet, now, verdict, evidence) -> None: + """Record a gated match where admins can see it, and push nothing.""" + hold = { + "alert_id": alert_id, + "org_id": org_id, + "rule_id": rule.get("rule_id", ""), + "rule_name": rule.get("name", ""), + "call_id": call_id, + "node_id": node_id, + "talkgroup_id": talkgroup_id, + "talkgroup_name": talkgroup_name or "", + "matched_keywords": matched_keywords, + "transcript_snippet": snippet, + "triggered_at": now, + "status": "held", + "hold_reason": verdict["hold_reason"], + "confidence": verdict["confidence"], + "high_severity": verdict["high_severity"], + # One incident to wait on: the first the call linked to. A call that + # linked to none can never be corroborated, so it stays held. + "incident_id": (evidence["incident_ids"] or [None])[0], + "corroborating_calls": evidence["max_calls"], + } + try: + await fstore.doc_set("alert_holds", alert_id, hold, merge=False) + except Exception as e: + logger.warning(f"Alerter: could not save held alert: {e}") + return + logger.info( + f"Alert HELD ({verdict['hold_reason']}, confidence={verdict['confidence']}, " + f"high_severity={verdict['high_severity']}, calls={evidence['max_calls']}): " + f"rule='{rule.get('name')}' call={call_id} keywords={matched_keywords}" + ) + + +async def _release_corroborated(rules: list[dict], evidence: dict, call_id: str, + pushed_rules: set[str]) -> None: + """ + Push holds that were waiting only on corroboration once their incident + has enough calls. A low-confidence hold is never released this way: a + second call on the incident says the event is real, not that the first + call was heard right. A rule this call already pushed for supersedes its + hold instead of alerting twice. + """ + if evidence["max_calls"] < settings.alert_corroboration_min_calls: + return + by_id = {r.get("rule_id", ""): r for r in rules} + for iid in evidence["incident_ids"]: + try: + holds = await fstore.collection_list("alert_holds", incident_id=iid, status="held") + except Exception as e: + logger.warning(f"Alerter: could not read holds for incident {iid}: {e}") + continue + for hold in holds: + if hold.get("hold_reason") != "needs_corroboration": + continue + rule = by_id.get(hold.get("rule_id", "")) + now = datetime.now(timezone.utc).isoformat() + if rule is None or hold.get("rule_id", "") in pushed_rules: + status = "superseded" if rule is not None else "rule_gone" + await _set_hold_status(hold, status, call_id, now) + continue + event = {k: hold.get(k) for k in ( + "alert_id", "org_id", "rule_id", "rule_name", "call_id", "node_id", + "talkgroup_id", "talkgroup_name", "matched_keywords", + "transcript_snippet", "confidence", "high_severity", + )} + event.update({ + "triggered_at": now, + "acknowledged": False, + "released_from_hold": True, + "corroborated_by_call_id": call_id, + "corroborating_calls": evidence["max_calls"], + }) + try: + await fstore.doc_set("alert_events", hold["alert_id"], event, merge=False) + except Exception as e: + logger.warning(f"Alerter: could not release held alert {hold.get('alert_id')}: {e}") + continue + await _set_hold_status(hold, "released", call_id, now) + pushed_rules.add(hold.get("rule_id", "")) + logger.info( + f"Alert released from hold: rule='{hold.get('rule_name')}' incident={iid} " + f"corroborated by call {call_id} ({evidence['max_calls']} calls)" + ) + webhook_url = rule.get("discord_webhook") + if webhook_url: + await _post_webhook(webhook_url, rule.get("name", ""), hold.get("talkgroup_name"), + hold.get("matched_keywords") or [], hold.get("transcript_snippet")) + + +async def _set_hold_status(hold: dict, status: str, call_id: str, now: str) -> None: + try: + await fstore.doc_update("alert_holds", hold["alert_id"], { + "status": status, "resolved_at": now, "resolved_by_call_id": call_id, + }) + except Exception as e: + logger.warning(f"Alerter: could not update hold {hold.get('alert_id')}: {e}") + async def _snippet_allowed(org_id: Optional[str]) -> bool: """ diff --git a/drb-c2-core/app/routers/alerts.py b/drb-c2-core/app/routers/alerts.py index ce3e769..327b484 100644 --- a/drb-c2-core/app/routers/alerts.py +++ b/drb-c2-core/app/routers/alerts.py @@ -28,6 +28,27 @@ async def list_alerts( return await fstore.collection_list("alert_events", **filters) +@router.get("/alerts/held") +async def list_held_alerts( + status: Optional[str] = Query("held", description="held | released | superseded | rule_gone; empty = all"), + org_id: Optional[str] = Query(None, description="Platform-admin only — omit for every org."), + _: dict = Depends(require_admin_token), +): + """ + Rule matches the confidence gate kept from being pushed (server-26#300). + Admin-only: a held match is by definition one we do not trust enough to + show, so it does not belong on any customer surface (and alert_holds has + no Firestore rule, so the browser cannot read it directly either). + """ + filters = {} + if status: + filters["status"] = status + if org_id: + filters["org_id"] = org_id + holds = await fstore.collection_list("alert_holds", **filters) + return sorted(holds, key=lambda h: h.get("triggered_at") or "", reverse=True) + + @router.post("/alerts/{alert_id}/acknowledge") async def acknowledge_alert(alert_id: str, decoded: dict = Depends(require_service_or_firebase_token)): doc = await fstore.doc_get("alert_events", alert_id) diff --git a/drb-c2-core/tests/test_alert_gate.py b/drb-c2-core/tests/test_alert_gate.py new file mode 100644 index 0000000..49f4a86 --- /dev/null +++ b/drb-c2-core/tests/test_alert_gate.py @@ -0,0 +1,168 @@ +""" +server-26#300 — confidence gate before any push alert. + +CrimeRadar pushed "shots fired at an elementary school" after STT heard "fire +alarm" as "firearm". A single uncorroborated transcript hit on a +high-severity term must be held, not pushed; a corroborated incident releases +it; low-confidence matches never push. +""" +import pytest +from unittest.mock import AsyncMock, patch + +from app.internal import alerter +from tests.fake_firestore import FakeFirestore + + +def _ev(**over): + base = {"incident_ids": [], "max_calls": 0, "severity": None, + "stt_confidence": None, "not_speech": False} + base.update(over) + return base + + +# ── the pure decision ─────────────────────────────────────────────────────── + +def test_talkgroup_only_rule_always_pushes(): + v = alerter._assess(["talkgroup:9048"], [], _ev(stt_confidence=0.1)) + assert v["push"] and v["confidence"] == 1.0 + + +def test_transcript_only_firearm_single_call_is_held(): + v = alerter._assess(["firearm"], [], _ev()) + assert not v["push"] and v["hold_reason"] == "needs_corroboration" + assert v["high_severity"] is True + + +def test_tagged_shots_fired_still_needs_corroboration(): + v = alerter._assess(["shots-fired"], ["shots-fired"], _ev(max_calls=1)) + assert not v["push"] and v["confidence"] == 0.8 + + +def test_corroborated_high_severity_pushes(): + v = alerter._assess(["shots-fired"], ["shots-fired"], _ev(max_calls=2)) + assert v["push"] + + +def test_major_incident_makes_any_content_match_high_severity(): + v = alerter._assess(["school"], [], _ev(severity="major")) + assert v["high_severity"] and not v["push"] + + +def test_ordinary_tag_match_pushes(): + v = alerter._assess(["mva"], ["mva"], _ev(max_calls=1)) + assert v["push"] and not v["high_severity"] + + +def test_low_stt_confidence_holds_even_ordinary_matches(): + v = alerter._assess(["mva"], ["mva"], _ev(stt_confidence=0.1)) + assert not v["push"] and v["hold_reason"] == "low_confidence" + assert v["confidence"] == round(0.8 * 0.1 / 0.3, 3) + + +def test_typical_radio_logprob_does_not_hold_a_transcript_match(): + """exp(avg_logprob) of ~0.4 is normal on vocoded audio (review of #300).""" + v = alerter._assess(["mva"], [], _ev(stt_confidence=0.4)) + assert v["push"] + + +def test_not_speech_never_pushes_content(): + v = alerter._assess(["mva"], ["mva"], _ev(not_speech=True)) + assert not v["push"] + + +# ── end to end through Firestore ──────────────────────────────────────────── + +RULE = {"rule_id": "r1", "name": "Weapons", "enabled": True, "org_id": "org-1", + "keywords": ["firearm", "shots-fired"], "discord_webhook": "https://discord.example/wh"} + + +@pytest.fixture +def fs(monkeypatch): + fake = FakeFirestore.install(monkeypatch) + fake.put("alert_rules", "r1", dict(RULE)) + return fake + + +def _posts(): + posted = [] + + class _Client: + async def __aenter__(self): return self + async def __aexit__(self, *a): return False + + async def post(self, url, json=None, **k): + posted.append(json) + + return posted, patch("httpx.AsyncClient", lambda *a, **k: _Client()) + + +async def _dispatch(call_id, tags, transcript): + await alerter.check_and_dispatch(call_id=call_id, node_id="n1", talkgroup_id=1, + talkgroup_name="PD Dispatch", tags=tags, transcript=transcript) + + +@pytest.mark.asyncio +async def test_crimeradar_case_is_held_then_released_by_corroboration(fs): + fs.put("calls", "c1", {"org_id": "org-1", "incident_ids": ["inc-1"]}) + fs.put("incidents", "inc-1", {"call_ids": ["c1"], "severity": "moderate"}) + posted, http = _posts() + with http: + await _dispatch("c1", [], "report of a firearm at the elementary school") + assert posted == [], "a single uncorroborated firearm hit must not push" + assert fs.data.get("alert_events", {}) == {} + (hold,) = fs.data["alert_holds"].values() + assert hold["status"] == "held" and hold["hold_reason"] == "needs_corroboration" + assert hold["incident_id"] == "inc-1" + + # A second call joins the incident; it does not itself mention a firearm. + fs.put("calls", "c2", {"org_id": "org-1", "incident_ids": ["inc-1"]}) + fs.put("incidents", "inc-1", {"call_ids": ["c1", "c2"], "severity": "moderate"}) + with http: + await _dispatch("c2", [], "units on scene, stand by") + assert len(posted) == 1 and "Weapons" in posted[0]["content"] + (event,) = fs.data["alert_events"].values() + assert event["released_from_hold"] is True and event["corroborated_by_call_id"] == "c2" + assert fs.get("alert_holds", hold["alert_id"])["status"] == "released" + + +@pytest.mark.asyncio +async def test_low_confidence_hold_is_not_released_by_corroboration(fs): + fs.put("calls", "c1", {"org_id": "org-1", "incident_ids": ["inc-1"], "stt_confidence": 0.1}) + fs.put("incidents", "inc-1", {"call_ids": ["c1"]}) + posted, http = _posts() + with http: + await _dispatch("c1", [], "firearm") + (hold,) = fs.data["alert_holds"].values() + assert hold["hold_reason"] == "low_confidence" + + fs.put("calls", "c2", {"org_id": "org-1", "incident_ids": ["inc-1"]}) + fs.put("incidents", "inc-1", {"call_ids": ["c1", "c2"]}) + with http: + await _dispatch("c2", [], "units on scene") + assert posted == [] + assert fs.get("alert_holds", hold["alert_id"])["status"] == "held" + + +@pytest.mark.asyncio +async def test_rule_pushed_now_supersedes_its_hold_instead_of_alerting_twice(fs): + fs.put("calls", "c1", {"org_id": "org-1", "incident_ids": ["inc-1"]}) + fs.put("incidents", "inc-1", {"call_ids": ["c1"]}) + posted, http = _posts() + with http: + await _dispatch("c1", ["shots-fired"], "shots fired") + fs.put("calls", "c2", {"org_id": "org-1", "incident_ids": ["inc-1"]}) + fs.put("incidents", "inc-1", {"call_ids": ["c1", "c2"]}) + with http: + await _dispatch("c2", ["shots-fired"], "more shots fired") + assert len(posted) == 1 + (hold,) = fs.data["alert_holds"].values() + assert hold["status"] == "superseded" + + +@pytest.mark.asyncio +async def test_held_alerts_are_listed_for_admins(fs): + from app.routers import alerts as alerts_router + fs.put("alert_holds", "h1", {"alert_id": "h1", "status": "held", "triggered_at": "2026-10-27T01:00:00"}) + fs.put("alert_holds", "h2", {"alert_id": "h2", "status": "released", "triggered_at": "2026-10-27T02:00:00"}) + out = await alerts_router.list_held_alerts(status="held", org_id=None, _={}) + assert [h["alert_id"] for h in out] == ["h1"] From 90a3c27a24aa21110303704be16325c7c0bfdcbd Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:29:10 -0400 Subject: [PATCH 5/5] Docs: document the STT provider, extraction and alert-gate env vars (server-26#297, server-26#298, server-26#300) Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/.env.example | 43 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 42 insertions(+), 1 deletion(-) diff --git a/drb-c2-core/.env.example b/drb-c2-core/.env.example index 80f3996..352dd28 100644 --- a/drb-c2-core/.env.example +++ b/drb-c2-core/.env.example @@ -36,8 +36,49 @@ NWS_USER_AGENT=DRB (you@example.com) # Comma list of NWS state/marine area codes served NWS_AREA=NY -# OpenAI — for transcription (Whisper), intelligence extraction, embeddings, and summaries +# OpenAI — transcription (whisper-1, and the STT fallback), extraction, embeddings OPENAI_API_KEY= +# OpenAI transcription model: whisper-1 | gpt-4o-mini-transcribe | gpt-4o-transcribe +STT_MODEL=whisper-1 + +# STT provider (server-26#297). Any OpenAI-compatible /audio/transcriptions API. +# auto (default): groq if GROQ_API_KEY is set, compat if STT_BASE_URL is set, +# otherwise openai — so leaving these blank keeps today's OpenAI behaviour. +STT_PROVIDER=auto +# Groq: whisper-large-v3-turbo, $0.04/audio-hour (min 10s billed per request). +GROQ_API_KEY= +# Model on the non-OpenAI provider. +STT_COMPAT_MODEL=whisper-large-v3-turbo +# Override the endpoint/key (blank = provider default: https://api.groq.com/openai/v1 +# and GROQ_API_KEY). Set both for a self-hosted OpenAI-compatible server. +STT_BASE_URL= +STT_API_KEY= +# Retry on OpenAI STT_MODEL when the provider above fails (needs OPENAI_API_KEY). +STT_FALLBACK_TO_OPENAI=true +# Area places / local names / vocabulary in the STT prompt (non-OpenAI provider only). +STT_VOCAB_PROMPT_ENABLED=true +# Estimated prompt token budget (Whisper keeps only the last 224). +STT_PROMPT_MAX_TOKENS=200 + +# Scene extraction / embeddings models (server-26#298) +EXTRACTION_MODEL=gpt-4o-mini +EMBEDDING_MODEL=text-embedding-3-small +# Skip correction + extraction for radio housekeeping (roll call, bare 10-4s). +# Default false = shadow mode (verdict recorded, nothing skipped). Turn on only +# after a shadow window shows no real events flagged (server-26#127). +CHATTER_GATE_ENABLED=false +# In-process geocode cache (Maps terms allow caching lat/lng <= 30 days) +GEOCODE_CACHE_TTL_SECONDS=86400 +GEOCODE_CACHE_NEGATIVE_TTL_SECONDS=3600 +GEOCODE_CACHE_MAX_ENTRIES=5000 + +# Push-alert confidence gate (server-26#300). Below the floor -> held (admin-only +# GET /alerts/held). High-severity matches also need N calls on the incident or +# the high confidence bar. +ALERT_MIN_CONFIDENCE=0.5 +ALERT_HIGH_SEVERITY_MIN_CONFIDENCE=0.85 +ALERT_CORROBORATION_MIN_CALLS=2 +ALERT_STT_REFERENCE=0.3 SUMMARY_INTERVAL_MINUTES=15 CORRELATION_WINDOW_HOURS=4 EMBEDDING_SIMILARITY_THRESHOLD=0.82