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 <noreply@anthropic.com>

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

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
+    # acknowledgements, unit check-ins) and records it on the call doc. The
-    # doc, but does NOT skip extraction anywhere below — every path runs
+    # live cost gate (server-26#298, gate_chatter below) is a stricter view of
-    # exactly as it did before this landed. Deliberately ahead of the ≤5-word
+    # 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
+    # Residual risk of the live gate, unchanged from the #127 review: a
-    # write skip_reason="non_event_chatter" instead of just recording the
+    # digit-hyphen street address ("72-Holland") alone in a transmission reads
-    # verdict) once a live shadow-mode window confirms 0 false positives on
+    # as a unit ID. CHATTER_GATE_ENABLED=false restores shadow mode.
-    # 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.
     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))
+        # server-26#88: bool("false") is True, which let a model answering
-        reassignment:       bool           = bool(scene.get("reassignment", False))
+        # 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:
+        result = await _geocode_lookup(location_str, bounds)
-            r = await client.get(
+        if result is None:
-                "https://maps.googleapis.com/maps/api/geocode/json",
+            return None
-                params=params,
+        location_type = result.get("geometry", {}).get("location_type", "")
-            )
+        # Reject only APPROXIMATE — a region/city boundary centroid, which is
-            r.raise_for_status()
+        # what an ungeocodable string degrades to and is genuinely useless.
-            data = r.json()
+        #
-            if data.get("status") != "OK" or not data.get("results"):
+        # ROOFTOP-only was too strict and emptied the map: dispatch names
-                return None
+        # places the way people speak, and Google returns GEOMETRIC_CENTER for
-            result = data["results"][0]
+        # exactly those forms — intersections ("Lake Street and Veterans
-            location_type = result.get("geometry", {}).get("location_type", "")
+        # Memorial Drive") and named POIs ("Brewster Station"). Both are
-            # Reject only APPROXIMATE — a region/city boundary centroid, which is
+        # precise enough to plot and to proximity-match; requiring a street
-            # what an ungeocodable string degrades to and is genuinely useless.
+        # address threw away nearly every real dispatch location, leaving only
-            #
+        # numbered addresses geocoded.
-            # ROOFTOP-only was too strict and emptied the map: dispatch names
+        if location_type not in ("ROOFTOP", "RANGE_INTERPOLATED", "GEOMETRIC_CENTER"):
-            # 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}
             logger.info(
-                f"Geocoded '{location_str}' → {coords} "
+                f"Geocoding rejected '{location_str}' — imprecise result "
-                f"({dist_km:.1f}km from {ref_label}) [{location_type}]"
+                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}")

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)

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()

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

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()

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