AI providers: review follow-ups to 0060ccd

Replay usage key gemini_usage -> llm_usage (backend + ReplayTab), re-price the
correction/correlation estimates for gpt-4o-mini, no temperature for o-series
or gpt-5 models, and refresh comments still naming Gemini.

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

drb-c2-core/.env.example

@@ -36,8 +36,52 @@ 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=
+# Since 2026-11-08 OpenAI also serves the correlation decision and transcript
+# correction (CORR_CHEAP_MODEL / TRANSCRIPT_CORRECTION_MODEL, default gpt-4o-mini).
+# GEMINI_API_KEY is only needed if you point either of those at a gemini-* model.
+# 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

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
@@ -48,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
@@ -68,27 +122,23 @@ class Settings(BaseSettings):
     # NWS alerts cache TTL; the issue asks 60-120 s (floored at 30 s in code).
     weather_alerts_ttl_s: int = 90
 
-    # Gemini (intelligence extraction, embeddings, incident summaries)
+    # Gemini: only used if CORR_CHEAP_MODEL / TRANSCRIPT_CORRECTION_MODEL is set
+    # to a gemini-* ID (app/internal/llm.py). OpenAI serves both by default.
     gemini_api_key: Optional[str] = None
-    # Correlation consensus models
+    # Correlation LLM (server-26#299). The rules engine decides every confident
-    # corr_cheap_model  — first-pass LLM correlator (runs on every call)
+    # case; this model is called at most ONCE per scene, and only when the rules
-    # corr_smart_model  — tiebreaker (only fires when rules and cheap LLM disagree)
+    # flag the decision as ambiguous (incident_correlator._ambiguity). There is
-    # Both IDs below were retired by Google and returned 404 on every call from
+    # no second "smart" tier any more: the cheap/smart consensus tiebreak ran on
-    # some point before 2026-08-18 until they were corrected. Because a failed
+    # ~15% of scenes and resolved nearly every disagreement as "new" (#115).
-    # LLM call falls back to the rules decision, nothing broke loudly -- the
+    # A model ID was once retired by its provider and 404'd on every call for
-    # entire LLM tier and the consensus tiebreak were simply dead in production
+    # days; a failed LLM call falls back to the rules decision, so nothing broke
-    # while correlation behaviour was being tuned against rules-only output.
+    # loudly. Verify the ID against the provider's model list before changing.
-    # Verify against https://ai.google.dev/gemini-api/docs/models before changing.
+    corr_cheap_model: str = "gpt-4o-mini"   # OpenAI since 2026-11-08 (was gemini-3.6-flash); a gemini-* ID still works
-    corr_cheap_model: str = "gemini-3.6-flash"   # was gemini-2.0-flash (shut down)
-    # gemini-2.5-pro was closed to new projects by 2026-09 (every tiebreak 404'd
-    # in the first replay run, server-26#170); Google lists no stable Pro model,
-    # so the smart tier is the newest stable Flash instead.
-    corr_smart_model: str = "gemini-3.8-flash"   # was gemini-2.5-pro, gemini-1.5-pro
     # Transcript correction (server-26#36). Runs inside transcription, once per
     # transcribed call above MIN_WORDS_FOR_CORRECTION, so it is priced like STT
     # rather than like the correlation tier — cheap model on purpose.
     transcript_correction_enabled: bool = True
-    transcript_correction_model: str = "gemini-3.6-flash"
+    transcript_correction_model: str = "gpt-4o-mini"  # OpenAI since 2026-11-08; a gemini-* ID still works
     # Retry Whisper once when its output is degenerate. The same clip produced a
     # 56-word ten-code counting run on one attempt and real speech on the next
     # (2026-08-23, call e49ea32c), so a hallucination is a coin-flip rather than
@@ -109,6 +159,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.
@@ -122,6 +196,39 @@ class Settings(BaseSettings):
     embedding_similarity_threshold: float = 0.93  # slow-path: requires location corroboration
     embedding_no_location_threshold: float = 0.97 # slow-path: match without location (very high bar)
     embedding_cross_tg_threshold: float = 0.85    # cross-TG path: same dept + 2+ shared units
+    # server-26#86: the content-divergence veto on a unit-overlap link. This is
+    # a DIVERGENCE bar, not a match bar, so it sits deliberately below
+    # embedding_similarity_threshold: two calls about different jobs on one
+    # dispatch channel still share ten-codes, unit IDs and address phrasing, so
+    # "not a match" (< 0.93) is not evidence of "a different job". Below this
+    # bar the link is vetoed; between the two bars the link is neither
+    # confirmed nor refused, so it is AMBIGUOUS and goes to the LLM (#299)
+    # instead of silently merging. 0.82 is the historical literal, unmeasured
+    # against real score distributions.
+    embedding_divergence_threshold: float = 0.82
+    # A unit-overlap link is only a confident one while the incident is fresh:
+    # idle under this many minutes (the old hard-coded 15 of the divergence
+    # veto) AND open under unit_link_confident_max_span_minutes. Past either,
+    # shared units alone are mechanism B (CORRELATION_REVIEW_0820.md) — patrol
+    # units work job after job on one channel — so the link is ambiguous.
+    unit_link_fresh_idle_minutes: int = 15
+    unit_link_confident_max_span_minutes: int = 30
+    # server-26#189: an incident open this long accepts no more WEAK links
+    # (unit overlap without location/content agreement, or an LLM link) — only
+    # vehicle, location or embedding evidence, until the hard
+    # incident_max_duration_minutes cap. The 09-27 "Rollover Entrapment"
+    # sink absorbed unrelated Ch 1 jobs for 2h34m this way. Thin replies are
+    # exempt: they never refresh updated_at, so they cannot keep a sink alive.
+    incident_weak_link_max_span_minutes: int = 90
+    # server-26#189: an LLM link may not pull a call geocoded farther than
+    # this from a non-pursuit incident's verified pin (the same 4x
+    # location_proximity_km the slow path accepts). The LLM tier had no
+    # distance check at all, which is how one incident drew a 37 km path.
+    incident_max_spread_km: float = 2.0
+    # server-26#301: an upload arriving more than this long after the call
+    # started is correlated as of its started_at (clock pin), not now. Well
+    # above normal upload + STT lag so on-time calls are never back-dated.
+    late_upload_anchor_minutes: int = 10
     location_proximity_km: float = 0.5        # radius for location-proximity matching
     geocode_max_km: float = 40.0              # reject geocode results farther than this from the node
     incident_auto_resolve_minutes: int = 90   # auto-resolve after N minutes with no new calls (major / unknown severity)
@@ -147,9 +254,7 @@ class Settings(BaseSettings):
     # at 9.7 min. Every correct thin attach in that dump was <= 3.4 min idle
     # and every wrong one was >= 8.2, so 5 separates them with room on both
     # sides. Genuine back-and-forth is handled by the 30-second tier-1 path
-    # above this. Also the escape hatch in routers/upload.py's LLM-orphan gate
+    # above this.
-    # (_recent_incident_on_same_talkgroup, server-26#115) — check both call
-    # sites before retuning this.
     tg_dispatch_thin_idle_minutes: int = 5
 
     # ── Hard caps: an incident past either of these stops accepting calls ──────

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

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

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

@@ -32,7 +32,7 @@ from typing import Optional
 from app.internal.logger import logger
 from app.config import settings
 
-TIERS = ("transcription", "correlation_cheap", "correlation_smart", "extraction")
+TIERS = ("transcription", "correlation_cheap", "extraction")
 
 # Consecutive failures a TRANSIENT condition must reach before it alerts.
 # Permanent conditions skip this entirely and alert on failure #1.

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:
     """

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

@@ -14,13 +14,18 @@ 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
+A COST GATE SINCE server-26#298, through `gate_chatter`, not
-extraction based on this verdict. `intelligence.extract_scenes` calls
+`classify_chatter` — built but OFF by default (`chatter_gate_enabled`) until a
-`classify_chatter` purely to record the verdict on the call doc
+shadow window shows gated calls don't over-merge. `classify_chatter` is still
-(`chatter_classifier_verdict` / `chatter_classifier_reason`) so it becomes
+the shadow verdict recorded on every call doc (`chatter_classifier_verdict` /
-observable in the next `/admin` correlation-debug dump, exactly like
+`chatter_classifier_reason`), so the `/admin` tallies keep measuring the same
-`corr_gate_veto` (server-26#115 / PR #126). See the TODO at that call site for
+thing. The gate is stricter on the one shape the #127 review flagged as a live
-what has to be true before this flips 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. When enabled, 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 +138,36 @@ def classify_chatter(transcript: Optional[str]) -> tuple[bool, Optional[str]]:
         return False, None
 
     return True, "bare_acknowledgement"
+
+
+# Words a roll call is made of, beyond the filler/designator/phonetic sets.
+_ROLL_CALL_WORDS = frozenset({
+    "roll", "call", "all", "cars", "car", "patrol", "patrols", "troop", "staff",
+    "concludes", "time", "came", "because", "volunteer", "nyc", "kl", "henry",
+    "here", "present", "s", "m", "it",
+})
+# A roll call may name a place or two ("Post 4, Ossining"); more than this many
+# leftover words means something else is being said in the same transmission.
+_ROLL_CALL_MAX_LEFTOVER = 2
+
+
+def gate_chatter(transcript: Optional[str]) -> tuple[bool, Optional[str]]:
+    """
+    Whether to skip paid AI for this transcript (server-26#298).
+
+    `classify_chatter`'s verdict, with roll_call narrowed: the transcript must
+    be roll-call content end to end, not a roll call mentioned on the way to
+    something real. Precision over recall still — a false positive here means
+    a real event never gets a scene.
+    """
+    is_chatter, reason = classify_chatter(transcript)
+    if not is_chatter:
+        return False, None
+    if reason == "roll_call":
+        leftover = [
+            t for t in _tokenize(transcript or "")
+            if not _is_filler_token(t) and t not in _ROLL_CALL_WORDS
+        ]
+        if len(leftover) > _ROLL_CALL_MAX_LEFTOVER:
+            return False, None
+    return True, reason

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

@@ -16,7 +16,6 @@ what it actually spent instead of an estimate.
 """
 import json
 import threading
-from contextvars import ContextVar
 from typing import Optional
 
 from app.config import settings
@@ -27,18 +26,6 @@ _client_lock = threading.Lock()
 # Models that rejected a thinking level: retried without one from then on.
 _no_thinking_level: set[str] = set()
 
-_usage_sink: ContextVar[Optional[dict]] = ContextVar("drb_gemini_usage", default=None)
-
-
-def collect_usage(sink: Optional[dict]):
-    """Route token counts for the current context into `sink` (a replay run). Returns a reset token."""
-    return _usage_sink.set(sink)
-
-
-def reset_usage(token) -> None:
-    _usage_sink.reset(token)
-
-
 def _get_client():
     global _client
     with _client_lock:
@@ -71,14 +58,8 @@ def _record(purpose: str, model: str, usage) -> None:
     prompt = getattr(usage, "prompt_token_count", None) or 0
     output = getattr(usage, "candidates_token_count", None) or 0
     thoughts = getattr(usage, "thoughts_token_count", None) or 0
-    logger.info(f"gemini usage {purpose} {model}: in={prompt} out={output} thinking={thoughts}")
+    from app.internal.llm import record_usage
-    sink = _usage_sink.get()
+    record_usage(purpose, model, prompt, output, thoughts)
-    if sink is not None:
-        row = sink.setdefault(f"{purpose}:{model}", {"calls": 0, "in": 0, "out": 0, "thinking": 0})
-        row["calls"] += 1
-        row["in"] += prompt
-        row["out"] += output
-        row["thinking"] += thoughts
 
 
 def generate_json(model: str, prompt: str, *, purpose: str,

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

@@ -43,6 +43,13 @@ Matching priority (in order):
 Calls with no incident_type skip new-incident creation but still run paths 1–3,
 so unclassified calls (short transport end, "en route", etc.) can link to an
 existing incident via talkgroup match.
+
+Decision flow (server-26#299): the rules engine decides every confident case on
+its own. When it flags its decision as ambiguous (`_ambiguity`: opening an
+incident while live candidates exist, orphaning an unfit same-talkgroup
+follow-up, or linking on unverified shared units), routers/upload.py makes ONE
+cheap LLM call (llm_correlator.decide) and commits its answer instead. There is
+no second LLM tier and no consensus vote.
 """
 import math
 import re
@@ -65,26 +72,65 @@ _SEVERITY_RANK = {"routine": 0, "minor": 1, "moderate": 2, "major": 3}
 
 def _max_severity(current: Optional[str], new: Optional[str]) -> str:
     """
-    Highest of two severities on the four-level ladder — the merge rule used
+    Highest of two severities on the four-level ladder. A malformed or
-    every time a call attaches to an existing incident.
+    unrecognized value from either side ranks as "routine" so it can never
-
+    suppress a real escalation.
-    Severity is monotonic by design: it only ever rises, never falls, as more
-    calls link. An incident briefly assessed "major" genuinely was major at
-    that moment; a later call that sounds calmer ("units clear", dispatcher
-    moving on) is evidence the SITUATION is winding down, not that the earlier
-    read was wrong. That's what `status`/`resolved_at` are for — resolution
-    retires an incident, it doesn't retroactively erase how serious it was.
-    Every severity-driven surface (worst-first incident rail, "Major only"
-    filter, map colour) exists to make sure a major event is never missed;
-    downgrading severity mid-incident would silently defeat that on the exact
-    incidents it exists to protect. A malformed/unrecognized value from either
-    side ranks as "routine" so it can never suppress a real escalation.
     """
     current = current if current in _SEVERITY_RANK else "routine"
     new     = new if new in _SEVERITY_RANK else "routine"
     return current if _SEVERITY_RANK[current] >= _SEVERITY_RANK[new] else new
 
 
+# server-26#190 — explicit stand-down / unfounded dispositions. Deliberately a
+# short list of phrases that only ever mean "there is less here than was
+# reported"; a calmer-SOUNDING follow-up ("Engine 5 on scene", "copy") is not
+# one, because extraction judges each transmission on its own words and a
+# working fire's routine status traffic would otherwise talk it down to
+# routine. Left out: "disregard"/"cancel" ("disregard my last" is a radio
+# correction), "all clear" ("primary search all clear" on a working fire) and
+# "code 4" (scene secure, not "nothing happened").
+_STAND_DOWN_RE = re.compile(
+    r"\b(?:unfounded|false alarm|no (?:fire|flames|smoke showing)|nothing (?:showing|found|evident)"
+    r"|gone on arrival|g\.?o\.?a\.?|stand(?:ing)? down|stood down"
+    r"|fire(?:'s| is)? out)\b",
+    re.IGNORECASE,
+)
+
+
+def _is_stand_down(transcript: Optional[str]) -> bool:
+    return bool(transcript and _STAND_DOWN_RE.search(transcript))
+
+
+def _severity_after_link(
+    inc: dict, call_severity: Optional[str], transcript: Optional[str],
+) -> dict:
+    """
+    The incident's severity fields after one more call links (server-26#190).
+
+    `severity` is the CURRENT read and drives the live surfaces (worst-first
+    rail, map colour, severity-scaled quiet timer). It rises immediately on
+    any worse call, and falls only on an explicit stand-down disposition
+    ("they lit a fireplace... only smoke showing, no flames") — then to that
+    call's own severity. It used to only ever ratchet up, which left a false
+    Major on the wall for the full 90-minute Major quiet timer after units
+    reported nothing there.
+
+    `severity_peak` keeps the old monotonic meaning (the worst this incident
+    was ever judged), so history and any "was this ever major" question still
+    have an answer that later chatter cannot erase.
+    """
+    current = inc.get("severity") if inc.get("severity") in _SEVERITY_RANK else "routine"
+    peak    = _max_severity(inc.get("severity_peak"), current)
+    new     = call_severity if call_severity in _SEVERITY_RANK else "routine"
+    out     = {"severity_peak": _max_severity(peak, new)}
+    if _SEVERITY_RANK[new] < _SEVERITY_RANK[current] and _is_stand_down(transcript):
+        out["severity"] = new
+        out["severity_deescalated_from"] = current
+    else:
+        out["severity"] = _max_severity(current, new)
+    return out
+
+
 # Maximum plausible ground speed for a moving incident (pursuit/transport).
 # ~3 miles/min ≈ 180 mph — well above real pursuit speeds, but generous enough
 # to tolerate GPS drift and call-timing jitter.  Anything faster is a bad geocode
@@ -278,16 +324,88 @@ def has_event_substance(ctx: dict) -> bool:
     a vehicle, a geocode, or a tag.
 
     This is the substance half of the incident-creation gate (see
-    `_run_decision`, "Severity, not type, decides..."), factored out so the
+    `resolve_new_incident_type`). `call_units` and `location` are deliberately
-    consensus LLM-orphan gate in routers/upload.py mirrors it exactly and can
+    excluded — radio protocol puts a unit ID and a place name in almost every
-    never drop a call the creation gate would have opened. `call_units` and
+    transmission, so counting them as substance makes the check trivially true.
-    `location` are deliberately excluded — radio protocol puts a unit ID and a
-    place name in almost every transmission, so counting them as substance
-    makes the check trivially true.
     """
     return bool(ctx.get("call_vehicles") or ctx.get("coords") or ctx.get("tags"))
 
 
+def resolve_new_incident_type(ctx: dict) -> Optional[str]:
+    """
+    The type a NEW incident opened by this call gets, or None when the call is
+    not incident-worthy and must orphan instead. The one creation gate, shared
+    by the rules engine and the LLM tier (llm_correlator._parse_response), so
+    an LLM "new" can neither open what the rules gate refuses nor be demoted
+    to orphan for lacking a type the rules gate would have defaulted.
+
+    Severity, not type, decides whether a call is incident-worthy. Requiring a
+    concrete incident_type meant a channel whose traffic never classifies —
+    transit/rail administration, records lookups, prisoner transports — could
+    never open a SECOND incident: every later call funnelled into whichever
+    incident was created first (2026-08-16: TG 9048, 28 calls / 49 min)
+    alongside 30+ permanent orphans. So anything the extractor judged a real
+    event (minor+), or that carries substance (`has_event_substance`), opens
+    an incident under the neutral "other" type. `call_units` and `location`
+    are NOT substance — the first version of this gate counted them, turned
+    "11-Victor, 72 at Holland Station" into its own incident, and left 37 of
+    50 incidents open, one call each.
+    """
+    resolved = ctx.get("incident_type") or _infer_type_from_tags(ctx.get("tags") or [])
+    if resolved:
+        return resolved
+    if ctx.get("call_severity") in ("minor", "moderate", "major") or has_event_substance(ctx):
+        return "other"
+    return None
+
+
+def is_clearance_only(ctx: dict) -> bool:
+    """
+    "45-9, I'm clear." carries one fact: which unit is done. Only the rules
+    engine's unit match can say which incident that is; an LLM link here would
+    apply the clear to whatever incident it picked, so such a call is never
+    sent to the LLM tier.
+    """
+    units = ctx.get("call_units") or []
+    cleared = ctx.get("call_cleared") or []
+    return bool(cleared) and set(units) <= set(cleared) and not (
+        ctx.get("tags") or ctx.get("location") or ctx.get("call_vehicles") or ctx.get("incident_type")
+    )
+
+
+_LLM_TRUE  = frozenset({"true", "yes", "1"})
+_LLM_FALSE = frozenset({"false", "no", "0", "", "none", "null"})
+
+
+def parse_llm_bool(value) -> bool:
+    """
+    A yes/no field from an LLM, failing CLOSED (server-26#88).
+
+    `bool(value)` reads the JSON *string* "false" as True, so a model that
+    quoted its booleans would resolve whatever incident the scene linked into.
+    Only a real bool, or an exact true-ish token, counts as True; anything
+    else — "false", "maybe", a number, a dict — is False. For `resolved` and
+    `reassignment` a missed True costs a stale open incident or one missed
+    unit release; a false True closes a live incident or blanks a call's units.
+
+    Public because the extraction boundary (intelligence.py) is where it
+    belongs; routers/upload.py applies it too, so a value that reaches the
+    scene loop un-coerced is still validated there.
+    """
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        token = value.strip().lower()
+        if token in _LLM_TRUE:
+            return True
+        if token not in _LLM_FALSE:
+            logger.warning(f"parse_llm_bool: unrecognised value {value!r} read as False")
+        return False
+    if value is not None:
+        logger.warning(f"parse_llm_bool: non-boolean {type(value).__name__} {value!r} read as False")
+    return False
+
+
 def _infer_type_from_tags(tags: list[str]) -> Optional[str]:
     """Return an incident type inferred from tags, or None if ambiguous."""
     for tag in tags:
@@ -483,7 +601,7 @@ def _resolve_incident_title(
     of strictly higher severity.  Rationale in the commit message; in short, an
     incident's identity is the event that opened it, and the one situation
     where the header must change is the one where things got worse.  This
-    mirrors `_max_severity`: monotonic, never walked back by later chatter.
+    mirrors `severity_peak`: monotonic, never walked back by later chatter.
 
     Two non-renames are still allowed, because neither replaces an event name:
       • filling in a placeholder title on an incident that opened on a call
@@ -656,6 +774,43 @@ def _incident_at_capacity(inc: dict, now: datetime) -> Optional[str]:
     return None
 
 
+def _weak_link_capped(inc: dict, now: datetime) -> bool:
+    """server-26#189 — past this span an incident takes only strong evidence."""
+    return _incident_span_minutes(inc, now) >= settings.incident_weak_link_max_span_minutes
+
+
+def llm_candidates(ctx: dict) -> list[dict]:
+    """
+    The incidents the LLM tier may see and link into: the rules engine's
+    `recent` pool minus anything at the hard caps or past the weak-link span
+    cap (server-26#189). Every LLM link is weak evidence — a model's reading
+    of two transcripts — so an incident that has been open long enough to
+    have become a catch-all is not offered to it at all.
+    """
+    now = ctx["now"]
+    return [
+        inc for inc in ctx.get("recent") or []
+        if not _incident_at_capacity(inc, now) and not _weak_link_capped(inc, now)
+    ]
+
+
+def llm_link_allowed(inc: dict, ctx: dict) -> bool:
+    """
+    Distance guard on an LLM-chosen link (server-26#189). When both the call
+    and a non-pursuit incident are geocoded and farther apart than
+    `incident_max_spread_km`, the model's link is refused: every rules path
+    already has a location check of its own, and the LLM tier had none.
+    """
+    coords = ctx.get("coords")
+    pin = _verified_pin(inc)
+    if not coords or not pin:
+        return True
+    if set(inc.get("tags") or []) & _PURSUIT_TAGS:
+        return True
+    dist = _haversine_km(coords["lat"], coords["lng"], pin["lat"], pin["lng"])
+    return dist <= settings.incident_max_spread_km
+
+
 def _drop_capped(incidents: list[dict], now: datetime) -> list[dict]:
     """Remove over-cap incidents from a candidate pool (see _incident_at_capacity)."""
     kept: list[dict] = []
@@ -979,7 +1134,7 @@ def _run_decision(ctx: dict) -> dict:
     # Hard caps, enforced regardless of which path would have matched. Normally
     # a no-op because _build_context already filtered these out; re-applied here
     # so the guarantee holds for any caller that assembles a ctx directly
-    # (tests, the LLM tiebreak path) rather than through _build_context.
+    # (tests) rather than through _build_context.
     _capped_ids    = {
         inc.get("incident_id")
         for inc in (list(ctx["all_active"]) + list(ctx["recent"]))
@@ -1013,6 +1168,10 @@ def _run_decision(ctx: dict) -> dict:
 
     matched_incident: Optional[dict] = None
     corr_debug: dict = {}
+    # Why the fast path refused a same-talkgroup candidate, when it did. A
+    # "no_signal" refusal is the one the rules cannot settle: a substantive
+    # call on a channel with a live incident and no evidence either way.
+    tg_reject: Optional[str] = None
 
     # ── 1. Fast path: talkgroup match with recency gate ──────────────────────
     #
@@ -1168,13 +1327,14 @@ def _run_decision(ctx: dict) -> dict:
                     "corr_incident_idle_min": round(_incident_idle_minutes(candidate, now), 1),
                     "corr_fit_signal": fit_signal,
                 }
-                if fit_signal == "unit_overlap" and call_units:
+                if fit_signal.startswith("unit_") and call_units:
                     corr_debug["corr_matched_units"] = _matching_units(call_units, candidate.get("units"))
                 logger.info(
                     f"Correlator fast-path: call {call_id} → {candidate['incident_id']} "
                     f"(signal={fit_signal})"
                 )
             else:
+                tg_reject = fit_signal
                 logger.info(
                     f"Correlator fast-path skipped: call {call_id} — {fit_signal} "
                     f"from {candidate['incident_id']}; will attempt new incident"
@@ -1207,13 +1367,14 @@ def _run_decision(ctx: dict) -> dict:
                     "corr_candidates": len(tg_recent),
                     "corr_fit_signal": fit_signal,
                 }
-                if fit_signal == "unit_overlap" and call_units:
+                if fit_signal.startswith("unit_") and call_units:
                     corr_debug["corr_matched_units"] = _matching_units(call_units, candidate.get("units"))
                 logger.info(
                     f"Correlator fast-path (disambig {len(tg_recent)} candidates): "
                     f"call {call_id} → {candidate['incident_id']} (signal={fit_signal})"
                 )
             else:
+                tg_reject = fit_signal
                 logger.info(
                     f"Correlator fast-path disambig: no candidate fits call {call_id} "
                     f"across {len(tg_recent)} incidents ({fit_signal}) — will attempt new incident"
@@ -1249,28 +1410,27 @@ def _run_decision(ctx: dict) -> dict:
                 and inc.get("updated_at", "") > best_unit_inc.get("updated_at", "")
                 for inc in all_active
             )
-            # Location conflict check: same unit at a clearly different location
+            # Same unit-overlap judgement as the fast path (_unit_fit): a
-            # means the officer has moved to a new scene (e.g. dispatch sent
+            # location conflict means the officer has moved to a new scene, a
-            # them to a second call without an explicit reassignment signal).
+            # sink past the weak-link span cap takes no unit-only links
-            location_conflict = False
+            # (server-26#189), and an old incident with no location or content
-            if coords and not reassigned_away:
+            # agreement is linked but flagged unverified, i.e. ambiguous.
-                inc_coords_uc = best_unit_inc.get("location_coords")
+            uc_fit, uc_signal = (False, "reassigned_away")
-                if inc_coords_uc:
+            if not reassigned_away:
-                    uc_dist = _haversine_km(
+                uc_fit, uc_signal = _unit_fit(
-                        coords["lat"], coords["lng"],
+                    best_unit_inc, coords, call_embedding, now,
-                        inc_coords_uc["lat"], inc_coords_uc["lng"],
+                    settings.location_proximity_km,
-                    )
+                )
-                    if uc_dist > settings.location_proximity_km:
+            if not uc_fit:
-                        location_conflict = True
+                logger.info(
-                        logger.info(
+                    f"Correlator unit-continuity skipped: call {call_id} vs "
-                            f"Correlator unit-continuity skipped: call {call_id} — "
+                    f"{best_unit_inc['incident_id']} — {uc_signal}"
-                            f"unit match but location conflict ({uc_dist:.2f}km) with "
+                )
-                            f"{best_unit_inc['incident_id']}"
+            else:
-                        )
-            if not reassigned_away and not location_conflict:
                 matched_incident = best_unit_inc
                 corr_debug = {
                     "corr_path": "unit-continuity",
+                    "corr_fit_signal": uc_signal,
                     "corr_incident_idle_min": round(_incident_idle_minutes(best_unit_inc, now), 1),
                     # Unlike this file's other two paths (fast/single, fast/disambig),
                     # a match here is unit-driven by construction (call_unit_set &
@@ -1445,63 +1605,66 @@ def _run_decision(ctx: dict) -> dict:
 
     # ── Decision output ───────────────────────────────────────────────────────
     if matched_incident:
-        return {
+        action, resolved_type = "link", incident_type
-            "action": "link",
+    elif not create_if_new:
-            "matched_incident": matched_incident,
+        action, resolved_type = "orphan", None
-            "incident_type": incident_type,
+    else:
-            "corr_debug": corr_debug,
+        resolved_type = resolve_new_incident_type(ctx)
-        }
+        action = "new" if resolved_type else "orphan"
-
+        if resolved_type and resolved_type != incident_type:
-    if not create_if_new:
-        return {"action": "orphan", "matched_incident": None, "incident_type": None, "corr_debug": corr_debug}
-
-    # Attempt type inference from tags before giving up on creation
-    resolved_type = incident_type
-    if not resolved_type and tags:
-        resolved_type = _infer_type_from_tags(tags)
-        if resolved_type:
             logger.info(
-                f"Correlator: inferred incident_type={resolved_type!r} from tags {tags} for call {call_id}"
+                f"Correlator: call {call_id} has no incident_type — opening a "
+                f"{resolved_type!r} incident (severity={call_severity}, tags={tags})"
             )
 
-    # Severity, not type, decides whether a call is incident-worthy.
+    decision = {
-    #
+        "action": action,
-    # Requiring a concrete incident_type here meant a channel whose traffic never
+        "matched_incident": matched_incident,
-    # classifies — transit/rail administration, records lookups, prisoner
-    # transports — could never open a SECOND incident.  Every later call on that
-    # talkgroup then funnelled into whichever incident happened to be created
-    # first, producing hour-long incidents made of unrelated transmissions
-    # (2026-08-16: TG 9048, 28 calls / 49 min) alongside 30+ permanent orphans.
-    #
-    # Anything the extractor judged a real event, or that carries any concrete
-    # content, now opens an incident under the neutral "other" type.  Only
-    # content-free routine traffic is still left for the thin path to attach.
-    #
-    # `call_units` and `location` are NOT substance.  Radio protocol puts a unit
-    # ID in essentially every transmission and a place name in most of them, so
-    # including them made has_substance true almost always and the severity check
-    # dead code — the first version of this gate turned "11-Victor, 72 at Holland
-    # Station" into its own incident and left 37 of 50 incidents open, one call
-    # each.  A vehicle, a geocode, or a tag means the extractor found something
-    # beyond who was speaking and where they stood.
-    if not resolved_type:
-        has_substance = has_event_substance(ctx)
-        if call_severity in ("minor", "moderate", "major") or has_substance:
-            resolved_type = "other"
-            logger.info(
-                f"Correlator: call {call_id} has no incident_type — opening an 'other' "
-                f"incident (severity={call_severity}, substance={has_substance})"
-            )
-
-    if not resolved_type:
-        return {"action": "orphan", "matched_incident": None, "incident_type": None, "corr_debug": corr_debug}
-
-    return {
-        "action": "new",
-        "matched_incident": None,
         "incident_type": resolved_type,
         "corr_debug": corr_debug,
     }
+    decision["ambiguous"] = _ambiguity(decision, tg_reject, ctx)
+    return decision
+
+
+def _ambiguity(decision: dict, tg_reject: Optional[str], ctx: dict) -> Optional[str]:
+    """
+    Why the rules decision is NOT confident enough to commit alone, or None
+    when it is (server-26#299). Only an ambiguous decision is sent to the LLM
+    tier, once; every other decision is the rules engine's alone.
+
+    Ambiguous:
+      • "new_with_candidates" — the rules would OPEN an incident while live,
+        LLM-eligible incidents exist. Opening is the expensive mistake: this
+        system over-splits, and rules-only correlation opened ~1.8x the
+        incidents per hour of LLM-assisted (#170, 2026-10-04 control window)
+        — duplicates of a running job and typed radio housekeeping.
+      • "unfit_tg_candidate" — the rules would ORPHAN a substantive call that
+        has a live incident on its own talkgroup but no evidence for or
+        against it (fit test "no_signal"): an unfit follow-up.
+      • "unverified_unit_link" — the rules would LINK on shared units alone
+        into an incident that is no longer fresh, with no location or content
+        agreement (mechanism B; #103: the content veto has nothing to measure
+        when the scene has no embedding). Between the divergence and match
+        embedding bars (#86) a link lands here too.
+
+    Never ambiguous: thin calls (nothing to reason about), clearance-only
+    calls (only the unit match can place a 10-8), link-only sweeps
+    (create_if_new=False — rules-only by design), or when the LLM would have
+    no candidates to choose from.
+    """
+    if not ctx.get("create_if_new", True) or ctx.get("is_thin_call"):
+        return None
+    if is_clearance_only(ctx) or not llm_candidates(ctx):
+        return None
+    action = decision["action"]
+    if action == "new":
+        return "new_with_candidates"
+    if action == "orphan" and tg_reject == "no_signal":
+        return "unfit_tg_candidate"
+    if action == "link" and (decision["corr_debug"] or {}).get("corr_fit_signal") == "unit_overlap_unverified":
+        return "unverified_unit_link"
+    return None
 
 
 # ─────────────────────────────────────────────────────────────────────────────
@@ -1514,7 +1677,7 @@ async def _apply_and_log(decision: dict, ctx: dict) -> Optional[str]:
 
     server-26#96: every scene of a multi-scene call reaches this function
     independently (upload.py's ``for scene_index, scene in enumerate(scenes):``
-    loop → _correlate_with_consensus → preview_correlation/apply_correlation),
+    loop → _correlate → preview_correlation/apply_correlation),
     and every scene writes to the SAME call doc. Writing corr_debug only at
     the flat top level meant scene 2's write silently clobbered scene 1's
     corr_path/corr_consensus/etc — the call doc ended up describing a splice
@@ -1553,18 +1716,14 @@ async def _apply_and_log(decision: dict, ctx: dict) -> Optional[str]:
                 "incident_id":   incident_id,
                 "corr_debug":    corr_debug,
                 # server-26#139: this scene's OWN extracted incident_type/
-                # severity, as read by _call_is_substanceless's ctx at
+                # severity, as read from ctx at decision time — not the call
-                # decision time — not the call doc's flat top-level field,
+                # doc's flat top-level field, which is last-scene-wins
-                # which is last-scene-wins (server-26#96) and was the reason
+                # (server-26#96).
-                # #138's "type" veto couldn't be told apart from cross-scene
-                # contamination without re-guessing from a live dump.
                 # NOTE: unlike incident_type, call_severity is already
                 # coerced to "routine" when extraction emitted nothing
                 # (ctx build: `severity or "routine"`) — a scene reading
                 # "routine" here doesn't distinguish "extraction said
-                # routine" from "extraction said nothing". Don't split a
+                # routine" from "extraction said nothing".
-                # severity veto the way #138 splits the type veto without
-                # accounting for that.
                 "incident_type": ctx.get("incident_type"),
                 "severity":      ctx.get("call_severity"),
             }
@@ -1613,6 +1772,7 @@ async def _apply_decision(decision: dict, ctx: dict) -> Optional[str]:
             talkgroup_name=talkgroup_name, incident_type=incident_type,
             cleared_units=call_cleared, refresh_activity=not thin_link,
             call_severity=call_severity, call_srcaddr=call_srcaddr,
+            call_transcript=ctx.get("scene_transcript"),
         )
         return matched_incident["incident_id"]
 
@@ -1798,6 +1958,65 @@ def _disambiguate(
     return best
 
 
+def _unit_fit(
+    inc: dict,
+    call_coords: Optional[dict],
+    call_embedding: Optional[list],
+    now: Optional[datetime],
+    proximity_km: float,
+) -> tuple[bool, str]:
+    """
+    Judge a link whose evidence is a shared unit. Same officer usually means
+    same call — but on a shared dispatch channel a patrol unit works job after
+    job, so shared units alone grow chains (mechanism B,
+    CORRELATION_REVIEW_0820.md). In order:
+
+      • call and incident both geocoded: far apart → (False,
+        "unit_loc_conflict"), the officer moved to a new scene; close →
+        (True, "unit_location_match").
+      • embeddings agree at the match bar → (True, "unit_content_match").
+      • incident past the weak-link span cap (#189) → (False,
+        "weak_link_span_cap"): a long-running incident stops taking
+        unit-only evidence.
+      • incident fresh (idle < unit_link_fresh_idle_minutes) and young (open
+        < unit_link_confident_max_span_minutes) → (True, "unit_overlap"). No
+        content veto here: an update on the same scene often shares little
+        vocabulary with the dispatch and does not restate the address.
+      • not fresh, embeddings below the DIVERGENCE bar (#86) → (False,
+        "content_divergence").
+      • anything else → (True, "unit_overlap_unverified"): the rules keep the
+        link but cannot vouch for it, so the decision is ambiguous and goes
+        to the LLM tier. This is where a scene with no embedding lands
+        (#103) — it used to skip the veto and link as if verified, which
+        removed the only brake on the unit loop for ~93% of calls.
+    """
+    inc_coords = inc.get("location_coords")
+    if call_coords and inc_coords:
+        dist_km = _haversine_km(
+            call_coords["lat"], call_coords["lng"],
+            inc_coords["lat"], inc_coords["lng"],
+        )
+        return (False, "unit_loc_conflict") if dist_km > proximity_km else (True, "unit_location_match")
+
+    sim: Optional[float] = None
+    inc_emb = inc.get("embedding")
+    if call_embedding and inc_emb:
+        sim = _cosine_similarity(call_embedding, inc_emb)
+        if sim >= settings.embedding_similarity_threshold:
+            return True, "unit_content_match"
+
+    if now is None:
+        return True, "unit_overlap_unverified"
+    if _weak_link_capped(inc, now):
+        return False, "weak_link_span_cap"
+    fresh = _idle_gate_minutes(inc, now) < settings.unit_link_fresh_idle_minutes
+    if fresh and _incident_span_minutes(inc, now) < settings.unit_link_confident_max_span_minutes:
+        return True, "unit_overlap"
+    if not fresh and sim is not None and sim < settings.embedding_divergence_threshold:
+        return False, "content_divergence"
+    return True, "unit_overlap_unverified"
+
+
 def _call_fits_incident(
     inc: dict,
     call_units: list[str],
@@ -1812,18 +2031,14 @@ def _call_fits_incident(
     Return (fits, signal) — fits is True when this call plausibly belongs to
     the incident; signal names the specific evidence that drove the decision.
 
-    fits=True  signals: "unit_overlap" | "vehicle_overlap" | "location_proximity"
+    fits=True  signals: see `_unit_fit` for the unit_* family | "vehicle_overlap"
+                        | "location_proximity"
     fits=False signals: "unit_loc_conflict" | "content_divergence"
-                        | "location_conflict" | "no_signal"
+                        | "weak_link_span_cap" | "location_conflict" | "no_signal"
 
     Evaluation order:
-    1. Unit overlap. Same officer = same call. Also runs a location-conflict
+    1. Unit overlap — judged by `_unit_fit`, which also decides whether the
-       guard: geocoded on both sides and clearly different → the officer has
+       link is confident or only "unit_overlap_unverified" (ambiguous).
-       moved to a new scene, false positive. No geocode on the call (or on
-       the incident, with a call geocode) AND the incident is old (≥ 15 min)
-       → content divergence as a location proxy (embedding similarity < 0.82
-       → different scene). Skipped for recent incidents (< 15 min) — an
-       update without re-stating the address is normal.
     2. Vehicle overlap → True.
     3. Location proximity. Both geocoded and close → True; far apart with no
        other positive signal → False.
@@ -1850,44 +2065,9 @@ def _call_fits_incident(
     inc_units = _unit_keys(inc.get("units"))
     matched_units = _matching_units(call_units, inc.get("units"))
     if matched_units:
-        if call_coords:
+        fit, signal = _unit_fit(inc, call_coords, call_embedding, now, proximity_km)
-            # Hard location conflict: geocoded on both sides and clearly different.
+        logger.info(f"  fits[{inc_id}]: unit_overlap matched={matched_units} → {signal}")
-            inc_coords_u = inc.get("location_coords")
+        return fit, signal
-            if inc_coords_u:
-                dist_km = _haversine_km(
-                    call_coords["lat"], call_coords["lng"],
-                    inc_coords_u["lat"], inc_coords_u["lng"],
-                )
-                if dist_km > proximity_km:
-                    logger.info(f"  fits[{inc_id}]: unit_overlap({matched_units}) but location_conflict dist={dist_km:.2f}km → unit_loc_conflict")
-                    return False, "unit_loc_conflict"
-            elif call_embedding and idle_min >= 15:
-                # Call has geocode but incident doesn't — fall back to content
-                # divergence as a location proxy.  Without this, stale incidents
-                # that never geocoded absorb unrelated calls purely on unit
-                # overlap (e.g. a patrol officer working a second scene 70 min
-                # after the original call).
-                inc_emb_u = inc.get("embedding")
-                if inc_emb_u:
-                    sim = _cosine_similarity(call_embedding, inc_emb_u)
-                    if sim < 0.82:
-                        logger.info(f"  fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence (has_call_coords/no_inc_coords) sim={sim:.3f} → content_divergence")
-                        return False, "content_divergence"
-        elif call_embedding and idle_min >= 15:
-            # No geocode available AND old incident: use content divergence as a
-            # location-proxy veto.  After 15+ minutes an officer at a completely
-            # different scene will be discussing clearly different content.
-            # Skip this for recent incidents — an officer updating on the same
-            # scene without re-stating the address is normal and their update
-            # won't share much vocabulary with the original dispatch.
-            inc_emb_u = inc.get("embedding")
-            if inc_emb_u:
-                sim = _cosine_similarity(call_embedding, inc_emb_u)
-                if sim < 0.82:
-                    logger.info(f"  fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence sim={sim:.3f} → content_divergence")
-                    return False, "content_divergence"
-        logger.info(f"  fits[{inc_id}]: unit_overlap matched={matched_units} → unit_overlap")
-        return True, "unit_overlap"
 
     # ── 2. Vehicle overlap ────────────────────────────────────────────────────
     inc_vehicles = set(inc.get("vehicles") or [])
@@ -2047,6 +2227,7 @@ async def _update_incident(
     refresh_activity: bool = True,
     call_severity: Optional[str] = None,
     call_srcaddr: Optional[str] = None,
+    call_transcript: Optional[str] = None,
 ) -> None:
     incident_id = inc["incident_id"]
 
@@ -2109,7 +2290,9 @@ async def _update_incident(
         "units_cleared":     units_cleared,
         "location_mentions": location_mentions,
         "summary_stale":     True,
-        "severity":          _max_severity(inc.get("severity"), call_severity),
+        # Current severity can step DOWN on an explicit stand-down; the peak
+        # never does (server-26#190, _severity_after_link).
+        **_severity_after_link(inc, call_severity, call_transcript),
         # Always all three, always together — writing one without the others is
         # what let the label and the pin drift apart (server-26#23).
         **location_fields,
@@ -2123,7 +2306,18 @@ async def _update_incident(
     # acknowledging.  The incident now ages from its last SUBSTANTIVE call, and
     # thin traffic rides along without extending its life.
     if refresh_activity:
-        updates["updated_at"] = _floor_at_started_at(inc, now).isoformat()
+        # Never move updated_at BACKWARD: a back-dated `now` (sweep, or a late
+        # upload's clock pin, server-26#301) would make a live incident look
+        # idle, failing on-time calls' recency gates and timer-closing it early.
+        when = _floor_at_started_at(inc, now)
+        try:
+            prev = datetime.fromisoformat(str(inc.get("updated_at") or "").replace("Z", "+00:00"))
+            if prev.tzinfo is None:
+                prev = prev.replace(tzinfo=timezone.utc)
+            when = max(when, prev)
+        except ValueError:
+            pass
+        updates["updated_at"] = when.isoformat()
         if is_new_call:  # a second scene of the same call is not a second call
             updates["substantive_call_count"] = (
                 inc.get("substantive_call_count")
@@ -2233,6 +2427,7 @@ async def _create_incident(
         "vehicles":          call_vehicles,
         "srcaddrs":          [call_srcaddr] if call_srcaddr else [],
         "severity":          call_severity,
+        "severity_peak":     call_severity,
         "summary":           None,
         "summary_stale":     True,
         "summary_last_run":  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,14 @@ 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 import incident_correlator
+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.
@@ -218,11 +221,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 +235,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 +277,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 +328,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 +399,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           = incident_correlator.parse_llm_bool(scene.get("resolved"))
+        reassignment:       bool           = incident_correlator.parse_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 +766,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 +791,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 +1031,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 +1053,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 +1068,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 +1078,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 +1087,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/internal/llm.py

@@ -0,0 +1,96 @@
+"""
+One entry point for the two small JSON-mode LLM jobs: the correlation decision
+on ambiguous scenes and transcript correction.
+
+The provider follows the model name: a `gemini-*` model goes through
+app/internal/gemini.py, anything else is an OpenAI chat model. Since the
+2026-11-08 provider consolidation both defaults are OpenAI (gpt-4o-mini), so
+one funded account (OpenAI) covers extraction, embeddings, summaries and these
+two jobs; Groq stays optional for transcription. Pointing either model setting
+back at a gemini-* ID needs no code change.
+
+Token counts from both providers land in the same usage sink, so a replay run
+(app/internal/replay.py) reports what it spent whichever provider served it.
+"""
+import json
+from contextvars import ContextVar
+from typing import Optional
+
+from app.config import settings
+from app.internal.logger import logger
+
+_usage_sink: ContextVar[Optional[dict]] = ContextVar("drb_llm_usage", default=None)
+
+
+def collect_usage(sink: Optional[dict]):
+    """Route token counts for the current context into `sink` (a replay run). Returns a reset token."""
+    return _usage_sink.set(sink)
+
+
+def reset_usage(token) -> None:
+    _usage_sink.reset(token)
+
+
+def record_usage(purpose: str, model: str, prompt: int, output: int, thinking: int = 0) -> None:
+    logger.info(f"llm usage {purpose} {model}: in={prompt} out={output} thinking={thinking}")
+    sink = _usage_sink.get()
+    if sink is not None:
+        row = sink.setdefault(f"{purpose}:{model}", {"calls": 0, "in": 0, "out": 0, "thinking": 0})
+        row["calls"] += 1
+        row["in"] += prompt
+        row["out"] += output
+        row["thinking"] += thinking
+
+
+def provider_for(model: str) -> str:
+    return "gemini" if (model or "").startswith("gemini") else "openai"
+
+
+def available(model: str) -> bool:
+    """Whether the provider serving `model` has a key configured."""
+    key = settings.gemini_api_key if provider_for(model) == "gemini" else settings.openai_api_key
+    return bool(key)
+
+
+def timeout_seconds(model: str) -> float:
+    if provider_for(model) == "gemini":
+        return settings.gemini_timeout_seconds
+    return (1 + max(0, settings.openai_max_retries)) * settings.openai_timeout_seconds
+
+
+def billing_fix(model: str) -> str:
+    if provider_for(model) == "gemini":
+        return "Top up billing at https://ai.studio/projects"
+    return "Top up billing at https://platform.openai.com/settings/organization/billing"
+
+
+def generate_json(model: str, prompt: str, *, purpose: str,
+                  thinking_level: Optional[str] = "minimal") -> dict:
+    """
+    Synchronous (run it via asyncio.to_thread). Returns the parsed JSON body.
+    Raises on API failure so callers' ai_health classification (billing /
+    dead model / transient) is unchanged. `thinking_level` only applies to
+    Gemini 3.x; OpenAI's gpt-4o family has no such setting.
+    """
+    if provider_for(model) == "gemini":
+        from app.internal import gemini
+        return gemini.generate_json(model, prompt, purpose=purpose, thinking_level=thinking_level)
+
+    from app.internal.ai_clients import openai_client
+    # JSON mode requires the word "JSON" in the prompt; both prompts say
+    # "Return JSON" / "Respond with JSON only".
+    kwargs = {}
+    # Reasoning models (o-series, gpt-5) reject a temperature; chat models get 0.
+    if not model.startswith(("o1", "o3", "o4", "gpt-5")):
+        kwargs["temperature"] = 0
+    resp = openai_client().chat.completions.create(
+        model=model,
+        messages=[{"role": "user", "content": prompt}],
+        response_format={"type": "json_object"},
+        **kwargs,
+    )
+    usage = getattr(resp, "usage", None)
+    record_usage(purpose, model,
+                 getattr(usage, "prompt_tokens", 0) or 0,
+                 getattr(usage, "completion_tokens", 0) or 0)
+    return json.loads(resp.choices[0].message.content or "{}")

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

@@ -1,23 +1,23 @@
 """
-LLM-based incident correlator using Gemini.
+LLM incident correlator (settings.corr_cheap_model, via app/internal/llm.py).
 
-Two functions are exposed:
+One function is exposed:
-  decide(call_id, ctx)                           — cheap first-pass (corr_cheap_model)
+  decide(call_id, ctx) — called by routers/upload.py ONLY when the rules
-  tiebreak(rules_decision, llm_decision, ctx)    — smart tiebreaker (corr_smart_model)
+                         engine marked its decision ambiguous
+                         (incident_correlator._ambiguity), at most once per
+                         scene. server-26#299 removed the second "smart"
+                         tiebreak tier and the cheap/smart consensus vote.
 
-Both return a decision dict compatible with _run_decision() in incident_correlator:
+It returns a decision dict compatible with incident_correlator._apply_decision:
   {"action": "link"|"new"|"orphan",
    "matched_incident": dict|None,
    "incident_type": str|None,
    "corr_debug": dict,
-   "reasoning": str}     ← extra field for logging/tiebreak comparison
+   "reasoning": str}
 
-decide() is skipped for thin calls (no content to reason about) and when
+or None — no API key, nothing to choose from, an answer that fails
-GEMINI_API_KEY is not set — in those cases returns None so the caller knows
+validation, or any LLM failure — and the caller then commits the rules
-to fall back to the rules decision.
+decision unchanged, so the pipeline never stalls on the LLM.
-
-Error handling: any Gemini failure returns None from decide() and the
-rules_decision from tiebreak() so the pipeline never stalls.
 """
 import asyncio
 from datetime import datetime, timezone
@@ -155,8 +155,9 @@ call that reports a new event, below.):
   same one — especially once that incident has been open a long time (see
   open:/calls:).
 - Reserve "new" for a call that clearly describes a DIFFERENT event from every
-  recent incident — a different place, different units, and a different subject,
+  recent incident — a different place or subject, not merely a different
-  not merely a different transmission about the same job.
+  transmission about the same job. Shared units alone do not make it the same
+  event (see EXCEPT above).
 - "orphan" a call that is not an incident at all: radio checks, roll call,
   a unit marking on/off duty or 10-8/10-98, mileage/log entries, a bare
   acknowledgement. Do not open a "new" incident for these.
@@ -166,103 +167,75 @@ call that reports a new event, below.):
 """
 
 
-def _build_decide_prompt(ctx: dict) -> str:
+def _build_decide_prompt(ctx: dict, candidates: list[dict]) -> str:
-    now    = ctx["now"]
+    now = ctx["now"]
-    recent = ctx["recent"]
+    inc_block = "\n".join(_inc_summary(inc, now) for inc in _prompt_incidents(candidates))
-    inc_block = (
-        "\n".join(_inc_summary(inc, now) for inc in _prompt_incidents(recent))
-        if recent else "(none)"
-    )
     return (
         "You are an incident correlator for a public safety radio monitoring system.\n\n"
         "A new radio call has arrived. Decide whether it belongs to an existing active incident, "
-        "represents a new incident, or should be orphaned (not enough information).\n\n"
+        "represents a new incident, or should be orphaned (not an incident at all).\n\n"
         f"NEW CALL:\n{_call_block(ctx)}\n\n"
-        f"ACTIVE INCIDENTS ({len(recent)} recent):\n{inc_block}\n"
+        f"ACTIVE INCIDENTS ({len(candidates)} recent):\n{inc_block}\n"
-        f"{_RULES}\n"
-        f"Respond with JSON only (no markdown):\n{_SCHEMA}"
-    )
-
-
-def _build_tiebreak_prompt(rules_decision: dict, llm_decision: dict, ctx: dict) -> str:
-    now    = ctx["now"]
-    recent = ctx["recent"]
-    inc_block = (
-        "\n".join(_inc_summary(inc, now) for inc in _prompt_incidents(recent))
-        if recent else "(none)"
-    )
-
-    def _fmt(d: dict, name: str) -> str:
-        action  = d.get("action", "?")
-        inc     = d.get("matched_incident")
-        inc_id  = inc["incident_id"] if inc else (d.get("incident_id") or "null")
-        reason  = d.get("reasoning") or (d.get("corr_debug") or {}).get("corr_fit_signal") or "—"
-        return f"  {name}: action={action}, incident_id={inc_id}, reasoning={reason!r}"
-
-    return (
-        "You are a senior incident correlator for a public safety radio monitoring system.\n\n"
-        "Two correlation engines disagree. You must make the final decision.\n\n"
-        f"NEW CALL:\n{_call_block(ctx)}\n\n"
-        f"ACTIVE INCIDENTS ({len(recent)} recent):\n{inc_block}\n\n"
-        "DISAGREEMENT:\n"
-        f"{_fmt(rules_decision, 'Rules engine')}\n"
-        f"{_fmt(llm_decision, 'LLM correlator')}\n"
         f"{_RULES}\n"
         f"Respond with JSON only (no markdown):\n{_SCHEMA}"
     )
 
 
 # ─────────────────────────────────────────────────────────────────────────────
-# Gemini API call (sync, runs in thread pool)
+# LLM call (sync, runs in thread pool)
 # ─────────────────────────────────────────────────────────────────────────────
 
 def _sync_gemini(model_name: str, prompt: str) -> dict:
-    from app.internal import gemini
+    from app.internal import llm
-    return gemini.generate_json(model_name, prompt, purpose="correlation")
+    return llm.generate_json(model_name, prompt, purpose="correlation")
 
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Decision parsing
 # ─────────────────────────────────────────────────────────────────────────────
 
-def _parse_response(raw: dict, ctx: dict) -> dict:
+def _parse_response(raw: dict, ctx: dict, candidates: list[dict]) -> Optional[dict]:
     """
-    Convert raw Gemini JSON output to a decision dict compatible with _run_decision().
+    Convert the model's raw JSON to a decision dict, or None when the answer is
-    Resolves incident_id → full incident doc from ctx["all_active"].
+    unusable and the rules decision should stand.
-    Handles type inference for "new" actions the same way as the rules engine.
-    """
-    from app.internal.incident_correlator import _infer_type_from_tags  # same-package import
 
-    action    = raw.get("action", "orphan")
+    A "link" must name one of the candidates the model was shown — not any
+    active incident: the shown list is the one with the #189 weak-link cap
+    applied — and pass the distance guard (incident_correlator.
+    llm_link_allowed). A "new" takes its type from the same creation gate the
+    rules use, so it is refused (orphan) exactly when the rules would refuse
+    it. An unknown action is unusable, not an orphan: the old "default to
+    orphan" silently dropped calls on a malformed reply.
+    """
+    from app.internal.incident_correlator import llm_link_allowed, resolve_new_incident_type
+
+    action    = raw.get("action")
     reasoning = raw.get("reasoning", "")
-
     if action not in ("link", "new", "orphan"):
-        action = "orphan"
+        logger.warning(f"LLM correlator: unusable action {action!r} — keeping the rules decision")
+        return None
 
     matched_incident: Optional[dict] = None
-
+    incident_type: Optional[str] = None
     if action == "link":
         inc_id = raw.get("incident_id")
-        if inc_id:
+        matched_incident = next((i for i in candidates if i.get("incident_id") == inc_id), None)
-            matched_incident = next(
-                (i for i in ctx["all_active"] if i.get("incident_id") == inc_id),
-                None,
-            )
         if not matched_incident:
             logger.warning(
-                f"LLM correlator: incident_id={inc_id!r} not in active incidents — orphaning"
+                f"LLM correlator: incident_id={inc_id!r} is not a shown candidate — keeping the rules decision"
             )
-            action = "orphan"
+            return None
-
+        if not llm_link_allowed(matched_incident, ctx):
-    incident_type: Optional[str] = None
+            logger.info(
-    if action in ("link", "new"):
+                f"LLM correlator: link to {inc_id} refused — call is geocoded beyond "
+                f"incident_max_spread_km of its pin (server-26#189)"
+            )
+            return None
         incident_type = ctx["incident_type"]
+    elif action == "new":
+        incident_type = resolve_new_incident_type(ctx)
         if not incident_type:
-            incident_type = _infer_type_from_tags(ctx["tags"])
-        if action == "new" and not incident_type:
-            # Can't create an incident without a type — demote to orphan
             action = "orphan"
-            matched_incident = None
 
     return {
         "action":            action,
@@ -279,54 +252,35 @@ def _parse_response(raw: dict, ctx: dict) -> dict:
 
 async def decide(call_id: str, ctx: dict) -> Optional[dict]:
     """
-    Run the cheap LLM correlator (corr_cheap_model) on the call.
+    One cheap-model correlation decision for an ambiguous scene. Returns None
-
+    (keep the rules decision) when the model's provider has no API key, there
-    Returns a decision dict or None if:
+    is no candidate to show, the answer fails validation, or the call fails.
-      - GEMINI_API_KEY is not configured
-      - the call is thin (content-free — no value from LLM)
-      - there are no recent active incidents to reason about
-      - Gemini fails
-
-    Callers should treat None as "fall back to rules decision".
     """
-    if not settings.gemini_api_key:
+    from app.internal.incident_correlator import llm_candidates
+
+    from app.internal import llm
+    if not llm.available(settings.corr_cheap_model):
         return None
-
+    candidates = llm_candidates(ctx)
-    if ctx["is_thin_call"]:
+    if not candidates:
-        return None  # thin calls have no transcript/units/coords to reason about
-
-    if _is_clearance_only(ctx):
-        # "45-9, I'm clear." carries one fact: which unit is done. Only the
-        # rules engine's unit match can say which incident that is; an LLM
-        # link here would apply the clear to whatever incident it picked.
         return None
 
-    if not ctx["recent"]:
-        return None  # no incidents to correlate against — rules handles new-only
-
     try:
-        prompt   = _build_decide_prompt(ctx)
+        prompt   = _build_decide_prompt(ctx, candidates)
         raw      = await asyncio.to_thread(_sync_gemini, settings.corr_cheap_model, prompt)
-        decision = _parse_response(raw, ctx)
+    except Exception as e:
+        await _log_llm_failure("LLM correlator", "correlation_cheap", call_id, settings.corr_cheap_model, e)
+        return None
+    await ai_health.report_healthy("correlation_cheap")
+    decision = _parse_response(raw if isinstance(raw, dict) else {}, ctx, candidates)
+    if decision:
         _id = (decision["matched_incident"] or {}).get("incident_id", "null")
         logger.info(
             f"LLM correlator ({settings.corr_cheap_model}): call {call_id} → "
             f"action={decision['action']} incident={_id} "
             f"reasoning={decision['reasoning']!r}"
         )
-        await ai_health.report_healthy("correlation_cheap")
+    return decision
-        return decision
-    except Exception as e:
-        await _log_llm_failure("LLM correlator", "correlation_cheap", call_id, settings.corr_cheap_model, e)
-        return None
-
-
-def _is_clearance_only(ctx: dict) -> bool:
-    units = ctx.get("call_units") or []
-    cleared = ctx.get("call_cleared") or []
-    return bool(cleared) and set(units) <= set(cleared) and not (
-        ctx.get("tags") or ctx.get("location") or ctx.get("call_vehicles") or ctx.get("incident_type")
-    )
 
 
 _dead_models: set[str] = set()
@@ -350,7 +304,7 @@ async def _log_llm_failure(where: str, tier: str, call_id: str, model: str, exc:
 
     if kind == "dead_model":
         await _log_tier_down(where, tier, model, "model is unavailable",
-                             "Update CORR_CHEAP_MODEL/CORR_SMART_MODEL in config.py", text)
+                             "Update CORR_CHEAP_MODEL in config.py", text)
         return
 
     # A depleted balance reads as 429, the same status as an ordinary rate limit,
@@ -358,20 +312,23 @@ async def _log_llm_failure(where: str, tier: str, call_id: str, model: str, exc:
     # dead account never does. ai_health.classify() keeps a burst of rate limits
     # at WARNING while an empty account escalates like a bad model ID.
     if kind == "billing":
-        await _log_tier_down(where, tier, model, "the Gemini account is out of credit",
+        from app.internal import llm
-                             "Top up billing at https://ai.studio/projects", text)
+        await _log_tier_down(where, tier, model, f"the {llm.provider_for(model)} account is out of credit",
+                             llm.billing_fix(model), text)
         return
 
     logger.warning(f"{where} failed for call {call_id}: {text}")
+    from app.internal import llm
     await ai_health.report_degraded(
-        tier, "gemini", model, "transient API error",
+        tier, llm.provider_for(model), model, "transient API error",
         "no action needed unless this persists", permanent=False,
     )
 
 
 async def _log_tier_down(where: str, tier: str, model: str, problem: str, fix: str, text: str) -> None:
     """ERROR once per model, not once per call — this runs at radio-traffic volume."""
-    await ai_health.report_degraded(tier, "gemini", model, problem, fix, permanent=True)
+    from app.internal import llm
+    await ai_health.report_degraded(tier, llm.provider_for(model), model, problem, fix, permanent=True)
     if model in _dead_models:
         return
     _dead_models.add(model)
@@ -379,37 +336,3 @@ async def _log_tier_down(where: str, tier: str, model: str, problem: str, fix: s
         f"{where}: {problem} ({model!r}) -- the LLM correlation tier is DISABLED "
         f"and every call is falling back to rules-only. {fix}. API said: {text}"
     )
-
-
-async def tiebreak(rules_decision: dict, llm_decision: dict, ctx: dict) -> dict:
-    """
-    Run the smart tiebreaker (corr_smart_model) when rules and LLM disagree.
-    Falls back to rules_decision on any error.
-    """
-    call_id = ctx["call_id"]
-    try:
-        prompt   = _build_tiebreak_prompt(rules_decision, llm_decision, ctx)
-        raw      = await asyncio.to_thread(_sync_gemini, settings.corr_smart_model, prompt)
-        decision = _parse_response(raw, ctx)
-        _id = (decision["matched_incident"] or {}).get("incident_id", "null")
-        logger.info(
-            f"LLM tiebreak ({settings.corr_smart_model}): call {call_id} → "
-            f"action={decision['action']} incident={_id} "
-            f"reasoning={decision['reasoning']!r}"
-        )
-        await ai_health.report_healthy("correlation_smart")
-        return decision
-    except Exception as e:
-        await _log_llm_failure("LLM tiebreak", "correlation_smart", call_id, settings.corr_smart_model, e)
-        return rules_decision
-
-
-def decisions_agree(rules: dict, llm: dict) -> bool:
-    """True if both decisions agree on action and (when action=="link") on the target incident."""
-    if rules["action"] != llm["action"]:
-        return False
-    if rules["action"] == "link":
-        r_id = (rules.get("matched_incident") or {}).get("incident_id")
-        l_id = (llm.get("matched_incident")   or {}).get("incident_id")
-        return r_id == l_id
-    return True

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] = []
 

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

@@ -65,14 +65,18 @@ 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
-    # Transcript correction (inside transcribe_call). gemini.generate_json only
+    # the worst case is every configured target timing out in turn.
-    # re-issues a request when the model rejects a thinking level — a fast
+    from app.internal.transcription import stt_targets
-    # 400, not a timeout — and google-genai itself does not retry.
+    stt_providers = max(1, len(stt_targets()))
-    correction = settings.gemini_timeout_seconds
+    stt = stt_providers * stt_rounds * openai_attempts * settings.stt_timeout_seconds
+    # Transcript correction and the correlation decision go through
+    # llm.generate_json; the bound depends on which provider serves the model.
+    from app.internal import llm
+    correction = llm.timeout_seconds(settings.transcript_correction_model)
     extraction = openai_attempts * settings.openai_timeout_seconds
     embeddings = n_scenes * openai_attempts * settings.openai_timeout_seconds
-    correlation = 2 * settings.gemini_timeout_seconds  # decide + tiebreak, scene 0
+    correlation = llm.timeout_seconds(settings.corr_cheap_model)  # at most one LLM call, scene 0 (#299)
     return stt + correction + extraction + embeddings + correlation + _NON_AI_SLACK_SECONDS
 
 
@@ -80,24 +84,28 @@ def _derive_min_minutes() -> int:
     return max(_FLOOR_MINUTES, math.ceil(pipeline_worst_case_seconds() / 60))
 
 
-# 21 with the defaults (stt 120s, openai 60s x2 attempts, gemini 60s, 3 scenes).
+# 22 with the defaults (stt 120s, openai 60s x2 attempts for every call, 3 scenes).
 MIN_MINUTES_SINCE_PIPELINE_START = _derive_min_minutes()
 
-# Standard link-only retry budget before a call is tombstoned corr_path="unlinked".
-MAX_SWEEP_ATTEMPTS = 3
-# server-26#115 — a call the consensus LLM-orphan gate parked (llm=orphan vs
-# rules=new, no substance) gets a longer budget. The gate fires before any
-# incident for the job may exist, so the substantive call that would justify
-# linking can land well after the standard ~6 min. Still link-only: a genuinely
-# thin call must not mint an incident, and the rules creation gate would re-orphan
-# it anyway.
-GATED_ORPHAN_SWEEP_ATTEMPTS = 10
 
+def _last_chance(call: dict, now: datetime) -> bool:
+    """
+    server-26#27 — the retry budget is wall-clock, not an attempt count.
 
-def _max_sweep_attempts(call: dict) -> int:
+    The old budget was 3 attempts, which with the #131 hold-back expired ~19
-    if call.get("corr_consensus") == "llm_orphan_gate":
+    minutes after the pipeline started, while the incident the orphan belongs
-        return GATED_ORPHAN_SWEEP_ATTEMPTS
+    to may be opened later than that (`f321b047` was tombstoned three minutes
-    return MAX_SWEEP_ATTEMPTS
+    after its own incident was created). An orphan is now retried on every
+    pass for as long as the scan query can see it, and tombstoned
+    corr_path="unlinked" only on its LAST visible pass: the one after which
+    its ended_at falls out of `recorrelation_scan_minutes`. A call with no
+    parseable ended_at gets no further retries.
+    """
+    ended = _parse_dt(call.get("ended_at"))
+    if ended is None:
+        return True
+    remaining = timedelta(minutes=settings.recorrelation_scan_minutes) - (now - ended)
+    return remaining <= timedelta(minutes=settings.summary_interval_minutes)
 
 
 async def recorrelation_loop() -> None:
@@ -149,9 +157,8 @@ async def _run_sweep_pass() -> None:
         ("status",   "==",  "ended"),
         ("ended_at", ">=",  cutoff),
     ])
-    # corr_path="unlinked" is written after the attempt budget is exhausted.
+    # corr_path="unlinked" is the tombstone written on an orphan's last
-    # Allows a few retries so a welfare-check call can link to an escalation
+    # visible pass (_last_chance); the scan window itself is the budget.
-    # incident that is created a few minutes later, without sweeping 30× forever.
     orphans = [
         c for c in recent_ended
         if not c.get("incident_ids") and not c.get("incident_id")
@@ -163,7 +170,6 @@ async def _run_sweep_pass() -> None:
         # the thin path minutes later and attached to whatever was most recent —
         # a second route into the over-merge the thin fix above addresses.
         and not c.get("skip_reason")
-        and c.get("corr_sweep_count", 0) < _max_sweep_attempts(c)
         and not _pipeline_likely_still_running(c, now)
     ]
 
@@ -173,14 +179,14 @@ async def _run_sweep_pass() -> None:
     logger.info(f"Re-correlation sweep: {len(orphans)} orphaned call(s) to check")
     linked = 0
     for call in orphans:
-        if await _recorrelate_orphan(call):
+        if await _recorrelate_orphan(call, now):
             linked += 1
 
     if linked:
         logger.info(f"Re-correlation sweep: linked {linked}/{len(orphans)} orphaned call(s)")
 
 
-async def _recorrelate_orphan(call: dict) -> bool:
+async def _recorrelate_orphan(call: dict, now: datetime) -> bool:
     """
     Attempt to link a single orphaned call to an existing incident.
     Returns True if a match was found and the call was linked.
@@ -223,12 +229,10 @@ async def _recorrelate_orphan(call: dict) -> bool:
         )
         return True
 
-    # Increment the attempt counter. Once the budget is reached the orphan filter
+    # The attempt count is kept for the admin debug view only; the budget is
-    # above will stop picking this call up, and we write corr_path="unlinked" as
+    # the scan window (_last_chance), after which the call is tombstoned.
-    # a permanent tombstone.
+    update: dict = {"corr_sweep_count": call.get("corr_sweep_count", 0) + 1}
-    attempts = call.get("corr_sweep_count", 0) + 1
+    if _last_chance(call, now):
-    update: dict = {"corr_sweep_count": attempts}
-    if attempts >= _max_sweep_attempts(call):
         update["corr_path"] = "unlinked"
     await fstore.doc_set("calls", call_id, update)
     return False

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

@@ -58,9 +58,10 @@ PREFETCH = 6
 # Estimates, not a bill — nothing in DRB reads a real invoice (server-26#45).
 USD_WHISPER_PER_MIN    = 0.006
 USD_PER_EXTRACTION     = 0.0005   # gpt-4o-mini scene extraction + embedding
-USD_PER_CORRECTION     = 0.0003   # Gemini flash transcript correction
+USD_PER_CORRECTION     = 0.0004   # gpt-4o-mini transcript correction (~1.5k in / 300 out tokens)
 USD_PER_GEOCODE        = 0.005    # Google geocode, roughly one per located scene
-USD_PER_LLM_CORRELATE  = 0.0005   # Gemini flash consensus decision
+USD_PER_LLM_CORRELATE  = 0.0004   # one gpt-4o-mini correlation decision (~2k in / 100 out; ambiguous scenes only)
+_LLM_CONSENSUS = ("llm", "rules_guard", "agreed", "tiebreak", "llm_orphan_gate")
 
 # Fields the pipeline writes onto a call doc. Stripped when a call is copied
 # into the sandbox so the replay recomputes them instead of inheriting the
@@ -73,6 +74,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 +410,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)
@@ -469,9 +474,9 @@ async def _run(run_id: str, org_id: str, calls: list[dict], mode: str,
     fl_token = force_flags(_flags_for(mode))
     ai_failures: list = []
     ai_token = ai_health.collect_sandbox_failures(ai_failures)
-    from app.internal import gemini
+    from app.internal import llm
     usage: dict = {}
-    usage_token = gemini.collect_usage(usage)
+    usage_token = llm.collect_usage(usage)
     try:
         sem = asyncio.Semaphore(PREFETCH)
 
@@ -564,7 +569,7 @@ async def _run(run_id: str, org_id: str, calls: list[dict], mode: str,
         metrics = compute_metrics(incidents, sb_calls)
         metrics["est_cost_usd"] = _running_cost(progress, metrics, mode)
         metrics["ai_failures"] = dict(Counter(f"{f['tier']}: {f['problem']}" for f in ai_failures))
-        metrics["gemini_usage"] = usage
+        metrics["llm_usage"] = usage
     except Exception as e:
         status = "failed"
         errors.append(f"run: {type(e).__name__}: {e}"[:300])
@@ -572,7 +577,7 @@ async def _run(run_id: str, org_id: str, calls: list[dict], mode: str,
         logger.error(f"Replay {run_id} failed: {e}")
     finally:
         ai_health._sandbox_failures.reset(ai_token)
-        gemini.reset_usage(usage_token)
+        llm.reset_usage(usage_token)
         unforce_flags(fl_token)
         fstore.exit_sandbox(sb_token)
         _cancel.discard(run_id)
@@ -636,7 +641,11 @@ def compute_metrics(incidents: list[dict], calls: list[dict]) -> dict:
             corr_path[r.get("corr_path") or "none"] += 1
             consensus[r.get("corr_consensus") or "none"] += 1
     linked = sum(1 for c in calls if c.get("incident_ids"))
-    llm = sum(n for k, n in consensus.items() if k not in ("none", "rules_only"))
+    # Scene decisions that made an LLM call (server-26#299: at most one each).
+    # "rules_fallback" is excluded — the call failed or was unusable, so it
+    # was mostly not billed. "agreed"/"tiebreak" are pre-#299 runs' values
+    # (a tiebreak was two calls, so old runs under-count by their tiebreaks).
+    llm = sum(n for k, n in consensus.items() if k in _LLM_CONSENSUS)
     return {
         "calls": len(calls),
         "calls_linked": linked,

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"(?<![\w-])\d+(?![\w-])")
+
+# "NEVER add information" is the other prompt rule nothing checked. A
+# correction substitutes sound-alikes, so it may shift the word count a little
+# ("Holmes Street and 4th and Rowe" -> "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:
@@ -227,11 +253,11 @@ def build_context_block(context: dict, talkgroup_name: Optional[str]) -> str:
 
 
 def _sync_gemini(model_name: str, prompt: str) -> dict:
-    from app.internal import gemini
+    from app.internal import llm
     # Correction rewrites text against vocabulary; keep a little reasoning
     # ("low") rather than the correlator's "minimal" until a replay shows
     # minimal doesn't hurt it.
-    return gemini.generate_json(model_name, prompt, purpose="correction", thinking_level="low")
+    return llm.generate_json(model_name, prompt, purpose="correction", thinking_level="low")
 
 
 async def correct(
@@ -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.
@@ -255,7 +282,8 @@ async def correct(
     so a correction that only fixed the joined transcript would never reach the
     model on exactly the multi-transmission calls that carry the most content.
     """
-    if not settings.gemini_api_key or not settings.transcript_correction_enabled:
+    from app.internal import llm
+    if not llm.available(settings.transcript_correction_model) or not settings.transcript_correction_enabled:
         return None, None, False
 
     if len((text or "").split()) < MIN_WORDS_FOR_CORRECTION:
@@ -330,6 +358,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 +384,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 +405,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 []

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

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

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

@@ -109,9 +109,9 @@ async def debug_correlation(
             "scene_index":            scene_index,
             "transcript":             scene.get("transcript"),
             "incident_id":            scene.get("incident_id"),
-            # server-26#139: this scene's OWN incident_type/severity, as seen
+            # server-26#139: this scene's OWN incident_type/severity at decision
-            # by _call_is_substanceless at decision time — not the call doc's
+            # time — not the call doc's flat top-level field, which is
-            # flat top-level field, which is last-scene-wins (server-26#96).
+            # last-scene-wins (server-26#96).
             "incident_type":          scene.get("incident_type"),
             "severity":               scene.get("severity"),
             "corr_path":              corr_debug.get("corr_path"),
@@ -126,7 +126,7 @@ async def debug_correlation(
             "corr_llm_reasoning":     corr_debug.get("corr_llm_reasoning"),
             "corr_llm_action":        corr_debug.get("corr_llm_action"),
             "corr_rules_action":      corr_debug.get("corr_rules_action"),
-            "corr_gate_veto":         corr_debug.get("corr_gate_veto"),
+            "corr_band":              corr_debug.get("corr_band"),
         }
 
     def _call_summary(call: dict) -> dict:
@@ -173,20 +173,18 @@ async def debug_correlation(
             "corr_matched_units":    call.get("corr_matched_units"),
             "corr_sweep_count":      call.get("corr_sweep_count"),
             "skip_reason":           call.get("skip_reason"),
-            # LLM consensus tier fields — written by upload.py's
+            # LLM tier fields — written by upload.py's _correlate /
-            # _correlate_with_consensus / llm_correlator.py, but previously
+            # llm_correlator.py, but previously dropped here, making it
-            # dropped here, making it impossible to tell from this endpoint
+            # impossible to tell from this endpoint whether the LLM
-            # whether the LLM correlation tier is actually running (server-26#24).
+            # correlation tier is actually running (server-26#24).
             "corr_consensus":        call.get("corr_consensus"),
             "corr_llm_reasoning":    call.get("corr_llm_reasoning"),
             "corr_llm_action":       call.get("corr_llm_action"),
             "corr_rules_action":     call.get("corr_rules_action"),
-            # server-26#115 — why an llm=orphan/rules=new disagreement escalated
+            # server-26#299 — why the rules marked this decision ambiguous and
-            # to tiebreak instead of being gated (see upload.py's
+            # sent it to the LLM (incident_correlator._ambiguity). Absent on a
-            # _call_is_substanceless). Present only on that disagreement shape;
+            # confident rules decision.
-            # written here specifically so a live measurement window can read
+            "corr_band":             call.get("corr_band"),
-            # the reason instead of reconstructing it by hand from the dump.
-            "corr_gate_veto":        call.get("corr_gate_veto"),
             # server-26#127 — shadow-mode upstream chatter classifier verdict.
             # Written by intelligence.extract_scenes on every transcript that
             # reaches real scene extraction (not on garbage/too-short skips).
@@ -298,7 +296,8 @@ async def debug_correlation(
         orphans_by_tg[tg_key]["count"] += 1
         if not o.get("incident_type") and not o.get("tags"):
             orphans_by_tg[tg_key]["no_type_count"] += 1
-        if (o.get("corr_sweep_count") or 0) >= 3:
+        # Tombstoned by the sweep's wall-clock budget (server-26#27).
+        if o.get("corr_path") == "unlinked":
             orphans_by_tg[tg_key]["sweep_exhausted_count"] += 1
 
     # ── Summary ───────────────────────────────────────────────────────────────
@@ -374,9 +373,8 @@ async def debug_correlation(
         "corr_fit_signal":        _tally(e.get("corr_fit_signal") for e in scene_entries),
         "corr_consensus":         _tally(e.get("corr_consensus") for e in scene_entries),
         "corr_llm_action":        _tally(e.get("corr_llm_action") for e in scene_entries),
-        # server-26#115 — this IS the number the escape-hatch fix exists to
+        # server-26#299 — how often, and why, the LLM tier was consulted.
-        # produce: why each llm=orphan/rules=new call escaped the gate.
+        "corr_band":              _tally(e.get("corr_band") for e in scene_entries),
-        "corr_gate_veto":         _tally(e.get("corr_gate_veto") for e in scene_entries),
         # server-26#127 — shadow-mode chatter classifier. The target
         # population is non-events, which land as orphans or single-call
         # incidents, NOT as a slice of every linked call -- tally `orphans`

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)

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

@@ -1,4 +1,5 @@
 import secrets
+from datetime import datetime, timedelta, timezone
 from typing import Optional
 from fastapi import APIRouter, BackgroundTasks, UploadFile, File, Form, HTTPException, Security
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
@@ -66,6 +67,24 @@ async def upload_call_audio(
     if len(data) > settings.upload_max_bytes:
         raise HTTPException(413, f"File too large (max {settings.upload_max_bytes // (1024*1024)} MB).")
 
+    # server-26#302: the edge node retries uploads (spooled for up to 72h), and
+    # a retry arrives whenever the node lost the response to an upload C2 had
+    # already stored. call_id is the idempotency key: audio already stored
+    # means this upload was already processed, so
+    # re-running Whisper/extraction/correlation would double the AI spend and
+    # double-link the call (the server-26#131 shape). Not atomic: two retries
+    # racing inside one request's lifetime can both pass this check.
+    # Keyed on stored audio only: the pipeline marker alone would refuse a
+    # retry forever (and hand the node a null URL to keep retrying) after a
+    # pipeline crash; such a call is recovered by the sweep or a reprocess.
+    existing = await fstore.doc_get("calls", call_id) or {}
+    if existing.get("audio_gcs_uri"):
+        logger.info(
+            f"Upload for call {call_id} from {node_id} is a retry of one already "
+            f"processed — returning the stored audio, pipeline not re-run"
+        )
+        return {"url": existing.get("audio_gcs_uri"), "already_uploaded": True}
+
     gcs_uri = await upload_audio(data, file.filename or "", call_id=call_id)
 
     if gcs_uri:
@@ -112,99 +131,56 @@ async def upload_call_audio(
     return {"url": gcs_uri}
 
 
-# server-26#115 — the consensus LLM-orphan gate only fires when the call is
+# New-event signals: an LLM "orphan" may not drop a call carrying one when the
-# genuinely substanceless. The earlier version tested `rules_decision["corr_debug"]`
+# rules would open an incident for it (server-26#299). A missed real event is
-# for a "positive signal", but corr_debug is EMPTY at preview time for
+# worse than one extra incident, and the sweep is link-only, so a dropped event
-# action=="new" (corr_path:"new" is written at APPLY time), so that test was
+# never comes back. These are the old #115 gate's vetoes, which existed because
-# always False and the gate dropped real events — a major "extinguishing fire",
+# a looser gate dropped typed and geocoded fire calls.
-# geocoded calls, pursuit updates. The substance test now runs against `ctx`,
+_LLM_ORPHAN_VETO_SEVERITIES = ("moderate", "major")
-# which is fully populated at preview time.
 
 
-def _recent_incident_on_same_talkgroup(ctx: dict) -> bool:
+def _llm_orphan_vetoed(ctx: dict) -> bool:
+    from app.internal.incident_correlator import has_event_substance
+    return bool(
+        ctx.get("reassignment") or ctx.get("incident_type")
+        or ctx.get("call_severity") in _LLM_ORPHAN_VETO_SEVERITIES
+        or has_event_substance(ctx)
+    )
+
+
+def _merge_llm_decision(rules: dict, llm: dict, ctx: dict) -> tuple[dict, str]:
     """
-    True when a recent incident is running on this call's own system +
+    The decision to commit for an ambiguous scene, and the corr_consensus
-    talkgroup, within `settings.tg_dispatch_thin_idle_minutes` (5 min) —
+    value that says who made it: "llm", or "rules_guard" when the LLM's
-    applied uniformly regardless of the talkgroup's name (server-26#134).
+    orphan was refused for a call with a strong new-event signal.
-    Covers "unit dispatched, thin ack 10-30s later": the ack has no
-    substance of its own but plainly belongs to the job just opened.
-
-    Reads ctx["recent"] (the rules engine's own candidate list — no extra
-    Firestore read). That list is status=="active" incidents only, so an
-    already-resolved or capacity-capped same-talkgroup incident won't be
-    seen here even if chronologically recent (server-26#115, unresolved —
-    would need a dedicated non-status-filtered query).
-
-    Whether this limitation explains the 2/24 unexplained gate misses in the
-    window #3 measurement is UNANSWERED, not confirmed either way — a prior
-    pass here claimed a "confirmed explanation" for both that turned out to
-    be self-contradictory. Read `corr_gate_veto` (written to corr_debug on
-    every escalation of this exact disagreement shape — see the caller) in
-    the next measurement window instead of guessing from the raw dump again.
-    # TODO(server-26#115): add a talkgroup-scoped incident lookup (any
-    # status, no capacity filter) if a future measurement window pins a real
-    # gate miss on a resolved/capped same-talkgroup incident.
     """
-    from app.internal.incident_correlator import _idle_gate_minutes
+    if llm["action"] == "orphan" and rules["action"] == "new" and _llm_orphan_vetoed(ctx):
-
+        return rules, "rules_guard"
-    tg_id = ctx.get("talkgroup_id")
+    if (
-    system_id = ctx.get("system_id")
+        llm["action"] == rules["action"] == "link"
-    if tg_id is None or not system_id:
+        and llm["matched_incident"]["incident_id"] == rules["matched_incident"]["incident_id"]
-        return False
+    ):
-    tg_str = str(tg_id)
+        # The LLM confirmed the rules' own link: keep the rules' path/fit
-    now = ctx.get("now") or clock.now()
+        # signal so corr_path histograms stay comparable across runs.
-    idle_limit = settings.tg_dispatch_thin_idle_minutes
+        return rules, "llm"
-    for inc in ctx.get("recent") or []:
+    final = {
-        if system_id not in (inc.get("system_ids") or []):
+        "action":           llm["action"],
-            continue
+        "matched_incident": llm["matched_incident"],
-        if tg_str not in (inc.get("talkgroup_ids") or []):
+        "incident_type":    llm["incident_type"],
-            continue
+        "corr_debug":       dict(llm.get("corr_debug") or {}),
-        if _idle_gate_minutes(inc, now) <= idle_limit:
+    }
-            return True
+    if llm["action"] == "link":
-    return False
+        # The rules' corr_path/fit signal described a different judgement
+        # (possibly a different incident); keep it under corr_rules_* only.
+        final["corr_debug"]["corr_path"] = "llm"
+    elif llm["action"] == "orphan" and rules["action"] == "link":
+        # Tombstone: the rules-only sweep would otherwise re-commit the very
+        # unverified unit link the LLM just refused.
+        final["corr_debug"]["corr_path"] = "llm_orphan"
+    return final, "llm"
 
 
-def _call_is_substanceless(ctx: dict) -> tuple[bool, Optional[str]]:
+async def _correlate(
-    """
-    True when the call carries nothing that marks it as a real event:
-      • no resolved incident_type and not a reassignment, AND
-      • severity is not moderate/major, AND
-      • no vehicle, geocode or tag (incident_correlator.has_event_substance —
-        the same predicate the incident-creation gate uses), AND
-      • no recent incident already running on the same talkgroup.
-    Only then may the LLM-orphan gate drop the call without a tiebreak.
-
-    Returns (substanceless, veto_reason). veto_reason names whichever
-    condition kept the tiebreak alive ("type" | "reassignment" | "severity" |
-    "substance" | "recent_tg"), or None when the call is substanceless. The
-    caller writes this into corr_debug on the escalation path so a live
-    measurement window can see *why* each llm=orphan/rules=new call escaped
-    the gate instead of inferring it after the fact from the raw dump —
-    exactly the guesswork that produced a wrong "confirmed explanation" for
-    2 window-#3 misses on the first pass of this fix.
-    """
-    from app.internal import incident_correlator
-
-    # The incident-creation gate skips the has_event_substance check entirely
-    # when a type resolved (incident_correlator._run_decision ~:1397), so a
-    # typed call — fire/medical/etc. — opens an incident on substance we do not
-    # re-check here. reassignment=True is dispatch pulling a unit onto a NEW
-    # job (units are blanked at :296 for exactly that reason): the strongest
-    # new-incident signal in the pipeline. Either one means "keep the tiebreak".
-    if ctx.get("incident_type"):
-        return False, "type"
-    if ctx.get("reassignment"):
-        return False, "reassignment"
-    if (ctx.get("call_severity") or "routine") in ("moderate", "major"):
-        return False, "severity"
-    if incident_correlator.has_event_substance(ctx):
-        return False, "substance"
-    if _recent_incident_on_same_talkgroup(ctx):
-        return False, "recent_tg"
-    return True, None
-
-
-async def _correlate_with_consensus(
     call_id: str,
     node_id: str,
     system_id: Optional[str],
@@ -222,19 +198,22 @@ async def _correlate_with_consensus(
     severity: Optional[str] = None,
     transcript: Optional[str] = None,
     scene_index: int = 0,
-) -> Optional[str]:
+) -> tuple[Optional[str], dict]:
     """
-    Consensus correlator: runs the rules engine and the cheap LLM in sequence.
+    Correlate one scene (server-26#299): the rules engine decides; only when
-    If they agree the rules decision is committed directly.
+    it marks its decision ambiguous is ONE cheap LLM call made, and a usable
-    If they disagree a smarter tiebreaker LLM makes the final call.
+    LLM answer is committed instead. Returns (incident_id, committed decision).
 
-    Falls back to rules-only when GEMINI_API_KEY is absent, the call is
+    corr_consensus on the call doc says who decided:
-    content-free (thin), or any LLM call fails.
+      "rules_only"     — confident rules decision, no LLM call
+      "llm"            — ambiguous, the LLM's answer was committed
+      "rules_fallback" — ambiguous, LLM unavailable/failed/unusable → rules
+      "rules_guard"    — ambiguous, LLM said orphan on a call with a strong
+                         new-event signal → rules "new" kept
+    and corr_band names why it was ambiguous.
 
-    ``scene_index`` (server-26#96) — which scene of the call this is, from the
+    ``scene_index`` (server-26#96) — which scene of the call this is, so the
-    caller's ``enumerate(scenes)`` loop. Threaded through so the call doc's
+    call doc's per-scene ``scenes`` map records this scene's own decision.
-    per-scene ``scenes`` map records this scene's own corr_debug/transcript
-    instead of colliding with every other scene's write on the flat fields.
     """
     from app.internal import incident_correlator, llm_correlator
 
@@ -247,68 +226,119 @@ async def _correlate_with_consensus(
         embedding=embedding, severity=severity, transcript=transcript,
         scene_index=scene_index,
     )
-    ctx             = preview["ctx"]
+    ctx   = preview["ctx"]
-    rules_decision  = preview["decision"]
+    rules = preview["decision"]
+    band  = rules.get("ambiguous")
 
-    llm_decision = await llm_correlator.decide(call_id, ctx)
+    if not band:
+        rules["corr_debug"]["corr_consensus"] = "rules_only"
+        return await incident_correlator.apply_correlation(preview), rules
 
-    if llm_decision is None:
+    llm = await llm_correlator.decide(call_id, ctx)
-        # LLM unavailable, skipped (thin call), or errored — rules wins.
+    if llm is None:
-        rules_decision["corr_debug"]["corr_consensus"] = "rules_only"
+        final, consensus = rules, "rules_fallback"
-        return await incident_correlator.apply_correlation(preview)
+    else:
-
+        final, consensus = _merge_llm_decision(rules, llm, ctx)
-    if llm_correlator.decisions_agree(rules_decision, llm_decision):
+    final["corr_debug"].update({
-        rules_decision["corr_debug"]["corr_consensus"] = "agreed"
+        "corr_consensus":    consensus,
-        rules_decision["corr_debug"]["corr_llm_reasoning"] = llm_decision.get("reasoning", "")
+        "corr_band":         band,
-        return await incident_correlator.apply_correlation(preview)
+        "corr_rules_action": rules["action"],
-
+    })
-    # server-26#115 — LLM-orphan gate.
+    if llm is not None:
-    # When the cheap LLM says `orphan`, the rules engine says `new`, and the call
+        final["corr_debug"]["corr_llm_action"]    = llm["action"]
-    # is genuinely substanceless (routine severity, no vehicle/geocode/tag, and
+        final["corr_debug"]["corr_llm_reasoning"] = llm.get("reasoning", "")
-    # no incident already running on this talkgroup), resolve to `orphan` and DO
-    # NOT pay for the smart tiebreaker. A bare rules `new` there means only
-    # "nothing to link to" — trivially true for radio housekeeping (check-ins,
-    # roll call, 10-8/10-98) — and the tiebreaker rubber-stamped it ~21/21 of the
-    # time on exactly this disagreement (CORRELATION_REVIEW_0907b.md). Any real
-    # signal (severity, coords, tags, a live same-talkgroup incident) still
-    # escalates, so an event the LLM misreads as orphan is not lost.
-    is_orphan_vs_new = llm_decision["action"] == "orphan" and rules_decision["action"] == "new"
-    substanceless, gate_veto_reason = _call_is_substanceless(ctx) if is_orphan_vs_new else (False, None)
-    if is_orphan_vs_new and substanceless:
-        logger.info(
-            f"Consensus gate for call {call_id}: llm=orphan vs rules=new and call "
-            f"is substanceless — resolving orphan, skipping tiebreak"
-        )
-        gated = {
-            "action": "orphan",
-            "matched_incident": None,
-            "incident_type": None,
-            "corr_debug": dict(rules_decision.get("corr_debug") or {}),
-        }
-        gated["corr_debug"].update({
-            "corr_consensus":     "llm_orphan_gate",
-            "corr_rules_action":  rules_decision["action"],
-            "corr_llm_action":    llm_decision["action"],
-            "corr_llm_reasoning": llm_decision.get("reasoning", ""),
-        })
-        return await incident_correlator.apply_correlation({"decision": gated, "ctx": ctx})
-
-    # Disagree — escalate to the smarter tiebreaker.
     logger.info(
-        f"Consensus disagreement for call {call_id}: "
+        f"Correlation for call {call_id} scene {scene_index}: ambiguous ({band}) — "
-        f"rules={rules_decision['action']} vs llm={llm_decision['action']} — tiebreak"
+        f"rules={rules['action']} llm={llm['action'] if llm else 'none'} → {consensus}"
     )
-    final = await llm_correlator.tiebreak(rules_decision, llm_decision, ctx)
+    return await incident_correlator.apply_correlation({"decision": final, "ctx": ctx}), final
-    final["corr_debug"]["corr_consensus"]    = "tiebreak"
+
-    final["corr_debug"]["corr_rules_action"] = rules_decision["action"]
+
-    final["corr_debug"]["corr_llm_action"]   = llm_decision["action"]
+def _may_resolve(decision: dict) -> bool:
-    if is_orphan_vs_new:
+    """
-        # server-26#115 — record *why* the llm=orphan/rules=new gate stood
+    server-26#88: honour a scene's `resolved` flag only on a link the
-        # down instead of leaving a future measurement window to guess it
+    correlator was confident about, or on the incident this very scene opened.
-        # from the raw dump (which produced a wrong "confirmed explanation"
+    A scene that attached by thin recency, or whose placement was ambiguous
-        # for 2/24 misses the first time around).
+    (an LLM pick, or a rules fallback in the ambiguous band), has no business
-        final["corr_debug"]["corr_gate_veto"] = gate_veto_reason
+    closing an incident it may not belong to — that closed live incidents
-    return await incident_correlator.apply_correlation({"decision": final, "ctx": ctx})
+    while the event was still on the air.
+    """
+    if decision.get("action") == "new":
+        return True
+    debug = decision.get("corr_debug") or {}
+    return (
+        decision.get("action") == "link"
+        and debug.get("corr_consensus") == "rules_only"
+        and debug.get("corr_path") != "fast/thin"
+    )
+
+
+async def _correlate_scenes(
+    call_id: str,
+    node_id: str,
+    system_id: Optional[str],
+    talkgroup_id: Optional[int],
+    talkgroup_name: Optional[str],
+    scenes: list[dict],
+) -> tuple[list[str], list[str]]:
+    """
+    Correlate each scene of one call independently — a single recording can
+    carry several events on a busy channel — and apply each scene's LLM
+    closure. Returns (incident_ids, merged tags). Shared by the live pipeline,
+    replay (via _extract_and_correlate) and the transcript-PATCH path.
+    """
+    from app.internal import incident_correlator
+
+    incident_ids: list[str] = []
+    all_tags: list[str] = []
+    for scene_index, scene in enumerate(scenes):
+        all_tags.extend(scene["tags"])
+        # When dispatch is pulling a unit to a NEW call (reassignment), suppress
+        # unit overlap so the new scene doesn't chain into the unit's previous
+        # incident. parse_llm_bool (#88): a quoted "false" is not True.
+        is_reassignment = incident_correlator.parse_llm_bool(scene.get("reassignment"))
+        corr_units = [] if is_reassignment else scene.get("units")
+        incident_id, decision = await _correlate(
+            call_id=call_id,
+            node_id=node_id,
+            system_id=system_id,
+            talkgroup_id=talkgroup_id,
+            talkgroup_name=talkgroup_name,
+            tags=scene["tags"],
+            incident_type=scene["incident_type"],
+            location=scene["location"],
+            location_coords=scene["location_coords"],
+            units=corr_units,
+            vehicles=scene.get("vehicles"),
+            cleared_units=scene.get("cleared_units"),
+            reassignment=is_reassignment,
+            embedding=scene.get("embedding"),
+            severity=scene.get("severity"),
+            transcript=scene.get("transcript"),
+            scene_index=scene_index,
+        )
+        if incident_id and incident_id not in incident_ids:
+            incident_ids.append(incident_id)
+        if incident_id and incident_correlator.parse_llm_bool(scene.get("resolved")):
+            if not _may_resolve(decision):
+                logger.info(
+                    f"Ignoring closure from call {call_id} scene {scene_index}: link into "
+                    f"{incident_id} was not confident (server-26#88)"
+                )
+                continue
+            await fstore.doc_set("incidents", incident_id, {
+                "status": "resolved",
+                "resolved_at": clock.now().isoformat(),
+                "resolved_via": "llm_closure",
+                # One transmission read as "it's over" ("transport complete")
+                # closed the whole 09-22 bridge MVA at 14:44 and its next 56
+                # calls opened a second incident (server-26#170). Inferred from
+                # a single call, so provisional, like a timer close.
+                "reopenable": True,
+            })
+            await incident_correlator.maybe_resolve_parent(incident_id)
+            logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)")
+    return incident_ids, all_tags
 
 
 async def _resolve_flags(system_id: Optional[str]):
@@ -335,7 +365,7 @@ async def _run_extraction_pipeline(
     preserve_transcript_correction: bool = False,
 ) -> None:
     """Run steps 2-4 of the intelligence pipeline using an existing transcript."""
-    from app.internal import intelligence, incident_correlator, alerter
+    from app.internal import intelligence, alerter
 
     flags, _flag = await _resolve_flags(system_id)
 
@@ -353,45 +383,9 @@ async def _run_extraction_pipeline(
         )
 
         # Step 3: Correlate each scene to an incident independently.
-        # server-26#96: scene_index is threaded through so each scene's
+        incident_ids, all_tags = await _correlate_scenes(
-        # corr_debug/transcript lands in its own entry of the call doc's
+            call_id, node_id, system_id, talkgroup_id, talkgroup_name, scenes,
-        # `scenes` map instead of clobbering every other scene's write.
+        )
-        for scene_index, scene in enumerate(scenes):
-            all_tags.extend(scene["tags"])
-            # When dispatch is pulling a unit to a NEW call (reassignment), suppress unit
-            # overlap so the new scene doesn't chain into the unit's previous incident.
-            is_reassignment = bool(scene.get("reassignment"))
-            corr_units = [] if is_reassignment else scene.get("units")
-            incident_id = await _correlate_with_consensus(
-                call_id=call_id,
-                node_id=node_id,
-                system_id=system_id,
-                talkgroup_id=talkgroup_id,
-                talkgroup_name=talkgroup_name,
-                tags=scene["tags"],
-                incident_type=scene["incident_type"],
-                location=scene["location"],
-                location_coords=scene["location_coords"],
-                units=corr_units,
-                vehicles=scene.get("vehicles"),
-                cleared_units=scene.get("cleared_units"),
-                reassignment=is_reassignment,
-                embedding=scene.get("embedding"),
-                severity=scene.get("severity"),
-                transcript=scene.get("transcript"),
-                scene_index=scene_index,
-            )
-            if incident_id and incident_id not in incident_ids:
-                incident_ids.append(incident_id)
-            if scene["resolved"] and incident_id:
-                await fstore.doc_set("incidents", incident_id, {
-                    "status": "resolved",
-                    "resolved_at": clock.now().isoformat(),
-                    "resolved_via": "llm_closure",
-                    "reopenable": True,  # provisional, see _extract_and_correlate
-                })
-                await incident_correlator.maybe_resolve_parent(incident_id)
-                logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)")
     else:
         scope = "globally" if not flags["correlation_enabled"] else f"system {system_id}"
         logger.info(f"Correlation disabled ({scope}) — skipping scene extraction and correlation for call {call_id} (reprocess)")
@@ -430,7 +424,7 @@ async def _extract_and_correlate(
     copy of it that can drift. Caller owns the correlation feature-flag check
     and alerting.
     """
-    from app.internal import intelligence, incident_correlator
+    from app.internal import intelligence
 
     # Step 2: Scene detection + intelligence extraction
     if scenes is None:
@@ -443,50 +437,9 @@ async def _extract_and_correlate(
             )
 
     # Step 3: Correlate each scene independently.
-    # A single recording can produce multiple incidents on a busy channel.
+    incident_ids, all_tags = await _correlate_scenes(
-    incident_ids: list[str] = []
+        call_id, node_id, system_id, talkgroup_id, talkgroup_name, scenes,
-    all_tags: list[str] = []
+    )
-    # server-26#96: scene_index is threaded through so each scene's
-    # corr_debug/transcript lands in its own entry of the call doc's
-    # `scenes` map instead of clobbering every other scene's write.
-    for scene_index, scene in enumerate(scenes):
-        all_tags.extend(scene["tags"])
-        is_reassignment = bool(scene.get("reassignment"))
-        corr_units = [] if is_reassignment else scene.get("units")
-        incident_id = await _correlate_with_consensus(
-            call_id=call_id,
-            node_id=node_id,
-            system_id=system_id,
-            talkgroup_id=talkgroup_id,
-            talkgroup_name=talkgroup_name,
-            tags=scene["tags"],
-            incident_type=scene["incident_type"],
-            location=scene["location"],
-            location_coords=scene["location_coords"],
-            units=corr_units,
-            vehicles=scene.get("vehicles"),
-            cleared_units=scene.get("cleared_units"),
-            reassignment=is_reassignment,
-            embedding=scene.get("embedding"),
-            severity=scene.get("severity"),
-            transcript=scene.get("transcript"),
-            scene_index=scene_index,
-        )
-        if incident_id and incident_id not in incident_ids:
-            incident_ids.append(incident_id)
-        if scene["resolved"] and incident_id:
-            await fstore.doc_set("incidents", incident_id, {
-                "status": "resolved",
-                "resolved_at": clock.now().isoformat(),
-                "resolved_via": "llm_closure",
-                # One transmission read as "it's over" ("transport complete")
-                # closed the whole 09-22 bridge MVA at 14:44 and its next 56
-                # calls opened a second incident (server-26#170). Inferred from
-                # a single call, so provisional, like a timer close.
-                "reopenable": True,
-            })
-            await incident_correlator.maybe_resolve_parent(incident_id)
-            logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)")
 
     # Correlator also runs for calls with no scenes (unclassified) to attempt
     # talkgroup-based linking even when no transcript could be produced.
@@ -501,8 +454,11 @@ 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
-            incident_id = await _correlate_with_consensus(
+        # 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(
                 call_id=call_id,
                 node_id=node_id,
                 system_id=system_id,
@@ -512,6 +468,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)
@@ -522,6 +483,33 @@ async def _extract_and_correlate(
     return incident_ids, all_tags, scenes
 
 
+def _late_upload_anchor(call_doc: Optional[dict]) -> Optional[datetime]:
+    """
+    The call's own started_at when its upload is late (server-26#301), else
+    None. "Late" is `late_upload_anchor_minutes` behind the real clock: an
+    on-time call keeps being judged against now exactly as before, so the live
+    path's recency windows are unchanged; only a spooled/retried upload is
+    back-dated. A started_at in the future or unparseable is ignored.
+    """
+    raw = (call_doc or {}).get("started_at")
+    if not raw:
+        return None
+    if isinstance(raw, datetime):   # mqtt_handler stores a native timestamp
+        started = raw
+    else:
+        try:
+            started = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
+        except ValueError:
+            return None
+    if started.tzinfo is None:
+        started = started.replace(tzinfo=timezone.utc)
+    lag = clock.now() - started
+    if lag > timedelta(minutes=settings.late_upload_anchor_minutes):
+        logger.info(f"Late upload: call started {lag} ago — correlating as of {started.isoformat()}")
+        return started
+    return None
+
+
 async def _run_intelligence_pipeline(
     call_id: str,
     node_id: str,
@@ -586,7 +574,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}"
@@ -596,15 +584,27 @@ async def _run_intelligence_pipeline(
     incident_ids: list[str] = []
     all_tags: list[str] = []
     if _flag("correlation_enabled"):
-        incident_ids, all_tags, _ = await _extract_and_correlate(
+        # server-26#301: a late upload (the node spools and retries for up to
-            call_id=call_id,
+        # 72h) is correlated as of when the call happened, not as of now — the
-            node_id=node_id,
+        # same clock pin replay uses, so every recency gate and every
-            system_id=system_id,
+        # started_at/updated_at/resolved_at it writes is on the call's own
-            talkgroup_id=talkgroup_id,
+        # timeline. Without it a call hours late linked into whatever incident
-            talkgroup_name=talkgroup_name,
+        # was live on its talkgroup now, or opened one at the wrong time.
-            transcript=transcript,
+        anchor = _late_upload_anchor(_call_doc)
-            segments=segments,
+        tok = clock.pin(anchor) if anchor else None
-        )
+        try:
+            incident_ids, all_tags, _ = await _extract_and_correlate(
+                call_id=call_id,
+                node_id=node_id,
+                system_id=system_id,
+                talkgroup_id=talkgroup_id,
+                talkgroup_name=talkgroup_name,
+                transcript=transcript,
+                segments=segments,
+            )
+        finally:
+            if tok is not None:
+                clock.unpin(tok)
     else:
         scope = "globally" if not flags["correlation_enabled"] else f"system {system_id}"
         logger.info(f"Correlation disabled ({scope}) — skipping scene extraction and correlation for call {call_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_ai_flag_gating.py

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

drb-c2-core/tests/test_ai_health.py

@@ -136,15 +136,15 @@ async def test_transient_failure_alerts_once_threshold_crossed():
 async def test_recovery_alerts_once_after_an_alerted_episode():
     with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook:
         await ai_health.report_degraded(
-            "correlation_smart", "gemini", "gemini-1.5-pro",
+            "correlation_cheap", "gemini", "gemini-3.6-flash",
             "the Gemini account is out of credit", "top up billing", permanent=True,
         )
         webhook.reset_mock()
 
-        await ai_health.report_healthy("correlation_smart")
+        await ai_health.report_healthy("correlation_cheap")
 
     webhook.assert_awaited_once()
-    state = ai_health.snapshot()["correlation_smart"]
+    state = ai_health.snapshot()["correlation_cheap"]
     assert state["degraded"] is False
     assert state["alerted"] is False
     assert state["consecutive_failures"] == 0

drb-c2-core/tests/test_ai_timeouts.py

@@ -153,8 +153,10 @@ def test_timeout_messages_classify_as_transient():
 
 
 def _corr_ctx():
+    from datetime import datetime, timezone
     return {"call_id": "c1", "is_thin_call": False, "recent": [{"incident_id": "i1"}],
-            "all_active": [], "call_units": ["12"], "call_cleared": [], "tags": ["mva"]}
+            "all_active": [], "call_units": ["12"], "call_cleared": [], "tags": ["mva"],
+            "now": datetime.now(timezone.utc)}
 
 
 @pytest.mark.asyncio
@@ -162,35 +164,22 @@ async def test_gemini_timeout_in_decide_falls_back_to_rules():
     def hang(model, prompt):
         raise httpx.ReadTimeout("The read operation timed out")
 
-    with patch.object(settings, "gemini_api_key", "test-key"), \
+    with patch.object(settings, "openai_api_key", "test-key"), \
-         patch.object(llm_correlator, "_build_decide_prompt", lambda ctx: "p"), \
+         patch.object(llm_correlator, "_build_decide_prompt", lambda ctx, cands: "p"), \
          patch.object(llm_correlator, "_sync_gemini", hang):
-        assert await llm_correlator.decide("c1", _corr_ctx()) is None  # None == rules_only
+        assert await llm_correlator.decide("c1", _corr_ctx()) is None  # None == keep the rules decision
     state = ai_health.snapshot()["correlation_cheap"]
     assert state["consecutive_failures"] == 1
     assert state["permanent"] is False
 
 
-@pytest.mark.asyncio
-async def test_gemini_timeout_in_tiebreak_keeps_the_rules_decision():
-    def hang(model, prompt):
-        raise httpx.ReadTimeout("The read operation timed out")
-
-    rules = {"action": "new", "matched_incident": None, "corr_debug": {}}
-    with patch.object(llm_correlator, "_build_tiebreak_prompt", lambda r, l, c: "p"), \
-         patch.object(llm_correlator, "_sync_gemini", hang):
-        out = await llm_correlator.tiebreak(rules, {"action": "orphan"}, _corr_ctx())
-    assert out is rules
-    assert ai_health.snapshot()["correlation_smart"]["permanent"] is False
-
-
 @pytest.mark.asyncio
 async def test_gemini_timeout_in_transcript_correction_keeps_raw_text():
     def hang(model, prompt):
         raise httpx.ReadTimeout("The read operation timed out")
 
     text = "engine 5 respond to a structure fire at 12 main street smoke showing"
-    with patch.object(settings, "gemini_api_key", "test-key"), \
+    with patch.object(settings, "openai_api_key", "test-key"), \
          patch.object(transcript_correction, "resolve_context", AsyncMock(return_value={})), \
          patch.object(transcript_correction, "_sync_gemini", hang):
         assert await transcript_correction.correct("c1", text) == (None, None, False)
@@ -233,8 +222,8 @@ def test_worst_case_bound_follows_the_settings():
          patch.object(settings, "gemini_timeout_seconds", 20.0), \
          patch.object(settings, "stt_retry_on_degenerate", True):
         bound = recorrelation_sweep.pipeline_worst_case_seconds(n_scenes=2)
-    # stt 2x2x100 + correction 20 + extraction 2x10 + embeds 2x2x10 + decide/tiebreak 2x20 + slack
+    # stt 2x2x100 + correction 20 + extraction 2x10 + embeds 2x2x10 + one correlation call 20 + slack
-    assert bound == 400 + 20 + 20 + 40 + 40 + recorrelation_sweep._NON_AI_SLACK_SECONDS
+    assert bound == 400 + 20 + 20 + 40 + 20 + recorrelation_sweep._NON_AI_SLACK_SECONDS
 
 
 def test_hold_back_covers_the_worst_case_and_never_drops_below_old_floor():

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"]

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", AsyncMock(return_value=(None, 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", AsyncMock(return_value=(None, 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_clearance_tracking.py

@@ -53,12 +53,11 @@ def test_a_unit_never_on_the_incident_cannot_close_it():
 
 
 def test_clearance_only_call_skips_the_llm():
-    from app.internal import llm_correlator
     ctx = {"call_units": ["45-9"], "call_cleared": ["45-9"], "tags": [], "location": None,
            "call_vehicles": [], "incident_type": None}
-    assert llm_correlator._is_clearance_only(ctx)
+    assert ic.is_clearance_only(ctx)
-    assert not llm_correlator._is_clearance_only({**ctx, "tags": ["mva"]})
+    assert not ic.is_clearance_only({**ctx, "tags": ["mva"]})
-    assert not llm_correlator._is_clearance_only({**ctx, "call_cleared": []})
+    assert not ic.is_clearance_only({**ctx, "call_cleared": []})
 
 
 def test_only_numbered_units_hold_an_incident_open():

drb-c2-core/tests/test_consensus_gate.py

@@ -1,430 +0,0 @@
-"""
-server-26#115 — two consensus-quality fixes.
-
-Fix 1 (routers/upload.py): when the cheap LLM says `orphan`, the rules engine
-says `new`, and the call is genuinely SUBSTANCELESS (routine severity, no
-vehicle/geocode/tag, and no incident already running on the same talkgroup),
-resolve to `orphan` and DO NOT pay for the smart tiebreaker. Radio housekeeping
-(unit check-ins, roll call, 10-8/10-98) was being promoted to incidents because
-the tiebreaker rubber-stamped the rules `new` ~21/21 of the time
-(CORRELATION_REVIEW_0907b.md).
-
-The substance test runs against `ctx` (fully populated at preview time), NOT
-against `rules_decision["corr_debug"]` — that dict is EMPTY at preview time for
-action=="new" (corr_path:"new" is written at APPLY time), so the first version of
-this gate fired on real events (a `major` "extinguishing fire", geocoded calls,
-pursuit updates).
-
-Fix 2 (incident_correlator.py): the `location` correlation path linked on a bare
-sub-`location_proximity_km` (0.5 km) distance alone, taking whichever incident
-came first in an unsorted `recent`. In a dense village two unrelated events
-routinely geocode that close. A `location` link now needs unit overlap with the
-candidate OR a distance under a tighter bar, and picks the NEAREST qualifying
-candidate. A unit-overlap location link is tagged `location_unit_overlap` so it
-does not merge into the fast path's bucket in the admin fit-signal histogram.
-"""
-from datetime import datetime, timedelta, timezone
-from unittest.mock import AsyncMock, patch
-
-import pytest
-
-from app.routers import upload
-from app.internal.incident_correlator import _run_decision, has_event_substance
-
-NOW = datetime(2026, 9, 7, 21, 30, 0, tzinfo=timezone.utc)
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Fix 1 — the LLM-orphan gate in _correlate_with_consensus
-# ─────────────────────────────────────────────────────────────────────────────
-
-def _preview(action, corr_debug=None, ctx=None):
-    base_ctx = {"call_id": "call-1"}
-    if ctx:
-        base_ctx.update(ctx)
-    return {
-        "decision": {
-            "action": action,
-            "matched_incident": None,
-            "incident_type": "other" if action == "new" else None,
-            "corr_debug": {} if corr_debug is None else dict(corr_debug),
-        },
-        "ctx": base_ctx,
-    }
-
-
-def _llm(action, reasoning="—"):
-    md = {"incident_id": "inc-1"} if action == "link" else None
-    return {"action": action, "matched_incident": md, "reasoning": reasoning}
-
-
-async def _run_consensus(preview, llm_decision):
-    tiebreak_result = {
-        "action": "new", "matched_incident": None, "incident_type": "other",
-        "corr_debug": {}, "reasoning": "tb",
-    }
-    with patch("app.internal.incident_correlator.preview_correlation",
-               new=AsyncMock(return_value=preview)), \
-         patch("app.internal.incident_correlator.apply_correlation",
-               new=AsyncMock(return_value="incident-x")) as m_apply, \
-         patch("app.internal.llm_correlator.decide",
-               new=AsyncMock(return_value=llm_decision)), \
-         patch("app.internal.llm_correlator.tiebreak",
-               new=AsyncMock(return_value=tiebreak_result)) as m_tiebreak:
-        await upload._correlate_with_consensus(
-            call_id="call-1", node_id="n1", system_id="sys-1",
-            talkgroup_id=9048, talkgroup_name="Dispatch", tags=[],
-            incident_type=None, location=None, location_coords=None,
-        )
-    return m_apply, m_tiebreak
-
-
-async def test_substanceless_no_recent_same_tg_incident_gates_without_tiebreak():
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}), _llm("orphan", "unit check-in, not an incident"),
-    )
-    m_tiebreak.assert_not_called()
-    m_apply.assert_called_once()
-    gated = m_apply.call_args[0][0]["decision"]
-    assert gated["action"] == "orphan"
-    dbg = gated["corr_debug"]
-    assert dbg["corr_consensus"] == "llm_orphan_gate"
-    assert dbg["corr_consensus"] != "tiebreak"
-    assert dbg["corr_rules_action"] == "new"
-    assert dbg["corr_llm_action"] == "orphan"
-    assert dbg["corr_llm_reasoning"] == "unit check-in, not an incident"
-
-
-@pytest.mark.parametrize("severity", ["moderate", "major"])
-async def test_moderate_or_major_severity_is_not_gated(severity):
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx={"call_severity": severity}), _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-
-
-async def test_routine_severity_alone_still_gates():
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx={"call_severity": "routine"}), _llm("orphan"),
-    )
-    m_tiebreak.assert_not_called()
-    assert m_apply.call_args[0][0]["decision"]["action"] == "orphan"
-
-
-async def test_call_with_coords_is_not_gated():
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx={"coords": {"lat": 41.15, "lng": -73.86}}),
-        _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-
-
-async def test_call_with_tags_is_not_gated():
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx={"tags": ["structure-fire"]}), _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-
-
-async def test_call_with_vehicles_is_not_gated():
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx={"call_vehicles": ["red sedan"]}), _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-
-
-async def test_call_with_resolved_incident_type_is_not_gated():
-    # The creation gate skips has_event_substance when a type resolved, so a
-    # typed call (fire/medical/…) opens an incident on substance the gate does
-    # not re-check — it must keep the tiebreak, not be dropped.
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx={"incident_type": "fire"}), _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-
-
-async def test_reassignment_call_is_not_gated():
-    # reassignment=True is dispatch pulling a unit onto a NEW job (units are
-    # blanked for exactly that reason) — the strongest new-incident signal.
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx={"reassignment": True}), _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-
-
-async def test_recent_incident_on_same_talkgroup_is_not_gated():
-    ctx = {
-        "system_id": "sys-1",
-        "talkgroup_id": 9048,
-        "talkgroup_name": "Dispatch",
-        "now": NOW,
-        "recent": [{
-            "incident_id": "inc-live",
-            "system_ids": ["sys-1"],
-            "talkgroup_ids": ["9048"],
-            "updated_at": (NOW - timedelta(minutes=1)).isoformat(),
-        }],
-    }
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx=ctx), _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-
-
-# server-26#115 window #3 (CORRELATION_REVIEW_0912.md): the escape hatch used
-# to treat ANY same-talkgroup incident inside the 2h correlation_window_hours
-# as "recent", which on a busy dispatch channel (3-13 incidents/2h) was
-# satisfied almost unconditionally — the gate fired 0/24 times against its own
-# target shape. It now only counts an incident as recent within
-# settings.tg_dispatch_thin_idle_minutes (5 min), applied uniformly regardless
-# of the talkgroup's name (owner correction, 2026-09-13 — see
-# test_channel_name_does_not_affect_the_window below for why the dichotomy
-# this originally had with incident_correlator's fast/thin idle selection was
-# removed here).
-
-async def test_recent_same_tg_incident_inside_new_short_window_still_escapes_gate():
-    ctx = {
-        "system_id": "sys-1",
-        "talkgroup_id": 9048,
-        "talkgroup_name": "Dispatch",
-        "now": NOW,
-        "recent": [{
-            "incident_id": "inc-live",
-            "system_ids": ["sys-1"],
-            "talkgroup_ids": ["9048"],
-            # 3 min ago — inside tg_dispatch_thin_idle_minutes (5).
-            "updated_at": (NOW - timedelta(minutes=3)).isoformat(),
-        }],
-    }
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx=ctx), _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-
-
-async def test_recent_same_tg_incident_older_than_short_window_now_gates():
-    # Regression test for the fix: 8 minutes is past the 5-minute bound but
-    # still inside the OLD 2-hour correlation_window_hours lookback. Before
-    # the fix this escaped the gate on any channel; after the fix it gates.
-    ctx = {
-        "system_id": "sys-1",
-        "talkgroup_id": 9048,
-        "talkgroup_name": "Dispatch",
-        "now": NOW,
-        "recent": [{
-            "incident_id": "inc-stale",
-            "system_ids": ["sys-1"],
-            "talkgroup_ids": ["9048"],
-            "updated_at": (NOW - timedelta(minutes=8)).isoformat(),
-        }],
-    }
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx=ctx), _llm("orphan"),
-    )
-    m_tiebreak.assert_not_called()
-    assert m_apply.call_args[0][0]["decision"]["action"] == "orphan"
-
-
-async def test_channel_name_does_not_affect_the_window():
-    # Owner correction, 2026-09-13 (direct scanning experience): a talkgroup
-    # named "tac"/"tactical" only sees materially different traffic during a
-    # real incident, and that's rare -- the bulk of traffic on any monitored
-    # channel, including high-risk stops and pursuits, runs on the main
-    # channel regardless of what it's named. An earlier version of this used
-    # a longer 15-minute window on anything not literally named "dispatch"/
-    # "patched"/"primary" (mirroring incident_correlator's fast/thin idle
-    # selection); that meant a busy single-channel department not literally
-    # named "dispatch" silently got the more permissive window and could
-    # reproduce #115's original bug. Same 8-minute age as the dispatch-named
-    # test above, but on a channel named "Tac 3" -- must gate identically,
-    # not escape into a longer window just because of the name.
-    ctx = {
-        "system_id": "sys-1",
-        "talkgroup_id": 383,
-        "talkgroup_name": "Tac 3",
-        "now": NOW,
-        "recent": [{
-            "incident_id": "inc-tac",
-            "system_ids": ["sys-1"],
-            "talkgroup_ids": ["383"],
-            "updated_at": (NOW - timedelta(minutes=8)).isoformat(),
-        }],
-    }
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx=ctx), _llm("orphan"),
-    )
-    m_tiebreak.assert_not_called()
-    assert m_apply.call_args[0][0]["decision"]["action"] == "orphan"
-
-
-async def test_gate_veto_reason_is_recorded_on_the_escalation_path():
-    # server-26#115: a live measurement window must be able to see *why* an
-    # llm=orphan/rules=new call escaped the gate without guessing from the raw
-    # dump (which produced a wrong "confirmed explanation" for 2 window-#3
-    # misses the first time). corr_gate_veto names the surviving condition.
-    ctx = {"call_severity": "major"}
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx=ctx), _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-    final = m_apply.call_args[0][0]["decision"]
-    assert final["corr_debug"]["corr_gate_veto"] == "severity"
-
-
-async def test_gate_veto_reason_is_absent_when_the_disagreement_is_not_orphan_vs_new():
-    # corr_gate_veto is only meaningful for the llm=orphan/rules=new shape the
-    # gate targets — it must not appear (or be misleadingly None-vs-absent) on
-    # an unrelated disagreement shape.
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("link", {}), _llm("orphan"),
-    )
-    m_tiebreak.assert_called_once()
-    final = m_apply.call_args[0][0]["decision"]
-    assert "corr_gate_veto" not in final["corr_debug"]
-
-
-async def test_recent_incident_on_a_different_talkgroup_still_gates():
-    ctx = {
-        "system_id": "sys-1",
-        "talkgroup_id": 9048,
-        "recent": [{
-            "incident_id": "inc-other",
-            "system_ids": ["sys-1"],
-            "talkgroup_ids": ["1200"],
-        }],
-    }
-    m_apply, m_tiebreak = await _run_consensus(
-        _preview("new", {}, ctx=ctx), _llm("orphan"),
-    )
-    m_tiebreak.assert_not_called()
-    assert m_apply.call_args[0][0]["decision"]["action"] == "orphan"
-
-
-async def test_llm_link_vs_rules_new_still_escalates():
-    m_apply, m_tiebreak = await _run_consensus(_preview("new", {}), _llm("link", "same job"))
-    m_tiebreak.assert_called_once()
-
-
-async def test_llm_orphan_vs_rules_link_still_escalates():
-    # Not the gate condition (gate needs rules=="new"); must fall through.
-    m_apply, m_tiebreak = await _run_consensus(_preview("link", {}), _llm("orphan"))
-    m_tiebreak.assert_called_once()
-
-
-def test_has_event_substance_predicate():
-    assert has_event_substance({"coords": {"lat": 1, "lng": 2}})
-    assert has_event_substance({"tags": ["fire"]})
-    assert has_event_substance({"call_vehicles": ["sedan"]})
-    assert not has_event_substance({})
-    assert not has_event_substance({"coords": None, "tags": [], "call_vehicles": []})
-    # units and location are NOT substance — nearly every transmission has them.
-    assert not has_event_substance({"call_units": ["7-Adam"], "location": "Main St"})
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Fix 2 — tighten corr_path=location
-# ─────────────────────────────────────────────────────────────────────────────
-
-CALL_COORDS = {"lat": 41.150000, "lng": -73.860000}
-# ~0.39 km north of the call — inside location_proximity_km (0.5) but well
-# outside the tight bar (_LOCATION_TIGHT_PROXIMITY_KM, 0.2).
-FAR_INC_COORDS = {"lat": 41.153500, "lng": -73.860000}
-# ~0.13 km north of the call — inside the tight bar.
-NEAR_INC_COORDS = {"lat": 41.151200, "lng": -73.860000}
-# ~0.28 km north — inside the 0.5 radius, outside the 0.2 tight bar; used as a
-# second candidate that must lose the nearest-wins sort to NEAR_INC_COORDS.
-MID_INC_COORDS = {"lat": 41.152500, "lng": -73.860000}
-
-
-def _inc(incident_id, coords, units):
-    return {
-        "incident_id": incident_id,
-        "system_ids": ["sys-1"],
-        "talkgroup_ids": ["100"],           # different TGID → fast path is a no-op
-        "location_coords": coords,
-        "units": units,
-        "tags": [],
-        "type": "police",
-        "updated_at": (NOW - timedelta(minutes=6)).isoformat(),
-        "started_at": (NOW - timedelta(minutes=20)).isoformat(),
-        "status": "active",
-        "call_ids": ["c0"],
-    }
-
-
-def _loc_ctx(*, incidents, call_units):
-    return {
-        "call_id": "call-loc",
-        "all_active": list(incidents),
-        "recent": list(incidents),
-        "call_doc": {},
-        "call_embedding": None,
-        "call_units": call_units,
-        "call_vehicles": [],
-        "call_cleared": [],
-        "call_severity": "routine",
-        "coords": CALL_COORDS,
-        "is_thin_call": False,
-        "now": NOW,
-        "system_id": "sys-1",
-        "talkgroup_id": 999,               # not in inc.talkgroup_ids
-        "talkgroup_name": "Tactical",
-        "tags": [],
-        "incident_type": "police",
-        "location": "Main St",
-        "location_coords": CALL_COORDS,
-        "reassignment": True,              # suppress the unit-continuity path
-        "create_if_new": True,
-    }
-
-
-def test_location_path_in_radius_but_no_unit_overlap_no_tight_proximity_does_not_link(caplog):
-    ctx = _loc_ctx(
-        incidents=[_inc("inc-loc", FAR_INC_COORDS, ["7-Adam"])],
-        call_units=["3-Boy"],
-    )
-    with caplog.at_level("INFO", logger="drb-c2-core"):
-        decision = _run_decision(ctx)
-    # Reaches, and is rejected by, the new guard (not an earlier path).
-    assert "location-path skipped" in caplog.text
-    assert decision["action"] != "link"
-    assert (decision.get("corr_debug") or {}).get("corr_path") != "location"
-
-
-def test_location_path_links_on_unit_overlap_with_distinct_fit_signal():
-    ctx = _loc_ctx(
-        incidents=[_inc("inc-loc", FAR_INC_COORDS, ["5-Adam"])],
-        call_units=["5-Adam"],
-    )
-    decision = _run_decision(ctx)
-    assert decision["action"] == "link"
-    assert decision["corr_debug"]["corr_path"] == "location"
-    # NOT "unit_overlap" — that value belongs to the fast path's histogram bucket.
-    assert decision["corr_debug"]["corr_fit_signal"] == "location_unit_overlap"
-
-
-def test_location_path_links_on_tight_proximity_without_unit_overlap():
-    ctx = _loc_ctx(
-        incidents=[_inc("inc-loc", NEAR_INC_COORDS, ["7-Adam"])],
-        call_units=["3-Boy"],
-    )
-    decision = _run_decision(ctx)
-    assert decision["action"] == "link"
-    assert decision["corr_debug"]["corr_path"] == "location"
-    assert decision["corr_debug"]["corr_fit_signal"] == "location_proximity"
-
-
-def test_location_path_picks_nearest_in_radius_candidate():
-    # `recent` order puts the farther tight-proximity incident first; the guard
-    # must still select the nearest one.
-    ctx = _loc_ctx(
-        incidents=[
-            _inc("inc-mid", MID_INC_COORDS, ["3-Boy"]),   # ~0.28 km, tight-fail
-            _inc("inc-near", NEAR_INC_COORDS, ["3-Boy"]),  # ~0.13 km, tight-pass
-        ],
-        call_units=["3-Boy"],
-    )
-    decision = _run_decision(ctx)
-    assert decision["action"] == "link"
-    assert decision["matched_incident"]["incident_id"] == "inc-near"
-    assert decision["corr_debug"]["corr_path"] == "location"

drb-c2-core/tests/test_correlation_decision_flow.py

@@ -0,0 +1,432 @@
+"""
+server-26#299 — rules decide confident outcomes; ONE cheap LLM call only in the
+ambiguous band; no cheap/smart consensus tiebreak.
+
+Also pins the section fixes that ride on the new shape:
+  #86  divergence veto bar is a setting, and the band between it and the match
+       bar is ambiguous rather than a silent merge
+  #103 a scene with no embedding is "unverified", not verified
+  #88  LLM yes/no fields fail closed; a closure only counts on a confident link
+  #189 weak links refused past the span cap; LLM links distance-guarded
+  #190 severity steps down on an explicit stand-down, the peak never does
+  #27  sweep budget is wall-clock
+  #301 late uploads are correlated on the call's own clock
+  #302 a retried upload does not re-run the pipeline
+"""
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from app.config import settings
+from app.internal import incident_correlator as ic
+from app.internal import llm_correlator, recorrelation_sweep
+from app.routers import upload
+
+NOW = datetime(2026, 10, 27, 15, 0, 0, tzinfo=timezone.utc)
+
+
+def _inc(incident_id="inc-1", idle=1.0, span=None, **kw) -> dict:
+    updated = NOW - timedelta(minutes=idle)
+    started = NOW - timedelta(minutes=span if span is not None else idle)
+    inc = {
+        "incident_id": incident_id, "system_ids": ["sys-1"], "talkgroup_ids": ["383"],
+        "updated_at": updated.isoformat(), "started_at": started.isoformat(),
+        "status": "active", "call_ids": ["seed"], "type": "police",
+    }
+    inc.update(kw)
+    return inc
+
+
+def _ctx(incidents=(), **kw) -> dict:
+    incidents = list(incidents)
+    base = {
+        "call_id": "call-1", "all_active": incidents, "recent": incidents,
+        "call_doc": {}, "call_embedding": None, "call_units": [], "call_vehicles": [],
+        "call_cleared": [], "call_severity": "routine", "coords": None,
+        "is_thin_call": False, "now": NOW, "system_id": "sys-1", "talkgroup_id": 383,
+        "talkgroup_name": "Ch 1", "tags": [], "incident_type": None, "location": None,
+        "location_coords": None, "reassignment": False, "create_if_new": True,
+    }
+    base.update(kw)
+    return base
+
+
+def _vec(sim: float) -> tuple[list, list]:
+    """Two unit vectors with cosine similarity `sim`."""
+    import math
+    return [1.0, 0.0], [sim, math.sqrt(max(0.0, 1 - sim * sim))]
+
+
+# ── The band ────────────────────────────────────────────────────────────────
+
+def test_fresh_unit_link_is_confident():
+    d = ic._run_decision(_ctx([_inc(idle=2, units=["6-Adam"])], call_units=["6-Adam"]))
+    assert d["action"] == "link"
+    assert d["corr_debug"]["corr_fit_signal"] == "unit_overlap"
+    assert d["ambiguous"] is None
+
+
+def test_stale_unit_link_without_embedding_is_ambiguous_not_verified():
+    # #103: no embedding used to skip the veto and link as if verified.
+    d = ic._run_decision(_ctx([_inc(idle=20, units=["6-Adam"])], call_units=["6-Adam"]))
+    assert d["action"] == "link"
+    assert d["corr_debug"]["corr_fit_signal"] == "unit_overlap_unverified"
+    assert d["ambiguous"] == "unverified_unit_link"
+
+
+def test_young_but_long_running_unit_link_is_ambiguous():
+    # #189: busy channel keeps idle low; age is what reveals a sink forming.
+    inc = _inc(idle=1, span=settings.unit_link_confident_max_span_minutes + 5, units=["6-Adam"])
+    d = ic._run_decision(_ctx([inc], call_units=["6-Adam"]))
+    assert d["ambiguous"] == "unverified_unit_link"
+
+
+@pytest.mark.parametrize("sim,expected", [
+    (0.80, ("orphan", "content_divergence")),     # below the divergence bar → veto
+    (0.88, ("link", "unit_overlap_unverified")),  # between the bars → ambiguous (#86)
+    (0.95, ("link", "unit_content_match")),       # at the match bar → confident
+])
+def test_divergence_band_between_the_two_thresholds(sim, expected):
+    call_emb, inc_emb = _vec(sim)
+    inc = _inc(idle=20, units=["6-Adam"], embedding=inc_emb)
+    fit, signal = ic._call_fits_incident(
+        inc, ["6-Adam"], [], None, settings.location_proximity_km,
+        call_embedding=call_emb, now=NOW,
+    )
+    assert (("link" if fit else "orphan"), signal) == expected
+
+
+def test_divergence_bar_is_the_setting_not_a_literal():
+    call_emb, inc_emb = _vec(0.88)
+    inc = _inc(idle=20, units=["6-Adam"], embedding=inc_emb)
+    with patch.object(settings, "embedding_divergence_threshold", 0.90):
+        fit, signal = ic._call_fits_incident(
+            inc, ["6-Adam"], [], None, settings.location_proximity_km,
+            call_embedding=call_emb, now=NOW,
+        )
+    assert (fit, signal) == (False, "content_divergence")
+
+
+def test_new_with_live_candidates_is_ambiguous_without_none():
+    d = ic._run_decision(_ctx([_inc(talkgroup_ids=["999"])], incident_type="police"))
+    assert (d["action"], d["ambiguous"]) == ("new", "new_with_candidates")
+    d = ic._run_decision(_ctx([], incident_type="police"))
+    assert (d["action"], d["ambiguous"]) == ("new", None)
+
+
+def test_orphan_with_unfit_same_tg_candidate_is_ambiguous():
+    # Substantive (a location) but untyped, live incident on its talkgroup, no evidence.
+    d = ic._run_decision(_ctx([_inc(idle=2)], location="Mohegan Park Ave"))
+    assert (d["action"], d["ambiguous"]) == ("orphan", "unfit_tg_candidate")
+
+
+def test_thin_sweep_and_clearance_calls_are_never_ambiguous():
+    inc = _inc(idle=0.2)
+    assert ic._run_decision(_ctx([inc], is_thin_call=True))["ambiguous"] is None
+    assert ic._run_decision(_ctx([inc], incident_type="police", create_if_new=False))["ambiguous"] is None
+    stale = _inc(idle=20, units=["45-9"])
+    d = ic._run_decision(_ctx([stale], call_units=["45-9"], call_cleared=["45-9"]))
+    assert d["action"] == "link" and d["ambiguous"] is None
+
+
+# ── #189: caps on absorption ────────────────────────────────────────────────
+
+def test_weak_link_span_cap_refuses_unit_only_links():
+    inc = _inc(idle=1, span=settings.incident_weak_link_max_span_minutes + 1, units=["6-Adam"])
+    d = ic._run_decision(_ctx([inc], call_units=["6-Adam"], incident_type="accident"))
+    assert d["action"] == "new"
+    assert ic.llm_candidates(_ctx([inc])) == []
+
+
+def test_weak_link_span_cap_also_binds_unit_continuity():
+    inc = _inc(idle=1, span=settings.incident_weak_link_max_span_minutes + 1,
+               units=["6-Adam"], talkgroup_ids=["999"])
+    d = ic._run_decision(_ctx([inc], call_units=["6-Adam"]))
+    assert d["action"] == "orphan"
+
+
+def test_llm_link_distance_guard():
+    pin = {"lat": 41.0, "lng": -73.8}
+    inc = _inc(location="Main St", location_coords=pin, location_coords_source="Main St")
+    near = _ctx(coords={"lat": 41.005, "lng": -73.8})
+    far = _ctx(coords={"lat": 41.2, "lng": -73.8})   # ~22 km
+    assert ic.llm_link_allowed(inc, near)
+    assert not ic.llm_link_allowed(inc, far)
+    assert ic.llm_link_allowed({**inc, "tags": ["vehicle-pursuit"]}, far)
+
+
+# ── llm_correlator: one call, validated ─────────────────────────────────────
+
+def test_parse_rejects_links_to_unshown_incidents_and_bad_actions():
+    shown = [_inc("inc-1")]
+    ctx = _ctx(shown)
+    assert llm_correlator._parse_response({"action": "link", "incident_id": "nope"}, ctx, shown) is None
+    assert llm_correlator._parse_response({"action": "maybe"}, ctx, shown) is None
+    ok = llm_correlator._parse_response({"action": "link", "incident_id": "inc-1"}, ctx, shown)
+    assert ok["action"] == "link" and ok["matched_incident"]["incident_id"] == "inc-1"
+
+
+def test_parse_new_uses_the_rules_creation_gate():
+    ctx = _ctx(call_severity="minor")
+    assert llm_correlator._parse_response({"action": "new"}, ctx, [])["incident_type"] == "other"
+    out = llm_correlator._parse_response({"action": "new"}, _ctx(), [])
+    assert out["action"] == "orphan"
+
+
+async def _correlate_with(rules_decision, llm_return, ctx=None):
+    ctx = ctx or _ctx([_inc()])
+    preview = {"decision": rules_decision, "ctx": ctx}
+    with patch("app.internal.incident_correlator.preview_correlation", AsyncMock(return_value=preview)), \
+         patch("app.internal.incident_correlator.apply_correlation", AsyncMock(return_value="inc-x")) as apply, \
+         patch("app.internal.llm_correlator.decide", AsyncMock(return_value=llm_return)) as decide:
+        _, final = await upload._correlate(
+            call_id="call-1", node_id="n", system_id="sys-1", talkgroup_id=383,
+            talkgroup_name="Ch 1", tags=[], incident_type=None, location=None,
+            location_coords=None,
+        )
+    return final, decide, apply
+
+
+def _rules(action, ambiguous=None, inc=None):
+    return {"action": action, "matched_incident": inc, "incident_type": "police" if action == "new" else None,
+            "corr_debug": {}, "ambiguous": ambiguous}
+
+
+def _llm(action, inc=None):
+    return {"action": action, "matched_incident": inc, "incident_type": "police" if action == "new" else None,
+            "corr_debug": {"corr_llm_reasoning": "r"}, "reasoning": "r"}
+
+
+@pytest.mark.asyncio
+async def test_confident_rules_decision_never_calls_the_llm():
+    final, decide, _ = await _correlate_with(_rules("new"), _llm("orphan"))
+    decide.assert_not_awaited()
+    assert final["corr_debug"]["corr_consensus"] == "rules_only"
+
+
+@pytest.mark.asyncio
+async def test_ambiguous_decision_makes_exactly_one_llm_call_and_commits_it():
+    inc = _inc("inc-2")
+    final, decide, apply = await _correlate_with(_rules("new", "new_with_candidates"), _llm("link", inc))
+    decide.assert_awaited_once()
+    committed = apply.await_args.args[0]["decision"]
+    assert committed["action"] == "link" and committed["matched_incident"]["incident_id"] == "inc-2"
+    dbg = final["corr_debug"]
+    assert dbg["corr_consensus"] == "llm" and dbg["corr_path"] == "llm"
+    assert dbg["corr_band"] == "new_with_candidates"
+    assert (dbg["corr_rules_action"], dbg["corr_llm_action"]) == ("new", "link")
+
+
+@pytest.mark.asyncio
+async def test_llm_unavailable_keeps_the_rules_decision():
+    rules = _rules("new", "new_with_candidates")
+    final, decide, _ = await _correlate_with(rules, None)
+    assert final is rules and final["corr_debug"]["corr_consensus"] == "rules_fallback"
+
+
+@pytest.mark.asyncio
+async def test_llm_orphan_cannot_drop_a_major_or_reassignment_call():
+    final, _, _ = await _correlate_with(
+        _rules("new", "new_with_candidates"), _llm("orphan"), _ctx([_inc()], call_severity="major"))
+    assert final["action"] == "new" and final["corr_debug"]["corr_consensus"] == "rules_guard"
+    final, _, _ = await _correlate_with(
+        _rules("new", "new_with_candidates"), _llm("orphan"), _ctx([_inc()]))
+    assert final["action"] == "orphan" and final["corr_debug"]["corr_consensus"] == "llm"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("extra", [{"incident_type": "fire"}, {"tags": ["structure-fire"]},
+                                   {"coords": {"lat": 41.0, "lng": -73.8}}, {"reassignment": True}])
+async def test_llm_orphan_cannot_drop_a_typed_or_substantive_call(extra):
+    final, _, _ = await _correlate_with(
+        _rules("new", "new_with_candidates"), _llm("orphan"), _ctx([_inc()], **extra))
+    assert final["action"] == "new" and final["corr_debug"]["corr_consensus"] == "rules_guard"
+
+
+@pytest.mark.asyncio
+async def test_llm_orphan_over_a_rules_link_is_tombstoned_for_the_sweep():
+    inc = _inc()
+    rules = _rules("link", "unverified_unit_link", inc)
+    final, _, _ = await _correlate_with(rules, _llm("orphan"))
+    assert final["action"] == "orphan" and final["corr_debug"]["corr_path"] == "llm_orphan"
+
+
+@pytest.mark.asyncio
+async def test_back_dated_link_never_moves_updated_at_backward():
+    inc = _inc(idle=1, span=10)
+    with patch.object(ic, "fstore") as fs:
+        fs.doc_set = AsyncMock()
+        await ic._update_incident(inc, "late-call", 383, "sys-1", [], None, None, ["6-Adam"], [], None,
+                                  NOW - timedelta(hours=3))
+    assert fs.doc_set.await_args.args[2]["updated_at"] == inc["updated_at"]
+
+
+@pytest.mark.asyncio
+async def test_llm_confirming_the_rules_link_keeps_the_rules_path():
+    inc = _inc()
+    rules = _rules("link", "unverified_unit_link", inc)
+    rules["corr_debug"] = {"corr_path": "fast/single", "corr_fit_signal": "unit_overlap_unverified"}
+    final, _, _ = await _correlate_with(rules, _llm("link", inc))
+    assert final["corr_debug"]["corr_path"] == "fast/single"
+    assert final["corr_debug"]["corr_consensus"] == "llm"
+
+
+def test_no_tiebreak_tier_remains():
+    assert not hasattr(llm_correlator, "tiebreak")
+    assert not hasattr(settings, "corr_smart_model")
+    assert not hasattr(upload, "_correlate_with_consensus")
+
+
+# ── #88 ─────────────────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("value,expected", [
+    (True, True), (False, False), ("true", True), ("True", True), ("yes", True),
+    ("false", False), ("False", False), ("no", False), ("", False), (None, False),
+    ("maybe", False), (1, False), ({"x": 1}, False),
+])
+def test_parse_llm_bool_fails_closed(value, expected):
+    assert ic.parse_llm_bool(value) is expected
+
+
+def test_closure_only_honoured_on_a_confident_link_or_own_new_incident():
+    assert upload._may_resolve({"action": "new", "corr_debug": {"corr_consensus": "llm"}})
+    assert upload._may_resolve({"action": "link", "corr_debug": {"corr_consensus": "rules_only",
+                                                                  "corr_path": "fast/single"}})
+    assert not upload._may_resolve({"action": "link", "corr_debug": {"corr_consensus": "rules_only",
+                                                                      "corr_path": "fast/thin"}})
+    assert not upload._may_resolve({"action": "link", "corr_debug": {"corr_consensus": "llm",
+                                                                      "corr_path": "llm"}})
+    assert not upload._may_resolve({"action": "orphan", "corr_debug": {}})
+
+
+@pytest.mark.asyncio
+async def test_quoted_false_does_not_resolve_or_blank_units():
+    scene = {"tags": [], "incident_type": "police", "location": None, "location_coords": None,
+             "units": ["6-Adam"], "resolved": "false", "reassignment": "false"}
+    corr = AsyncMock(return_value=("inc-1", {"action": "new", "corr_debug": {}}))
+    with patch.object(upload, "_correlate", corr), patch.object(upload, "fstore") as fs:
+        fs.doc_set = AsyncMock()
+        await upload._correlate_scenes("call-1", "n", "sys-1", 383, "Ch 1", [scene])
+    assert corr.await_args.kwargs["units"] == ["6-Adam"]
+    assert corr.await_args.kwargs["reassignment"] is False
+    fs.doc_set.assert_not_awaited()
+
+
+# ── #190 ────────────────────────────────────────────────────────────────────
+
+def test_severity_escalates_immediately_and_keeps_its_peak():
+    out = ic._severity_after_link({"severity": "minor"}, "major", "shots fired")
+    assert out["severity"] == "major" and out["severity_peak"] == "major"
+
+
+def test_calmer_followup_without_stand_down_does_not_deescalate():
+    out = ic._severity_after_link({"severity": "major"}, "routine", "Engine 5 on scene, 10-4")
+    assert out["severity"] == "major" and "severity_deescalated_from" not in out
+
+
+def test_explicit_stand_down_deescalates_but_peak_stays():
+    inc = {"severity": "major"}
+    out = ic._severity_after_link(inc, "minor", "Only smoke showing, no flames.")
+    assert out == {"severity": "minor", "severity_peak": "major", "severity_deescalated_from": "major"}
+    later = ic._severity_after_link({**inc, **out}, "routine", "copy")
+    assert later["severity"] == "minor" and later["severity_peak"] == "major"
+
+
+@pytest.mark.parametrize("text", [
+    "call was unfounded", "false alarm, resetting the panel", "GOA, nothing found",
+    "Code 4, you can stand down", "fire is out",
+])
+def test_stand_down_phrases(text):
+    assert ic._is_stand_down(text)
+
+
+@pytest.mark.parametrize("text", ["disregard my last", "code 40", "flames showing",
+                                  "primary search all clear", "Code 4", None])
+def test_not_stand_down_phrases(text):
+    assert not ic._is_stand_down(text)
+
+
+# ── #27 ─────────────────────────────────────────────────────────────────────
+
+def test_sweep_tombstones_only_on_the_last_visible_pass():
+    scan = settings.recorrelation_scan_minutes
+    young = {"ended_at": (NOW - timedelta(minutes=20)).isoformat()}
+    last = {"ended_at": (NOW - timedelta(minutes=scan - 1)).isoformat()}
+    assert not recorrelation_sweep._last_chance(young, NOW)
+    assert recorrelation_sweep._last_chance(last, NOW)
+    assert recorrelation_sweep._last_chance({}, NOW)
+
+
+@pytest.mark.asyncio
+async def test_sweep_keeps_retrying_past_the_old_three_attempt_budget():
+    call = {"call_id": "c", "started_at": (NOW - timedelta(minutes=30)).isoformat(),
+            "ended_at": (NOW - timedelta(minutes=30)).isoformat(), "corr_sweep_count": 5}
+    with patch("app.internal.incident_correlator.correlate_call", AsyncMock(return_value=None)), \
+         patch.object(recorrelation_sweep, "fstore") as fs:
+        fs.doc_set = AsyncMock()
+        assert await recorrelation_sweep._recorrelate_orphan(call, NOW) is False
+    assert fs.doc_set.await_args.args[2] == {"corr_sweep_count": 6}
+
+
+# ── #301 / #302 ─────────────────────────────────────────────────────────────
+
+def test_late_upload_is_anchored_to_started_at_and_on_time_is_not():
+    with patch("app.internal.clock.now", return_value=NOW):
+        late = NOW - timedelta(hours=3)
+        assert upload._late_upload_anchor({"started_at": late.isoformat()}) == late
+        assert upload._late_upload_anchor({"started_at": late}) == late          # native timestamp
+        assert upload._late_upload_anchor({"started_at": (NOW - timedelta(minutes=1)).isoformat()}) is None
+        assert upload._late_upload_anchor({}) is None
+
+
+@pytest.mark.asyncio
+async def test_late_upload_correlates_on_the_calls_own_clock():
+    from app.internal import clock
+    seen = {}
+    started = NOW - timedelta(hours=3)
+
+    async def fake_extract(**kw):
+        seen["now"] = clock.now()
+        return [], [], []
+
+    flags = ({"stt_enabled": False, "correlation_enabled": True}, lambda k: k == "correlation_enabled")
+    with patch.object(upload, "fstore") as fs, \
+         patch.object(upload, "_resolve_flags", AsyncMock(return_value=flags)), \
+         patch.object(upload, "_extract_and_correlate", fake_extract), \
+         patch("app.internal.talkgroups.resolve", AsyncMock(return_value="Ch 1")), \
+         patch("app.internal.alerter.check_and_dispatch", AsyncMock()):
+        fs.doc_get = AsyncMock(return_value={"started_at": started.isoformat(), "talkgroup_name": "Ch 1"})
+        fs.doc_set = AsyncMock()
+        await upload._run_intelligence_pipeline("call-1", "n", "sys-1", 383, "Ch 1", None)
+    assert seen["now"] == started
+    assert clock.now() != started   # pin released
+
+
+@pytest.mark.asyncio
+async def test_retried_upload_returns_stored_audio_without_rerunning_the_pipeline():
+    from fastapi import BackgroundTasks
+    from fastapi.security import HTTPAuthorizationCredentials
+
+    docs = {
+        ("node_keys", "n"): {"api_key": "k"},
+        ("nodes", "n"): {"approval_status": "approved"},
+        ("calls", "call-1"): {"audio_gcs_uri": "gs://b/call-1.mp3"},
+    }
+
+    class _File:
+        filename = "a.mp3"
+
+        async def read(self):
+            return b"audio"
+
+    bt = BackgroundTasks()
+    with patch.object(upload, "fstore") as fs, patch.object(upload, "upload_audio", AsyncMock()) as up:
+        fs.doc_get = AsyncMock(side_effect=lambda coll, doc_id: docs.get((coll, doc_id)))
+        out = await upload.upload_call_audio(
+            background_tasks=bt, file=_File(), call_id="call-1", node_id="n",
+            credentials=HTTPAuthorizationCredentials(scheme="Bearer", credentials="k"),
+        )
+    assert out == {"url": "gs://b/call-1.mp3", "already_uploaded": True}
+    up.assert_not_awaited()
+    assert bt.tasks == []

drb-c2-core/tests/test_correlator_location_path.py

@@ -0,0 +1,141 @@
+"""
+server-26#115 — location-path corroboration, plus the shared substance predicate.
+
+(The consensus LLM-orphan gate that used to be tested here was removed with the
+cheap/smart tiebreak in server-26#299; its replacement is tested in
+test_correlation_decision_flow.py.)
+
+The `location` correlation path linked on a bare sub-`location_proximity_km`
+(0.5 km) distance alone, taking whichever incident came first in an unsorted
+`recent`. In a dense village two unrelated events routinely geocode that close.
+A `location` link now needs unit overlap with the candidate OR a distance under a
+tighter bar, and picks the NEAREST qualifying candidate. A unit-overlap location
+link is tagged `location_unit_overlap` so it does not merge into the fast path's
+bucket in the admin fit-signal histogram.
+"""
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from app.internal.incident_correlator import _run_decision, has_event_substance
+
+NOW = datetime(2026, 9, 7, 21, 30, 0, tzinfo=timezone.utc)
+
+
+def test_has_event_substance_predicate():
+    assert has_event_substance({"coords": {"lat": 1, "lng": 2}})
+    assert has_event_substance({"tags": ["fire"]})
+    assert has_event_substance({"call_vehicles": ["sedan"]})
+    assert not has_event_substance({})
+    assert not has_event_substance({"coords": None, "tags": [], "call_vehicles": []})
+    # units and location are NOT substance — nearly every transmission has them.
+    assert not has_event_substance({"call_units": ["7-Adam"], "location": "Main St"})
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Fix 2 — tighten corr_path=location
+# ─────────────────────────────────────────────────────────────────────────────
+
+CALL_COORDS = {"lat": 41.150000, "lng": -73.860000}
+# ~0.39 km north of the call — inside location_proximity_km (0.5) but well
+# outside the tight bar (_LOCATION_TIGHT_PROXIMITY_KM, 0.2).
+FAR_INC_COORDS = {"lat": 41.153500, "lng": -73.860000}
+# ~0.13 km north of the call — inside the tight bar.
+NEAR_INC_COORDS = {"lat": 41.151200, "lng": -73.860000}
+# ~0.28 km north — inside the 0.5 radius, outside the 0.2 tight bar; used as a
+# second candidate that must lose the nearest-wins sort to NEAR_INC_COORDS.
+MID_INC_COORDS = {"lat": 41.152500, "lng": -73.860000}
+
+
+def _inc(incident_id, coords, units):
+    return {
+        "incident_id": incident_id,
+        "system_ids": ["sys-1"],
+        "talkgroup_ids": ["100"],           # different TGID → fast path is a no-op
+        "location_coords": coords,
+        "units": units,
+        "tags": [],
+        "type": "police",
+        "updated_at": (NOW - timedelta(minutes=6)).isoformat(),
+        "started_at": (NOW - timedelta(minutes=20)).isoformat(),
+        "status": "active",
+        "call_ids": ["c0"],
+    }
+
+
+def _loc_ctx(*, incidents, call_units):
+    return {
+        "call_id": "call-loc",
+        "all_active": list(incidents),
+        "recent": list(incidents),
+        "call_doc": {},
+        "call_embedding": None,
+        "call_units": call_units,
+        "call_vehicles": [],
+        "call_cleared": [],
+        "call_severity": "routine",
+        "coords": CALL_COORDS,
+        "is_thin_call": False,
+        "now": NOW,
+        "system_id": "sys-1",
+        "talkgroup_id": 999,               # not in inc.talkgroup_ids
+        "talkgroup_name": "Tactical",
+        "tags": [],
+        "incident_type": "police",
+        "location": "Main St",
+        "location_coords": CALL_COORDS,
+        "reassignment": True,              # suppress the unit-continuity path
+        "create_if_new": True,
+    }
+
+
+def test_location_path_in_radius_but_no_unit_overlap_no_tight_proximity_does_not_link(caplog):
+    ctx = _loc_ctx(
+        incidents=[_inc("inc-loc", FAR_INC_COORDS, ["7-Adam"])],
+        call_units=["3-Boy"],
+    )
+    with caplog.at_level("INFO", logger="drb-c2-core"):
+        decision = _run_decision(ctx)
+    # Reaches, and is rejected by, the new guard (not an earlier path).
+    assert "location-path skipped" in caplog.text
+    assert decision["action"] != "link"
+    assert (decision.get("corr_debug") or {}).get("corr_path") != "location"
+
+
+def test_location_path_links_on_unit_overlap_with_distinct_fit_signal():
+    ctx = _loc_ctx(
+        incidents=[_inc("inc-loc", FAR_INC_COORDS, ["5-Adam"])],
+        call_units=["5-Adam"],
+    )
+    decision = _run_decision(ctx)
+    assert decision["action"] == "link"
+    assert decision["corr_debug"]["corr_path"] == "location"
+    # NOT "unit_overlap" — that value belongs to the fast path's histogram bucket.
+    assert decision["corr_debug"]["corr_fit_signal"] == "location_unit_overlap"
+
+
+def test_location_path_links_on_tight_proximity_without_unit_overlap():
+    ctx = _loc_ctx(
+        incidents=[_inc("inc-loc", NEAR_INC_COORDS, ["7-Adam"])],
+        call_units=["3-Boy"],
+    )
+    decision = _run_decision(ctx)
+    assert decision["action"] == "link"
+    assert decision["corr_debug"]["corr_path"] == "location"
+    assert decision["corr_debug"]["corr_fit_signal"] == "location_proximity"
+
+
+def test_location_path_picks_nearest_in_radius_candidate():
+    # `recent` order puts the farther tight-proximity incident first; the guard
+    # must still select the nearest one.
+    ctx = _loc_ctx(
+        incidents=[
+            _inc("inc-mid", MID_INC_COORDS, ["3-Boy"]),   # ~0.28 km, tight-fail
+            _inc("inc-near", NEAR_INC_COORDS, ["3-Boy"]),  # ~0.13 km, tight-pass
+        ],
+        call_units=["3-Boy"],
+    )
+    decision = _run_decision(ctx)
+    assert decision["action"] == "link"
+    assert decision["matched_incident"]["incident_id"] == "inc-near"
+    assert decision["corr_debug"]["corr_path"] == "location"

drb-c2-core/tests/test_correlator_merge_caps.py

@@ -332,13 +332,18 @@ def test_duration_cap_blocks_the_thin_path_too():
     assert _run_decision(_ctx(all_active=[inc], recent=[inc]))["action"] == "orphan"
 
 
-def test_incident_just_under_the_duration_cap_still_accepts_calls():
+def test_incident_just_under_the_duration_cap_still_accepts_strong_evidence():
+    # Past the weak-link span cap (server-26#189) shared units alone no longer
+    # link, but strong evidence (a shared vehicle) still does until the hard cap.
     inc = _long_running(settings.incident_max_duration_minutes - 10)
     inc["units"] = ["6-Adam"]
+    inc["vehicles"] = ["red sedan"]
     decision = _run_decision(_ctx(
-        all_active=[inc], recent=[inc], is_thin_call=False, call_units=["6-Adam"],
+        all_active=[inc], recent=[inc], is_thin_call=False,
+        call_units=["7-Boy"], call_vehicles=["red sedan"],
     ))
     assert decision["action"] == "link"
+    assert decision["corr_debug"]["corr_fit_signal"] == "vehicle_overlap"
 
 
 def test_call_count_cap_forces_a_new_incident():

drb-c2-core/tests/test_gemini_helper.py

@@ -34,12 +34,13 @@ def test_minimal_thinking_by_default_and_usage_lands_in_the_sink():
     models = _FakeModels()
     a, b = _patched(models)
     sink = {}
-    tok = gemini.collect_usage(sink)
+    from app.internal import llm
+    tok = llm.collect_usage(sink)
     try:
         with a, b:
             assert gemini.generate_json("m1", "p", purpose="correlation") == {"action": "link"}
     finally:
-        gemini.reset_usage(tok)
+        llm.reset_usage(tok)
     assert models.configs == [{"thinking_level": "minimal"}]
     assert sink == {"correlation:m1": {"calls": 1, "in": 1200, "out": 30, "thinking": 0}}
 

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.incident_correlator import parse_llm_bool as _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

drb-c2-core/tests/test_llm_provider.py

@@ -0,0 +1,48 @@
+"""app/internal/llm.py: provider follows the model name (2026-11-08 consolidation)."""
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from app.config import settings
+from app.internal import llm
+
+
+def test_provider_follows_model_name():
+    assert llm.provider_for("gpt-4o-mini") == "openai"
+    assert llm.provider_for("gemini-3.6-flash") == "gemini"
+
+
+def test_available_checks_the_right_key():
+    with patch.object(settings, "openai_api_key", "k"), patch.object(settings, "gemini_api_key", None):
+        assert llm.available("gpt-4o-mini") and not llm.available("gemini-3.6-flash")
+
+
+def test_defaults_are_openai():
+    assert llm.provider_for(settings.corr_cheap_model) == "openai"
+    assert llm.provider_for(settings.transcript_correction_model) == "openai"
+
+
+def test_openai_path_parses_json_and_records_usage():
+    resp = SimpleNamespace(
+        choices=[SimpleNamespace(message=SimpleNamespace(content='{"action": "new"}'))],
+        usage=SimpleNamespace(prompt_tokens=120, completion_tokens=9),
+    )
+    calls = {}
+
+    class FakeClient:
+        def __init__(self):
+            self.chat = SimpleNamespace(completions=SimpleNamespace(create=self.create))
+
+        def create(self, **kw):
+            calls.update(kw)
+            return resp
+
+    sink = {}
+    tok = llm.collect_usage(sink)
+    try:
+        with patch("app.internal.ai_clients.openai_client", FakeClient):
+            out = llm.generate_json("gpt-4o-mini", "Return JSON", purpose="correlation")
+    finally:
+        llm.reset_usage(tok)
+    assert out == {"action": "new"}
+    assert calls["response_format"] == {"type": "json_object"}
+    assert sink["correlation:gpt-4o-mini"] == {"calls": 1, "in": 120, "out": 9, "thinking": 0}

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)

drb-c2-core/tests/test_transcript_correction.py

@@ -57,7 +57,7 @@ def _api_key():
     The dev venv has no GEMINI_API_KEY, and correct() returns early without one
     — which would make every assertion below pass for the wrong reason.
     """
-    with patch.object(tc.settings, "gemini_api_key", "test-key"):
+    with patch.object(tc.settings, "openai_api_key", "test-key"):
         yield
 
 
@@ -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

drb-c2-core/tests/test_upload_short_transcript_correlation.py

@@ -29,7 +29,7 @@ async def _run_ingest(skip_reason):
          patch("app.internal.firestore.doc_get_cached",
                AsyncMock(return_value={"system_id": "sys-1", "ai_flags": {}})), \
          patch.object(upload, "fstore") as fs, \
-         patch.object(upload, "_correlate_with_consensus", AsyncMock(return_value=None)) as corr, \
+         patch.object(upload, "_correlate", AsyncMock(return_value=(None, {}))) as corr, \
          patch("app.internal.transcription.transcribe_call",
                AsyncMock(return_value=("10-8", []))), \
          patch("app.internal.intelligence.extract_scenes", AsyncMock(return_value=[])), \

drb-frontend/app/features/page.tsx

@@ -17,7 +17,7 @@ const SECTIONS: {
     body:
       "The correlation engine groups related transmissions — across talkgroups and even across nodes — into a single incident. A pursuit renders as a path through every checkin point heard while it moved; a structure fire or accident renders as a pin at the location dispatch gave.",
     points: [
-      "Hybrid rule + LLM correlation with a cheap/smart consensus tiebreak",
+      "Rules decide clear-cut matches; an LLM weighs in only on the ambiguous ones",
       "Distance, timing, shared units, and talkgroup signals all feed the match",
       "Every call keeps its correlation debug trail for admins to audit",
     ],

drb-frontend/components/admin/ReplayTab.tsx

@@ -361,10 +361,10 @@ function RunDetail({ run }: { run: ReplayRun }) {
           paths: {Object.entries(m.corr_path).map(([k, v]) => `${k} ${v}`).join(" · ")}
         </p>
       )}
-      {m?.gemini_usage && Object.keys(m.gemini_usage).length > 0 && (
+      {m?.llm_usage && Object.keys(m.llm_usage).length > 0 && (
         <p className="text-xs font-mono text-ink-muted">
           Gemini tokens:{" "}
-          {Object.entries(m.gemini_usage)
+          {Object.entries(m.llm_usage)
             .map(([k, u]) => `${k} ${u.calls} calls, in ${u.in.toLocaleString()} / out ${u.out.toLocaleString()} / thinking ${u.thinking.toLocaleString()}`)
             .join(" · ")}
         </p>

drb-frontend/lib/types.ts

@@ -669,7 +669,7 @@ export interface ReplayMetrics {
   llm_decisions: number;
   est_cost_usd: number;
   ai_failures?: Record<string, number>;
-  gemini_usage?: Record<string, { calls: number; in: number; out: number; thinking: number }>;
+  llm_usage?: Record<string, { calls: number; in: number; out: number; thinking: number }>;
 }
 
 export interface ReplayRun {

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 }}