Merge pull request 'correlator/intelligence: let 10-8s actually close incidents' (#174) from fix/clearance-signal into main

Replay of 09-22 10:00-12:00 ET (server-26#170): 0 of 19 incidents resolved
on a clear, 19 on the idle timer, although 25 transmissions said 10-8/clear.
Three independent breaks:

1. Short clears never reached extraction. "45-9, I'm clear." is <=5 words,
   so extract_scenes skipped it before GPT and cleared_units stayed empty.
   A rule parser now names the unit when it precedes the status word
   (never guesses: "10-8, 10-8." / "CMT clear." clear nobody) and returns a
   minimal scene that links by unit overlap but cannot open an incident.
2. Clearance compared unit strings exactly, so "11-Adam" clearing never
   removed "11 Adam". Now by _normalize_unit key.
3. units_active collected "Desk", "Central", "Division", "unknown", plate
   phonetics — none of which ever clear, so all-clear could never pass.
   Only units carrying a number (and not a ten-code) are tracked now; the
   rest stay in `units` for matching.

c2-core: 463 pass.

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

drb-c2-core/app/config.py

@@ -45,7 +45,10 @@ class Settings(BaseSettings):
     # while correlation behaviour was being tuned against rules-only output.
     # Verify against https://ai.google.dev/gemini-api/docs/models before changing.
     corr_cheap_model: str = "gemini-3.6-flash"   # was gemini-2.0-flash (shut down)
-    corr_smart_model: str = "gemini-2.5-pro"     # was gemini-1.5-pro (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.

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

@@ -25,6 +25,7 @@ transcription.py) need the exact same judgment call and must not each grow
 their own slightly-different copy that drifts.
 """
 import asyncio
+from contextvars import ContextVar
 from datetime import datetime, timezone
 from typing import Optional
 
@@ -59,6 +60,14 @@ def _default_state() -> dict:
 
 _state: dict[str, dict] = {t: _default_state() for t in TIERS}
 
+# Set by a replay run to a list it owns; report_degraded appends there instead
+# of touching _state while inside a sandbox (see app/internal/replay.py).
+_sandbox_failures: ContextVar[Optional[list]] = ContextVar("drb_ai_sandbox_failures", default=None)
+
+
+def collect_sandbox_failures(sink: Optional[list]):
+    return _sandbox_failures.set(sink)
+
 
 def classify(text: str) -> str:
     """
@@ -112,6 +121,11 @@ async def report_degraded(
     if fstore.in_sandbox():
         # A replay's rate limits are not a live outage, and must never page
         # the AI-alert webhook or flip /health/ai (app/internal/replay.py).
+        # They are the run's own problem, so they go to the run instead.
+        sink = _sandbox_failures.get()
+        if sink is not None:
+            sink.append({"tier": tier, "provider": provider, "model": model,
+                         "problem": problem, "permanent": permanent})
         return
     if tier not in _state:
         _state[tier] = _default_state()

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

@@ -224,6 +224,28 @@ def _normalize_unit(unit: str) -> str:
     return key or unit.strip().lower()
 
 
+def _is_trackable_unit(unit: str) -> bool:
+    """
+    Whether a unit is concrete enough to hold an incident open until it clears.
+
+    Extraction lists everything that sounds like a unit — "Desk", "Central",
+    "Division", "sergeant", "unknown", and plate phonetics ("John Henry
+    Zebra"). None of those ever transmit a 10-8, so while they sat in
+    units_active the all-clear gate below could never pass: in the first
+    replay (server-26#170, 09-22 10:00-12:00) 0 of 19 incidents resolved on
+    a clear and every one had such a name in units_active. A real radio unit
+    ID carries a number ("45-9", "11-Adam 2", "Whitestone 1", "E-14"), so
+    only those gate resolution. The others are still kept in `units` and
+    still match for correlation.
+    """
+    if _TEN_CODE_RE.match((unit or "").strip()):
+        return False  # "10-8" read back as a unit ID is the status, not a unit
+    return any(ch.isdigit() for ch in unit or "")
+
+
+_TEN_CODE_RE = re.compile(r"^10[\s-]?\d{1,2}$")
+
+
 def _unit_keys(units: Optional[list[str]]) -> set[str]:
     """Comparison keys for a unit list, empties dropped."""
     return {k for k in (_normalize_unit(u) for u in (units or [])) if k}
@@ -1910,11 +1932,15 @@ def _apply_unit_clearance(inc: dict, cleared: list[str]) -> tuple[list[str], lis
     """
     units_active  = list(inc.get("units_active")  or [])
     units_cleared = list(inc.get("units_cleared") or [])
+    # Compared by normalised key: the unit that cleared as "11-Adam" is the
+    # one that went active as "11 Adam", and exact equality left it active.
+    cleared_keys = _unit_keys(cleared)
+    units_active = [u for u in units_active if _normalize_unit(u) not in cleared_keys]
+    known_cleared = _unit_keys(units_cleared)
     for u in cleared:
-        if u in units_active:
-            units_active.remove(u)
-        if u not in units_cleared:
+        if _normalize_unit(u) not in known_cleared:
             units_cleared.append(u)
+            known_cleared.add(_normalize_unit(u))
     auto_resolved = bool(units_cleared) and not units_active
     return units_active, units_cleared, auto_resolved
 
@@ -2009,9 +2035,11 @@ async def _update_incident(
     # units_active = units currently on scene; units_cleared = units back in service
     units_active  = list(inc.get("units_active")  or [])
     units_cleared = list(inc.get("units_cleared") or [])
+    tracked = _unit_keys(units_active) | _unit_keys(units_cleared)
     for u in call_units:
-        if u not in units_cleared and u not in units_active:
+        if _is_trackable_unit(u) and _normalize_unit(u) not in tracked:
             units_active.append(u)
+            tracked.add(_normalize_unit(u))
     inc_with_active_update = {**inc, "units_active": units_active, "units_cleared": units_cleared}
     units_active, units_cleared, _ = _apply_unit_clearance(inc_with_active_update, cleared_units or [])
 
@@ -2143,7 +2171,7 @@ async def _create_incident(
         "system_ids":        [system_id] if system_id else [],
         "tags":              tags + ["auto-generated"],
         "units":             call_units,
-        "units_active":      list(call_units),
+        "units_active":      [u for u in call_units if _is_trackable_unit(u)],
         "units_cleared":     [],
         "vehicles":          call_vehicles,
         "srcaddrs":          [call_srcaddr] if call_srcaddr else [],

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

@@ -15,6 +15,7 @@ import re
 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
 # Location validity is defined once, by the module that owns the incident's
@@ -246,18 +247,26 @@ async def extract_scenes(
             f"Intelligence: call {call_id} — transcript too short for extraction "
             f"({len(transcript.split())} words), skipping"
         )
+        cleared_unit = _short_clearance_unit(transcript)
         try:
             # Severity is still recorded: a five-word acknowledgement is genuinely
             # routine traffic, and downstream code treats a missing severity as
             # "not yet processed" rather than "nothing happened".
-            await fstore.doc_set("calls", call_id, {
+            updates = {
                 "skip_reason": "transcript_too_short",
                 "severity":    "routine",
                 "chatter_classifier_verdict": chatter_is_chatter,
                 "chatter_classifier_reason":  chatter_reason,
-            })
+            }
+            if cleared_unit:
+                updates["units"] = [cleared_unit]
+                updates["cleared_units"] = [cleared_unit]
+            await fstore.doc_set("calls", call_id, updates)
         except Exception:
             pass
+        if cleared_unit:
+            logger.info(f"Intelligence: call {call_id} — short clearance from {cleared_unit!r}")
+            return [_clearance_scene(transcript, cleared_unit)]
         return []
 
     try:
@@ -268,11 +277,26 @@ async def extract_scenes(
     except Exception:
         pass
 
-    raw_scenes: list[dict] = await asyncio.to_thread(
-        _sync_extract,
-        transcript, talkgroup_name, talkgroup_id, system_id, segments, vocabulary, ten_codes,
-        unit_format_hint,
-    )
+    try:
+        raw_scenes: list[dict] = await asyncio.to_thread(
+            _sync_extract,
+            transcript, talkgroup_name, talkgroup_id, system_id, segments, vocabulary, ten_codes,
+            unit_format_hint,
+        )
+    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}")
+        await ai_health.report_degraded(
+            "extraction", "openai", "gpt-4o-mini",
+            {"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"),
+            permanent=kind != "transient",
+        )
+        return []
+    await ai_health.report_healthy("extraction")
 
     if not raw_scenes:
         return []
@@ -453,6 +477,52 @@ async def extract_scenes(
     return processed
 
 
+# "45-9, I'm clear." / "Vehicle 1, clear." / "Car 12 10-8" — a unit reporting
+# itself back in service is the one signal that ends an incident, and it is
+# almost always five words or fewer, which is exactly the population the
+# too-short skip above keeps away from GPT. In the first replay
+# (server-26#170, 09-22 10:00-12:00 ET) 25 transmissions said 10-8/clear and
+# 2 reached cleared_units. Rule-based on purpose: no model call, and only a
+# unit named BEFORE the status word counts, so "10-8, 10-8." or "CMT clear."
+# (no number) clears nobody rather than guessing.
+_CLEAR_WORD_RE = re.compile(
+    r"\b(clear|10-?8|10-?98|back in service|in service|available)\b", re.IGNORECASE
+)
+_TEN_CODE_TOKEN_RE = re.compile(r"^10-?\d{1,2}$")
+_UNIT_PREFIX_WORDS = {"unit", "car", "vehicle", "engine", "ladder", "medic", "rescue", "post", "truck", "squad"}
+
+
+def _short_clearance_unit(transcript: str) -> Optional[str]:
+    m = _CLEAR_WORD_RE.search(transcript or "")
+    if not m:
+        return None
+    before = [t.strip(".,;:!?") for t in transcript[: m.start()].split()]
+    before = [t for t in before if t]
+    for i, tok in enumerate(before[:4]):
+        if not any(ch.isdigit() for ch in tok) or _TEN_CODE_TOKEN_RE.match(tok):
+            continue
+        prev = before[i - 1] if i else ""
+        if prev.lower() in _UNIT_PREFIX_WORDS:
+            return f"{prev} {tok}"
+        nxt = before[i + 1] if i + 1 < len(before) else ""
+        if nxt.isalpha() and nxt.lower() not in {"i'm", "im", "is", "are", "to", "we're", "copy"} \
+                and nxt[0].isupper():
+            return f"{tok} {nxt}"  # "11 Adam, clear"
+        return tok
+    return None
+
+
+def _clearance_scene(transcript: str, unit: str) -> dict:
+    """A minimal scene for a rule-parsed clearance: the unit, and nothing that
+    could make the incident-creation gate open a new incident for it."""
+    return {
+        "tags": [], "incident_type": None, "location": None, "location_coords": None,
+        "resolved": False, "severity": "routine", "vehicles": [], "units": [unit],
+        "cleared_units": [unit], "reassignment": False, "transcript": transcript,
+        "transcript_corrected": None, "segment_indices": [], "embedding": None,
+    }
+
+
 def _geo_dist_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
     """Haversine distance in km between two lat/lon points."""
     R = 6371.0
@@ -806,9 +876,11 @@ def _sync_extract(
     except json.JSONDecodeError as e:
         logger.warning(f"GPT-4o-mini returned non-JSON: {e}")
         return []
-    except Exception as e:
-        logger.warning(f"GPT-4o-mini extraction failed: {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
+    # ai_health. Swallowing it here made "OpenAI is down" indistinguishable
+    # from "nothing happened on the radio" — the extraction tier existed in
+    # /health/ai but nothing ever reported to it.
 
 
 def _sync_embed(text: str) -> Optional[list[float]]:

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

@@ -39,7 +39,7 @@ from datetime import datetime, timedelta, timezone
 from typing import Optional
 
 from app.config import settings
-from app.internal import clock
+from app.internal import ai_health, clock
 from app.internal import firestore as fstore
 from app.internal.feature_flags import force_flags, unforce_flags
 from app.internal.logger import logger
@@ -174,10 +174,16 @@ def _pipeline_time(call: dict) -> datetime:
     return _as_dt(call.get("ended_at")) or _call_time(call)
 
 
+def _duration_s(call: dict) -> float:
+    # Call docs carry no duration field; the node reports start and end.
+    start, end = _as_dt(call.get("started_at")), _as_dt(call.get("ended_at"))
+    return max(0.0, (end - start).total_seconds()) if start and end else 0.0
+
+
 def estimate(calls: list[dict], mode: str) -> dict:
     n = len(calls)
     with_transcript = sum(1 for c in calls if c.get("transcript_corrected") or c.get("transcript"))
-    audio_min = sum(float(c.get("duration_s") or 0) for c in calls) / 60
+    audio_min = sum(_duration_s(c) for c in calls) / 60
     with_audio = sum(1 for c in calls if c.get("audio_gcs_uri"))
     # Roughly a third of calls carry a geocodable location (09-22 dump: 92/373).
     per_call = USD_PER_EXTRACTION + USD_PER_LLM_CORRELATE + USD_PER_GEOCODE / 3
@@ -461,6 +467,8 @@ async def _run(run_id: str, org_id: str, calls: list[dict], mode: str,
 
     sb_token = fstore.enter_sandbox(sandbox_root(run_id))
     fl_token = force_flags(_flags_for(mode))
+    ai_failures: list = []
+    ai_token = ai_health.collect_sandbox_failures(ai_failures)
     try:
         sem = asyncio.Semaphore(PREFETCH)
 
@@ -482,6 +490,14 @@ async def _run(run_id: str, org_id: str, calls: list[dict], mode: str,
             if run_id in _cancel:
                 status = "cancelled"
                 break
+            fatal = _fatal_ai_failure(ai_failures)
+            if fatal:
+                # An unfunded or retired model fails every call the same way;
+                # finishing the run would only produce a sandbox of orphans
+                # that looks like a correlation result and isn't one.
+                status = "failed"
+                errors.append(f"aborted: {fatal}")
+                break
 
             t = _pipeline_time(call)
             last_t = t
@@ -519,7 +535,7 @@ async def _run(run_id: str, org_id: str, calls: list[dict], mode: str,
                     if prepared["transcript"] and mode != "reuse":
                         progress["extractions"] += 1
                 if mode == "audio":
-                    progress["audio_minutes"] += float(call.get("duration_s") or 0) / 60
+                    progress["audio_minutes"] += _duration_s(call) / 60
             except Exception as e:
                 progress["errors"] += 1
                 if len(errors) < 20:
@@ -544,12 +560,14 @@ async def _run(run_id: str, org_id: str, calls: list[dict], mode: str,
         sb_calls = await fstore.collection_list("calls")
         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))
     except Exception as e:
         status = "failed"
         errors.append(f"run: {type(e).__name__}: {e}"[:300])
         metrics = None
         logger.error(f"Replay {run_id} failed: {e}")
     finally:
+        ai_health._sandbox_failures.reset(ai_token)
         unforce_flags(fl_token)
         fstore.exit_sandbox(sb_token)
         _cancel.discard(run_id)
@@ -565,6 +583,21 @@ async def _run(run_id: str, org_id: str, calls: list[dict], mode: str,
     logger.info(f"Replay {run_id} {status}: {progress}")
 
 
+FATAL_AFTER = 5
+
+
+def _fatal_ai_failure(failures: list) -> Optional[str]:
+    """A tier that failed permanently (no credit, dead model) FATAL_AFTER times."""
+    permanent = Counter(
+        f"{f['tier']} ({f['provider']} {f['model']}): {f['problem']}"
+        for f in failures if f.get("permanent")
+    )
+    for what, n in permanent.items():
+        if n >= FATAL_AFTER:
+            return what
+    return None
+
+
 def _running_cost(progress: dict, metrics: dict, mode: str) -> float:
     usd = progress["audio_minutes"] * USD_WHISPER_PER_MIN
     if mode == "audio":

drb-c2-core/tests/test_clearance_tracking.py

@@ -0,0 +1,43 @@
+"""
+Dispatch→10-8 lifecycle, as measured by the first replay (server-26#170):
+0 of 19 incidents resolved on a clear although 25 transmissions said one.
+Three independent breaks, each pinned here.
+"""
+from app.internal import incident_correlator as ic
+from app.internal.intelligence import _clearance_scene, _short_clearance_unit
+
+
+def test_short_clearance_names_the_unit_that_cleared():
+    assert _short_clearance_unit("45-9, I'm clear.") == "45-9"
+    assert _short_clearance_unit("Vehicle 1, clear.") == "Vehicle 1"
+    assert _short_clearance_unit("11 Adam, clear") == "11 Adam"
+    assert _short_clearance_unit("Car 12 10-8") == "Car 12"
+
+
+def test_short_clearance_never_guesses():
+    for t in ("10-8, 10-8.", "CMT clear.", "10-8, I'm back now. Clear.",
+              "10-8, thank you.", "Show us 10-8, post 4.", "7, Charlie Central.", "10-4."):
+        assert _short_clearance_unit(t) is None, t
+
+
+def test_clearance_scene_cannot_open_an_incident():
+    scene = _clearance_scene("45-9, I'm clear.", "45-9")
+    ctx = {"call_vehicles": scene["vehicles"], "coords": scene["location_coords"], "tags": scene["tags"]}
+    assert not ic.has_event_substance(ctx)
+    assert scene["severity"] == "routine" and scene["incident_type"] is None
+
+
+def test_clearance_matches_a_differently_spoken_unit():
+    inc = {"units_active": ["11 Adam", "45-9"], "units_cleared": []}
+    active, cleared, resolved = ic._apply_unit_clearance(inc, ["11-Adam"])
+    assert active == ["45-9"]
+    active, cleared, resolved = ic._apply_unit_clearance(
+        {"units_active": active, "units_cleared": cleared}, ["45 9"])
+    assert active == [] and resolved
+
+
+def test_only_numbered_units_hold_an_incident_open():
+    for junk in ("Desk", "Central", "Division", "sergeant", "unknown", "John", "Zebra", "10-8", "10 4"):
+        assert not ic._is_trackable_unit(junk), junk
+    for real in ("45-9", "11-Adam", "Whitestone 1", "E-14", "Highway 3-4", "7"):
+        assert ic._is_trackable_unit(real), real

drb-c2-core/tests/test_replay.py

@@ -369,3 +369,44 @@ async def test_replay_never_touches_live_ai_health_or_review_queue():
     finally:
         fstore.exit_sandbox(tok)
     assert ai_health.snapshot() == before
+
+
+@pytest.mark.asyncio
+async def test_run_aborts_when_an_ai_account_is_dead(store):
+    """An unfunded OpenAI account made the first smoke run a sandbox of 290
+    orphans that looked like a result. A permanently failing tier now stops
+    the run and names the cause."""
+    store.data["calls"] = {
+        f"call-{i}": _live_call(i, i, "Car 12 responding to an MVA on Main Street") for i in range(1, 30)
+    }
+
+    def broke(*a, **kw):
+        raise RuntimeError("Error code: 429 - You exceeded your current quota (insufficient_quota)")
+
+    with patch("app.internal.intelligence._sync_extract", broke), \
+         patch("app.internal.intelligence.classify_chatter", return_value=(False, None)):
+        run = await replay.start_run(
+            org_id="org-1", date_from=T0 - timedelta(hours=1), date_to=T0 + timedelta(hours=1),
+            mode="transcripts", system_ids=None, source_run_id=None, label="", actor="t")
+        await replay._active_task
+
+    run = store.data["replay_runs"][run["run_id"]]
+    assert run["status"] == "failed"
+    assert any("out of credit" in e for e in run["errors"])
+    assert run["progress"]["done"] < 29
+
+
+@pytest.mark.asyncio
+async def test_live_extraction_failure_reports_to_ai_health():
+    from app.internal import ai_health, intelligence
+
+    def broke(*a, **kw):
+        raise RuntimeError("insufficient_quota")
+
+    with patch.object(intelligence, "_sync_extract", broke), \
+         patch.object(ai_health, "report_degraded") as degraded, \
+         patch.object(fstore, "doc_set"), patch.object(fstore, "doc_get_cached", return_value=None):
+        scenes = await intelligence.extract_scenes("c1", "Car 12 responding to an MVA on Main Street")
+    assert scenes == []
+    assert degraded.call_args.args[0] == "extraction"
+    assert degraded.call_args.kwargs["permanent"] is True

drb-frontend/components/admin/ReplayTab.tsx

@@ -324,7 +324,12 @@ function RunDetail({ run }: { run: ReplayRun }) {
   useEffect(() => {
     setData(null); setError(null);
     if (run.status === "running") return;
-    c2api.getReplayIncidents(run.run_id).then(setData).catch((e) => setError(String(e)));
+    c2api.getReplayIncidents(run.run_id).then((d) => {
+      setData(d);
+      // Exposed for in-page analysis (console / automation) of a run's
+      // sandbox — the same data this tab renders, nothing more.
+      (window as unknown as { __drbReplay?: unknown }).__drbReplay = { run, ...d };
+    }).catch((e) => setError(String(e)));
   }, [run.run_id, run.status]);
 
   const m = run.metrics;
@@ -356,6 +361,11 @@ function RunDetail({ run }: { run: ReplayRun }) {
           paths: {Object.entries(m.corr_path).map(([k, v]) => `${k} ${v}`).join(" · ")}
         </p>
       )}
+      {m?.ai_failures && Object.keys(m.ai_failures).length > 0 && (
+        <p className="text-xs font-mono text-amber-400">
+          AI failures: {Object.entries(m.ai_failures).map(([k, v]) => `${k} ×${v}`).join(" · ")}
+        </p>
+      )}
       {run.errors?.length > 0 && (
         <details className="text-xs font-mono text-red-400">
           <summary>{run.errors.length} error(s)</summary>

drb-frontend/lib/types.ts

@@ -339,6 +339,7 @@ export interface ReplayMetrics {
   corr_consensus: Record<string, number>;
   llm_decisions: number;
   est_cost_usd: number;
+  ai_failures?: Record<string, number>;
 }
 
 export interface ReplayRun {