""" 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 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. Never raises — failures are logged as warnings so the pipeline always completes. """ import uuid from datetime import datetime, timezone from typing import Optional from app.config import settings from app.internal.logger import logger from app.internal import firestore as fstore async def check_and_dispatch( call_id: str, node_id: str, talkgroup_id: Optional[int], talkgroup_name: Optional[str], tags: list[str], transcript: Optional[str], ) -> None: """ Check all enabled alert rules and fire events for any that match this call. """ try: # Scoped to the call's own org — an unscoped query here would let an # alert rule created by one org fire (and POST its Discord webhook) # on another org's radio traffic. org_id is resolved from the call # doc rather than threaded through as a new parameter, since every # caller of check_and_dispatch already has call_id and the call doc # is the single source of truth for a call's org once mqtt_handler.py # / upload.py have stamped it. None only for a call from a node that # predates tenancy and hasn't been through the backfill script yet — # such calls fall back to the pre-tenancy behaviour of checking # every rule regardless of org. call_doc = await fstore.doc_get("calls", call_id) org_id = (call_doc or {}).get("org_id") if org_id is not None: rules = await fstore.collection_list("alert_rules", enabled=True, org_id=org_id) else: rules = await fstore.collection_list("alert_rules", enabled=True) except Exception as e: logger.warning(f"Alerter: could not load rules: {e}") return # 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) if not matched_keywords: continue 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, "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, "acknowledged": False, "confidence": verdict["confidence"], "high_severity": verdict["high_severity"], "corroborating_calls": evidence["max_calls"], } try: await fstore.doc_set("alert_events", alert_id, event, merge=False) logger.info( f"Alert fired: rule='{rule.get('name')}' call={call_id} " f"keywords={matched_keywords}" ) except Exception as e: logger.warning(f"Alerter: could not save alert event: {e}") continue webhook_url = rule.get("discord_webhook") 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: """ Whether raw transcript text may be attached to an alert (server-26#85). Two gates, both of which must be open: 1. ``settings.alert_transcript_snippet_enabled`` -- the operator switch, default False, set from the environment and unreachable from the app. 2. ``alert_snippet_opt_in`` on the org document -- the customer's own explicit, contractual opt-in. Gate 1 exists because gate 2 alone is not a real control: the frontend reads and (per the Firestore rules, not ``auth.py``) can write org state directly from the browser, so an org owner could otherwise opt themselves into receiving person names lifted from live public-safety traffic. Board minutes #42 suppress names on every surface until E&O is bound. Fails CLOSED on any error, and on a call with no org (a pre-tenancy node that has not been backfilled), because the cost of wrongly withholding a snippet is a less informative alert and the cost of wrongly emitting one is unrecallable disclosure to a third party. """ if not settings.alert_transcript_snippet_enabled: return False if not org_id: return False try: org = await fstore.doc_get("organizations", org_id) except Exception as e: logger.warning( f"Alerter: could not read snippet opt-in for org={org_id}, " f"withholding transcript: {e}" ) return False return bool((org or {}).get("alert_snippet_opt_in")) def _match_rule( rule: dict, talkgroup_id: Optional[int], tags: list[str], transcript: Optional[str], ) -> list[str]: """Return list of matched keywords/reasons, or empty list if no match.""" matched: list[str] = [] # Talkgroup ID match rule_tg_ids = rule.get("talkgroup_ids", []) if rule_tg_ids and talkgroup_id is not None and talkgroup_id in rule_tg_ids: matched.append(f"talkgroup:{talkgroup_id}") # Keyword match against tags + transcript rule_keywords = [kw.lower() for kw in rule.get("keywords", [])] for kw in rule_keywords: if kw in tags: matched.append(kw) elif transcript and kw in transcript.lower(): matched.append(kw) return matched def _snippet(transcript: Optional[str], max_len: int = 200) -> Optional[str]: if not transcript: return None return transcript[:max_len] + ("…" if len(transcript) > max_len else "") async def _post_webhook( url: str, rule_name: str, talkgroup_name: Optional[str], matched_keywords: list[str], snippet: Optional[str], ) -> None: try: import httpx tg_label = talkgroup_name or "Unknown" kw_str = ", ".join(matched_keywords) body = ( f"**Alert: {rule_name}**\n" f"Talkgroup: {tg_label}\n" f"Matched: {kw_str}" ) if snippet: body += f"\n> {snippet}" async with httpx.AsyncClient(timeout=5.0) as client: await client.post(url, json={"content": body}) except Exception as e: logger.warning(f"Alerter: Discord webhook POST failed: {e}")