correlator: use srcaddr for thin-call disambiguation instead of pure recency (#158)

The fast/thin path (short acks like "10-4", enabled to attempt linking at
all by 1ffff25) had no identity signal available -- transcript_too_short
skips GPT extraction entirely, so call_units is always empty for this
population. It fell back to "most recently updated incident on this
talkgroup", which silently misattaches a short ack to the wrong incident
whenever two are live on the same busy dispatch channel at once.

metadata_watcher.py already captures the P25 source radio ID (srcaddr) on
every call independent of transcript content, and it already reaches the
call doc (models.py, mqtt_handler.py:245) -- it was just never read by the
correlator. Thread it through _build_context, check it against the
srcaddrs already seen on each TG-matched incident before falling back to
recency, and accumulate it on the incident (_update_incident/_create_incident)
so later calls from the same radio can match.

Verified: 422 pass, 0 fail (4 new tests).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

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 = [
+            if srcaddr_matches:
-                inc for inc in tg_recent
+                thin_pool = [max(srcaddr_matches, key=lambda inc: inc.get("updated_at", ""))]
-                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", ""))]
                 logger.info(
-                    f"Correlator fast-path thin (tier-1, ≤{THIN_CONVERSATIONAL_SECS}s): "
+                    f"Correlator fast-path thin (srcaddr match): radio {call_srcaddr} "
-                    f"using most-recent of {len(very_recent)} candidate(s) for call {call_id}"
+                    f"already on {len(srcaddr_matches)} candidate(s) for call {call_id}"
                 )
             else:
-                # Tier 2: less certain — require a single candidate inside the
+                very_recent = [
-                # channel's thin window.
-                thin_pool = [
                     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"Correlator fast-path thin (tier-1, ≤{THIN_CONVERSATIONAL_SECS}s): "
-                        f"— ambiguous, skipping thin call {call_id}"
+                        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.
+                    # talkgroup, with no content to check a fit against — or,
-                    "corr_fit_signal": "thin_recency",
+                    # 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,

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