diff --git a/drb-c2-core/app/internal/incident_correlator.py b/drb-c2-core/app/internal/incident_correlator.py index 4c3bda0..32f16f1 100644 --- a/drb-c2-core/app/internal/incident_correlator.py +++ b/drb-c2-core/app/internal/incident_correlator.py @@ -135,6 +135,62 @@ _TAG_TYPE_HINTS: dict[str, str] = { } +# Words that describe a unit's role rather than identifying it. Whisper hears +# the same officer as "Post 5", "5", and "5 post" within one conversation, so +# these carry no distinguishing information and are dropped before comparison. +_UNIT_NOISE_TOKENS = frozenset({"post", "unit", "units", "car"}) +_ORDINAL_RE = re.compile(r"^(\d+)(?:st|nd|rd|th)$") + + +def _normalize_unit(unit: str) -> str: + """ + Reduce a spoken unit ID to a comparison key. + + Dispatch audio names the same unit inconsistently and exact string equality + silently drops the follow-ups. Observed on 2026-08-17 in a single hour, each + pair being one unit that failed to match itself: + + "K-9A2" vs "K-9-A-2" punctuation + "5-1-6" vs "516" digits read out individually + "37" vs "37th Post" ordinal + role word + "11-Victor" vs "11 Victor" + + Case, punctuation and role words go; digit groups join up. What is + deliberately NOT done is matching a bare district letter — "Adam" is not + treated as "6-Adam", because every district has an Adam and collapsing them + would merge unrelated incidents. That costs a few links and is the right + trade: a missed link leaves an orphan the sweep can retry, a false link + corrupts an incident permanently. + """ + tokens = [t for t in re.split(r"[^a-z0-9]+", unit.strip().lower()) if t] + cleaned: list[str] = [] + for tok in tokens: + if tok in _UNIT_NOISE_TOKENS: + continue + ordinal = _ORDINAL_RE.match(tok) + cleaned.append(ordinal.group(1) if ordinal else tok) + key = "".join(cleaned) + # A unit made only of noise words ("Post") would normalise to "" and then + # collide with every other such unit, so fall back to the raw text. + return key or unit.strip().lower() + + +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} + + +def _matching_units(call_units: Optional[list[str]], inc_units: Optional[list[str]]) -> list[str]: + """ + Call-side units that also appear on the incident, compared by normalised key + but returned as the original spoken strings so debug output stays readable. + """ + inc_keys = _unit_keys(inc_units) + if not inc_keys: + return [] + return [u for u in (call_units or []) if _normalize_unit(u) in inc_keys] + + 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: @@ -461,8 +517,7 @@ def _run_decision(ctx: dict) -> dict: "corr_is_dispatch": is_dispatch, } if fit_signal == "unit_overlap" and call_units: - inc_unit_set = set(candidate.get("units") or []) - corr_debug["corr_matched_units"] = [u for u in call_units if u in inc_unit_set] + 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}, is_dispatch={is_dispatch})" @@ -502,8 +557,7 @@ def _run_decision(ctx: dict) -> dict: "corr_is_dispatch": is_dispatch, } if fit_signal == "unit_overlap" and call_units: - inc_unit_set = set(candidate.get("units") or []) - corr_debug["corr_matched_units"] = [u for u in call_units if u in inc_unit_set] + 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})" @@ -524,11 +578,11 @@ def _run_decision(ctx: dict) -> dict: # incident, the officer has moved on and we don't link back to the old one. # This correctly handles officers dispatched to a second call mid-shift. if not matched_incident and call_units and system_id and not reassignment: - call_unit_set = set(call_units) + call_unit_set = _unit_keys(call_units) unit_candidates = [ inc for inc in all_active if system_id in (inc.get("system_ids") or []) - and call_unit_set & set(inc.get("units") or []) + and call_unit_set & _unit_keys(inc.get("units")) ] # Apply idle cap: units get reassigned; a 20+ min gap means the officer # has almost certainly moved on or the incident closed. @@ -540,7 +594,7 @@ def _run_decision(ctx: dict) -> dict: best_unit_inc = max(unit_candidates, key=lambda i: i.get("updated_at", "")) reassigned_away = any( inc["incident_id"] != best_unit_inc["incident_id"] - and call_unit_set & set(inc.get("units") or []) + and call_unit_set & _unit_keys(inc.get("units")) and inc.get("updated_at", "") > best_unit_inc.get("updated_at", "") for inc in all_active ) @@ -615,7 +669,7 @@ def _run_decision(ctx: dict) -> dict: # • embedding similarity >= cross-TG threshold (same subject matter) # Requiring 2+ shared units prevents single-officer false positives. if not matched_incident and call_embedding and incident_type and call_units and system_id: - call_unit_set = set(call_units) + call_unit_set = _unit_keys(call_units) best_cross_score = 0.0 best_cross_inc: Optional[dict] = None for inc in recent: @@ -623,7 +677,7 @@ def _run_decision(ctx: dict) -> dict: continue if system_id not in (inc.get("system_ids") or []): continue - inc_units_set = set(inc.get("units") or []) + inc_units_set = _unit_keys(inc.get("units")) if len(call_unit_set & inc_units_set) < 2: continue inc_embedding = inc.get("embedding") @@ -635,7 +689,7 @@ def _run_decision(ctx: dict) -> dict: best_cross_inc = inc if best_cross_inc and best_cross_score >= settings.embedding_cross_tg_threshold: matched_incident = best_cross_inc - shared = len(call_unit_set & set(best_cross_inc.get("units") or [])) + shared = len(call_unit_set & _unit_keys(best_cross_inc.get("units"))) corr_debug = { "corr_path": "cross-tg", "corr_score": round(best_cross_score, 4), @@ -951,8 +1005,8 @@ def _disambiguate( elif idle_min < 15: score += 1.0 # > 15 min: no bonus — older incidents compete on content/units only - inc_units = set(inc.get("units") or []) - if inc_units and call_units and any(u in inc_units for u in call_units): + inc_units = _unit_keys(inc.get("units")) + if inc_units and call_units and (inc_units & _unit_keys(call_units)): score += 10.0 inc_vehicles = set(inc.get("vehicles") or []) @@ -1047,8 +1101,8 @@ def _call_fits_incident( inc_id = inc.get("incident_id", "?") # ── 1. Unit overlap ─────────────────────────────────────────────────────── - inc_units = set(inc.get("units") or []) - matched_units = [u for u in call_units if u in inc_units] if (inc_units and call_units) else [] + inc_units = _unit_keys(inc.get("units")) + matched_units = _matching_units(call_units, inc.get("units")) if matched_units: if is_dispatch: if call_coords: diff --git a/drb-c2-core/tests/test_correlator_gate.py b/drb-c2-core/tests/test_correlator_gate.py index e001e5d..7d628b7 100644 --- a/drb-c2-core/tests/test_correlator_gate.py +++ b/drb-c2-core/tests/test_correlator_gate.py @@ -16,7 +16,9 @@ Firestore. _update_incident writes, so its test patches fstore. import pytest from datetime import datetime, timedelta, timezone from unittest.mock import AsyncMock, patch -from app.internal.incident_correlator import _run_decision, _update_incident +from app.internal.incident_correlator import ( + _run_decision, _update_incident, _normalize_unit, _matching_units, +) NOW = datetime(2026, 8, 16, 21, 0, 0, tzinfo=timezone.utc) @@ -188,3 +190,60 @@ async def test_substantive_link_does_refresh_updated_at(): updates = mock_fstore.doc_set.await_args.args[2] assert updates["updated_at"] == NOW.isoformat() assert "last_thin_at" not in updates + + +# --------------------------------------------------------------------------- +# Unit-ID normalisation — dispatch names the same unit several ways +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("spoken,other", [ + ("K-9A2", "K-9-A-2"), # punctuation only + ("5-1-6", "516"), # digits read out individually + ("37", "37th Post"), # ordinal + role word + ("11-Victor", "11 Victor"), # hyphen vs space + ("Post 5", "5"), # bare role word + ("post 1-2", "Post 1-2"), # case +]) +def test_same_unit_spoken_differently_normalises_alike(spoken, other): + """Every pair here was observed as one real unit failing to match itself.""" + assert _normalize_unit(spoken) == _normalize_unit(other) + + +@pytest.mark.parametrize("a,b", [ + ("6-Adam", "Adam"), # every district has an Adam — must stay distinct + ("6-Adam", "7-Adam"), + ("11-Victor", "11-Xray"), + ("516", "517"), + ("3", "39"), +]) +def test_genuinely_different_units_stay_distinct(a, b): + assert _normalize_unit(a) != _normalize_unit(b) + + +def test_role_only_unit_does_not_collapse_to_empty(): + """ + "Post" is all noise words. Normalising it to "" would make every such unit + equal to every other, so it falls back to the raw text instead. + """ + assert _normalize_unit("Post") != "" + assert _normalize_unit("Post") != _normalize_unit("Unit") + + +def test_matching_units_reports_the_original_spoken_strings(): + """Debug output has to stay readable, so matches come back un-normalised.""" + assert _matching_units(["K-9A2", "6-Adam"], ["K-9-A-2"]) == ["K-9A2"] + + +def test_matching_units_empty_when_nothing_overlaps(): + assert _matching_units(["6-Adam"], ["7-Adam", "516"]) == [] + + +def test_normalised_units_link_a_call_that_exact_match_would_orphan(): + """End-to-end: the K-9A2 case that orphaned in the 2026-08-17 01:05Z dump.""" + inc = _incident(2.0) + inc["units"] = ["K-9-A-2"] + decision = _run_decision(_ctx( + all_active=[inc], recent=[inc], + call_units=["K-9A2"], is_thin_call=False, call_severity="routine", + )) + assert decision["action"] == "link"