From ff0a89e9f1a8a460c3cfcccc7bff75ebada45844 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:29:01 -0400 Subject: [PATCH] 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 --- drb-c2-core/app/config.py | 24 +++ drb-c2-core/app/internal/alerter.py | 209 ++++++++++++++++++++++++++- drb-c2-core/app/routers/alerts.py | 21 +++ drb-c2-core/tests/test_alert_gate.py | 168 +++++++++++++++++++++ 4 files changed, 421 insertions(+), 1 deletion(-) create mode 100644 drb-c2-core/tests/test_alert_gate.py diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 66ff4fb..652872f 100644 --- a/drb-c2-core/app/config.py +++ b/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. diff --git a/drb-c2-core/app/internal/alerter.py b/drb-c2-core/app/internal/alerter.py index d4ca3cb..c4c6ab7 100644 --- a/drb-c2-core/app/internal/alerter.py +++ b/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: """ diff --git a/drb-c2-core/app/routers/alerts.py b/drb-c2-core/app/routers/alerts.py index ce3e769..327b484 100644 --- a/drb-c2-core/app/routers/alerts.py +++ b/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) diff --git a/drb-c2-core/tests/test_alert_gate.py b/drb-c2-core/tests/test_alert_gate.py new file mode 100644 index 0000000..49f4a86 --- /dev/null +++ b/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"]