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