Alerts: confidence gate before any push, held alerts, release on corroboration (server-26#300)

No confidence field exists anywhere in scene/incident data, so the alerter
derives one: match weight (talkgroup 1.0, extracted tag 0.8, raw transcript
0.6) scaled down only when stt_confidence is under ALERT_STT_REFERENCE (0.3),
zero for not_speech. Below ALERT_MIN_CONFIDENCE, or high severity (term list,
or a major incident) without ALERT_CORROBORATION_MIN_CALLS calls or
ALERT_HIGH_SEVERITY_MIN_CONFIDENCE, the match is written to alert_holds
(no Firestore rule, admin-only GET /alerts/held) and logged instead of pushed.
A needs_corroboration hold is released when a later call brings its incident
to the call minimum; a rule pushed by that call supersedes its hold.

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

drb-c2-core/app/config.py

@@ -163,6 +163,30 @@ class Settings(BaseSettings):
     # else's PII. Both gates must be open before any snippet leaves.
     alert_transcript_snippet_enabled: bool = False
     place_verify_max_per_call: int = 3
+    # ── Alert confidence gate (server-26#300) ─────────────────────────────────
+    # Nothing in the pipeline emits a confidence, so the alerter derives one:
+    # how the rule matched (talkgroup 1.0 / extracted tag 0.8 / raw transcript
+    # text only 0.5 -- the CrimeRadar "fire alarm" -> "firearm" failure was a
+    # transcript match) scaled down when Whisper's own per-token probability
+    # (calls.stt_confidence) is below alert_stt_reference. Below the floor ->
+    # held, never pushed. High-severity matches also need the incident to
+    # have alert_corroboration_min_calls calls, or a confidence at or above
+    # alert_high_severity_min_confidence; held ones are released when a
+    # later call corroborates the incident. Starting values, not measured.
+    alert_min_confidence: float = 0.5
+    alert_high_severity_min_confidence: float = 0.85
+    alert_corroboration_min_calls: int = 2
+    # Low on purpose: exp(avg_logprob) on vocoded radio audio is often well
+    # under 0.5 for speech that is right, and Groq/OpenAI logprobs are not on
+    # a shared scale. Only clearly garbled audio should pull the score down.
+    alert_stt_reference: float = 0.3
+    alert_high_severity_terms: list[str] = [
+        "shots-fired", "shots fired", "shooting", "shot", "gun", "firearm", "weapon",
+        "stabbing", "stabbed", "officer-down", "officer down", "10-13",
+        "officer-needs-assistance", "active-shooter", "hostage", "barricade",
+        "explosion", "structure-fire", "structure fire", "working-fire",
+        "entrapment", "cardiac-arrest", "vehicle-pursuit", "pursuit",
+    ]
     # How close a candidate has to sound before it may rewrite a transcript.
     # Below this, Places Text Search will confidently hand back the nearest
     # business for any garbage string.

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