diff --git a/drb-c2-core/app/internal/incident_correlator.py b/drb-c2-core/app/internal/incident_correlator.py index 98d9393..fbeb4c4 100644 --- a/drb-c2-core/app/internal/incident_correlator.py +++ b/drb-c2-core/app/internal/incident_correlator.py @@ -875,11 +875,17 @@ async def _build_context( is_thin_call = _is_thin_call( call_units, call_vehicles, coords, tags, location, call_severity, reassignment ) + # server-26#158: the P25 source radio ID. Captured on every call by the + # edge node's metadata_watcher.py independent of transcript content, so + # it survives even when transcript_too_short skips GPT extraction + # entirely and leaves call_units empty — exactly the population the + # thin-call path below has no other identity signal for. + call_srcaddr = call_doc.get("srcaddr") return { "call_id": call_id, "org_id": org_id, "all_active": all_active, "recent": recent, "call_doc": call_doc, "call_embedding": call_embedding, - "scene_transcript": scene_transcript, + "scene_transcript": scene_transcript, "call_srcaddr": call_srcaddr, "call_units": call_units, "call_vehicles": call_vehicles, "call_cleared": call_cleared, "call_severity": call_severity, "coords": coords, "is_thin_call": is_thin_call, "now": now, @@ -933,6 +939,7 @@ def _run_decision(ctx: dict) -> dict: call_severity = ctx["call_severity"] coords = ctx["coords"] is_thin_call = ctx["is_thin_call"] + call_srcaddr = ctx.get("call_srcaddr") system_id = ctx["system_id"] talkgroup_id = ctx["talkgroup_id"] talkgroup_name = ctx["talkgroup_name"] @@ -1008,32 +1015,52 @@ def _run_decision(ctx: dict) -> dict: # incident idle up to tg_fast_path_idle_minutes (90) with no # single-candidate requirement and no fit test of any kind. Four # hours is not a bound, and neither is ninety minutes. + # server-26#158: identity beats guesswork. A thin call has no + # extracted units (GPT never ran), but it still carries the P25 + # radio ID that transmitted it — stronger, cheaper evidence than + # "most recently active" and immune to the exact failure this + # path exists to guard against: two incidents both live on one + # busy dispatch channel. If the radio that sent this call already + # has calls on one of the TG-matched incidents, that IS the + # thread, regardless of which incident is more recently updated + # or how many candidates are in the window. + srcaddr_matches = [ + inc for inc in tg_recent + if call_srcaddr and call_srcaddr in (inc.get("srcaddrs") or []) + ] THIN_CONVERSATIONAL_SECS = 30 thin_window_min = settings.tg_dispatch_thin_idle_minutes - very_recent = [ - inc for inc in tg_recent - if _idle_gate_minutes(inc, now) * 60 <= THIN_CONVERSATIONAL_SECS - ] - if very_recent: - # Tier 1: direct conversational reply — most recent wins. - thin_pool = [max(very_recent, key=lambda inc: inc.get("updated_at", ""))] + if srcaddr_matches: + thin_pool = [max(srcaddr_matches, key=lambda inc: inc.get("updated_at", ""))] logger.info( - f"Correlator fast-path thin (tier-1, ≤{THIN_CONVERSATIONAL_SECS}s): " - f"using most-recent of {len(very_recent)} candidate(s) for call {call_id}" + f"Correlator fast-path thin (srcaddr match): radio {call_srcaddr} " + f"already on {len(srcaddr_matches)} candidate(s) for call {call_id}" ) else: - # Tier 2: less certain — require a single candidate inside the - # channel's thin window. - thin_pool = [ + very_recent = [ inc for inc in tg_recent - if _idle_gate_minutes(inc, now) <= thin_window_min + if _idle_gate_minutes(inc, now) * 60 <= THIN_CONVERSATIONAL_SECS ] - if len(thin_pool) > 1: + if very_recent: + # Tier 1: direct conversational reply — most recent wins. + thin_pool = [max(very_recent, key=lambda inc: inc.get("updated_at", ""))] logger.info( - f"Correlator fast-path thin (tier-2): {len(thin_pool)} active incidents " - f"— ambiguous, skipping thin call {call_id}" + f"Correlator fast-path thin (tier-1, ≤{THIN_CONVERSATIONAL_SECS}s): " + f"using most-recent of {len(very_recent)} candidate(s) for call {call_id}" ) - thin_pool = [] + else: + # Tier 2: less certain — require a single candidate inside the + # channel's thin window. + thin_pool = [ + inc for inc in tg_recent + if _idle_gate_minutes(inc, now) <= thin_window_min + ] + if len(thin_pool) > 1: + logger.info( + f"Correlator fast-path thin (tier-2): {len(thin_pool)} active incidents " + f"— ambiguous, skipping thin call {call_id}" + ) + thin_pool = [] if not thin_pool: logger.info( @@ -1049,8 +1076,10 @@ def _run_decision(ctx: dict) -> dict: # no fit signal, so the admin debug view's "fit_signal # distribution" panel read empty on 95% of calls and looked # broken. Name what actually decided it: recency on this - # talkgroup, with no content to check a fit against. - "corr_fit_signal": "thin_recency", + # talkgroup, with no content to check a fit against — or, + # when the same radio ID already touched a candidate + # (server-26#158), that identity match instead of a guess. + "corr_fit_signal": "thin_srcaddr_match" if srcaddr_matches else "thin_recency", "corr_candidates": len(thin_pool), } logger.info( @@ -1511,6 +1540,7 @@ async def _apply_decision(decision: dict, ctx: dict) -> Optional[str]: call_cleared = ctx["call_cleared"] coords = ctx["coords"] now = ctx["now"] + call_srcaddr = ctx.get("call_srcaddr") incident_type = decision["incident_type"] if action == "link": @@ -1522,7 +1552,7 @@ async def _apply_decision(decision: dict, ctx: dict) -> Optional[str]: location, location_coords, call_units, call_vehicles, call_embedding, now, talkgroup_name=talkgroup_name, incident_type=incident_type, cleared_units=call_cleared, refresh_activity=not thin_link, - call_severity=call_severity, + call_severity=call_severity, call_srcaddr=call_srcaddr, ) return matched_incident["incident_id"] @@ -1554,6 +1584,7 @@ async def _apply_decision(decision: dict, ctx: dict) -> Optional[str]: call_id, org_id, incident_type, talkgroup_id, talkgroup_name, system_id, tags, location, location_coords, call_units, call_vehicles, call_embedding, call_severity, now, + call_srcaddr=call_srcaddr, ) if existing_master_id: @@ -1599,6 +1630,7 @@ async def _apply_decision(decision: dict, ctx: dict) -> Optional[str]: call_id, org_id, incident_type, talkgroup_id, talkgroup_name, system_id, tags, location, location_coords, call_units, call_vehicles, call_embedding, call_severity, now, + call_srcaddr=call_srcaddr, ) decision["corr_debug"]["corr_path"] = "new" @@ -1945,6 +1977,7 @@ async def _update_incident( cleared_units: Optional[list[str]] = None, refresh_activity: bool = True, call_severity: Optional[str] = None, + call_srcaddr: Optional[str] = None, ) -> None: incident_id = inc["incident_id"] @@ -1963,6 +1996,12 @@ async def _update_incident( merged_tags = list(dict.fromkeys((inc.get("tags") or []) + tags)) merged_units = list(dict.fromkeys((inc.get("units") or []) + call_units)) merged_vehicles = list(dict.fromkeys((inc.get("vehicles") or []) + call_vehicles)) + # server-26#158: accumulate every radio ID that has transmitted on this + # incident, so a later thin call from the same radio can identity-match + # instead of guessing off recency alone. + merged_srcaddrs = list(dict.fromkeys( + (inc.get("srcaddrs") or []) + ([call_srcaddr] if call_srcaddr else []) + )) # Unit activity tracking: units_active / units_cleared # units_active = units currently on scene; units_cleared = units back in service @@ -1993,6 +2032,7 @@ async def _update_incident( "tags": merged_tags, "units": merged_units, "vehicles": merged_vehicles, + "srcaddrs": merged_srcaddrs, "units_active": units_active, "units_cleared": units_cleared, "location_mentions": location_mentions, @@ -2059,6 +2099,7 @@ async def _create_incident( call_embedding: Optional[list], call_severity: str, now: datetime, + call_srcaddr: Optional[str] = None, ) -> str: incident_id = str(uuid.uuid4()) tg_label = ( @@ -2102,6 +2143,7 @@ async def _create_incident( "units_active": list(call_units), "units_cleared": [], "vehicles": call_vehicles, + "srcaddrs": [call_srcaddr] if call_srcaddr else [], "severity": call_severity, "summary": None, "summary_stale": True, diff --git a/drb-c2-core/tests/test_correlator_merge_caps.py b/drb-c2-core/tests/test_correlator_merge_caps.py index c932250..2767ab9 100644 --- a/drb-c2-core/tests/test_correlator_merge_caps.py +++ b/drb-c2-core/tests/test_correlator_merge_caps.py @@ -180,6 +180,55 @@ def test_tactical_thin_call_is_ambiguous_with_two_candidates(): assert decision["action"] == "orphan" +# --------------------------------------------------------------------------- +# server-26#158: srcaddr identity beats recency guesswork for thin calls +# --------------------------------------------------------------------------- + +def test_thin_call_srcaddr_match_resolves_tier2_ambiguity(): + """ + Same fixture as test_tactical_thin_call_is_ambiguous_with_two_candidates — + two candidates, tier-2 window, no unit ID parsed (transcript_too_short + skipped GPT). Without srcaddr this orphans. With it, the radio that sent + the call already touched inc-b, so that's the thread — not a guess. + """ + a = _incident(idle_minutes=3.0, incident_id="inc-a", srcaddrs=["9001"]) + b = _incident(idle_minutes=4.0, incident_id="inc-b", srcaddrs=["9002"]) + decision = _run_decision(_ctx( + all_active=[a, b], recent=[a, b], talkgroup_name=TACTICAL_TG, + call_srcaddr="9002", + )) + assert decision["action"] == "link" + assert decision["matched_incident"]["incident_id"] == "inc-b" + assert decision["corr_debug"]["corr_fit_signal"] == "thin_srcaddr_match" + + +def test_thin_call_srcaddr_match_overrides_recency_in_tier1(): + """ + Both candidates are inside the 30s conversational window, where recency + alone would pick inc-a (more recently updated) even though the radio that + sent this call has only ever touched inc-b — the exact busy-channel, + two-concurrent-incidents misattach server-26#158 was filed for. + """ + a = _incident(idle_minutes=0.1, incident_id="inc-a", srcaddrs=["9001"]) + b = _incident(idle_minutes=0.2, incident_id="inc-b", srcaddrs=["9002"]) + decision = _run_decision(_ctx( + all_active=[a, b], recent=[a, b], call_srcaddr="9002", + )) + assert decision["action"] == "link" + assert decision["matched_incident"]["incident_id"] == "inc-b" + + +def test_thin_call_with_no_srcaddr_match_falls_back_to_recency(): + """A radio ID that matches nothing on this talkgroup behaves exactly as + before — no regression for the ordinary case.""" + a = _incident(idle_minutes=0.1, incident_id="inc-a", srcaddrs=["9001"]) + decision = _run_decision(_ctx( + all_active=[a], recent=[a], call_srcaddr="unrelated-radio", + )) + assert decision["action"] == "link" + assert decision["corr_debug"]["corr_fit_signal"] == "thin_recency" + + def test_call_with_unit_overlap_does_attach(): """ Positive control: real evidence still links. Carrying units also means the