logan/server-26
1
0
Fork 0
Code Issues 57 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

correlator: remove is_dispatch and the tactical fit path entirely (#134)

Browse Source 
Full removal, not a hardcoded flag: _is_dispatch_channel, _DISPATCH_TG_RE, the is_dispatch parameter, and _call_fits_incident's tactical branch are gone. One evaluation path for every channel.

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

Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix
This commit is contained in:
Logan Cusano
2026-09-13 14:49:25 -04:00
co-authored by Claude Sonnet 5
parent 0473e6a583
commit 014f5e071f
9 changed files with 110 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
drb-c2-core/app/internal/incident_correlator.py
+63 -124
View File
@@ -98,13 +98,6 @@ _PURSUIT_PROXIMITY_KM = 20.0 # expanded radius for moving incidents
# overlap with the candidate OR a distance under this tighter bar.
_LOCATION_TIGHT_PROXIMITY_KM = 0.2
_DISPATCH_TG_RE = re.compile(
r"\bdispatch\b|\bdisp\b"
r"|\bpatched\b" # patched channels aggregate multiple call streams
r"|\bprimary\b", # "Primary" channels serve as shared backbones
re.IGNORECASE,
)
# Matches route/road identifiers in location strings for cross-system parent detection.
# Groups: numbered routes (Route 202, NY-9, US-6, I-87, CR-35) and named parkways/highways.
_ROAD_RE = re.compile(
@@ -494,13 +487,6 @@ def _resolve_incident_title(
return {}
def _is_dispatch_channel(talkgroup_name: Optional[str]) -> bool:
"""True when the talkgroup is a shared dispatch backbone (not a tactical/working channel)."""
if not talkgroup_name:
return False
return bool(_DISPATCH_TG_RE.search(talkgroup_name))
def _incident_idle_minutes(inc: dict, now: datetime) -> float:
"""Minutes since the incident was last updated (or started). Returns 9999 on parse error."""
try:
@@ -976,15 +962,10 @@ def _run_decision(ctx: dict) -> dict:
# directly on the Firestore call doc). Fall back to the call doc so that
# dispatch-channel strictness works regardless of how the call arrived.
effective_talkgroup_name = talkgroup_name or call_doc.get("talkgroup_name")
# server-26#134: always dispatch-strict now, not name-guessed — a
# channel labeled "tac"/"tactical" is rare and no less scrutinized in
# practice than any other. _call_fits_incident's tactical branch is
# kept, unreached, in case that's ever wrong.
is_dispatch = True
if effective_talkgroup_name != talkgroup_name:
logger.info(
f"Correlator: talkgroup_name missing from request for call {call_id}, "
f"resolved from call doc: {effective_talkgroup_name!r} → is_dispatch={is_dispatch}"
f"resolved from call doc: {effective_talkgroup_name!r}"
)
tg_matches = [
@@ -1028,10 +1009,7 @@ def _run_decision(ctx: dict) -> dict:
# single-candidate requirement and no fit test of any kind. Four
# hours is not a bound, and neither is ninety minutes.
THIN_CONVERSATIONAL_SECS = 30
thin_window_min = (
settings.tg_dispatch_thin_idle_minutes if is_dispatch
else settings.tg_thin_idle_minutes
)
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
@@ -1053,8 +1031,7 @@ def _run_decision(ctx: dict) -> dict:
if len(thin_pool) > 1:
logger.info(
f"Correlator fast-path thin (tier-2): {len(thin_pool)} active incidents "
f"on {'dispatch' if is_dispatch else 'tactical'} channel — "
f"ambiguous, skipping thin call {call_id}"
f"— ambiguous, skipping thin call {call_id}"
)
thin_pool = []
@@ -1084,14 +1061,14 @@ def _run_decision(ctx: dict) -> dict:
candidate = tg_recent[0]
logger.info(
f"Correlator fast/single: call {call_id} vs incident {candidate['incident_id']} "
f"tg_name={effective_talkgroup_name!r} is_dispatch={is_dispatch} "
f"tg_name={effective_talkgroup_name!r} "
f"idle={round(_incident_idle_minutes(candidate, now), 1)}min "
f"call_units={call_units} inc_units={candidate.get('units')} "
f"call_coords={'yes' if coords else 'no'} inc_coords={'yes' if candidate.get('location_coords') else 'no'}"
)
fit, fit_signal = _call_fits_incident(
candidate, call_units, call_vehicles, coords,
settings.location_proximity_km, is_dispatch=is_dispatch,
settings.location_proximity_km,
call_embedding=call_embedding, now=now,
reassignment=reassignment,
)
@@ -1101,13 +1078,12 @@ def _run_decision(ctx: dict) -> dict:
"corr_path": "fast/single",
"corr_incident_idle_min": round(_incident_idle_minutes(candidate, now), 1),
"corr_fit_signal": fit_signal,
"corr_is_dispatch": is_dispatch,
}
if fit_signal == "unit_overlap" and call_units:
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})"
f"(signal={fit_signal})"
)
else:
logger.info(
@@ -1123,14 +1099,14 @@ def _run_decision(ctx: dict) -> dict:
# dispatch channel should create its own incident, not be force-merged.
logger.info(
f"Correlator fast/disambig: call {call_id} vs incident {candidate['incident_id']} "
f"tg_name={effective_talkgroup_name!r} is_dispatch={is_dispatch} "
f"tg_name={effective_talkgroup_name!r} "
f"idle={round(_incident_idle_minutes(candidate, now), 1)}min "
f"call_units={call_units} inc_units={candidate.get('units')} "
f"call_coords={'yes' if coords else 'no'} inc_coords={'yes' if candidate.get('location_coords') else 'no'}"
)
fit, fit_signal = _call_fits_incident(
candidate, call_units, call_vehicles, coords,
settings.location_proximity_km, is_dispatch=is_dispatch,
settings.location_proximity_km,
call_embedding=call_embedding, now=now,
reassignment=reassignment,
)
@@ -1141,7 +1117,6 @@ def _run_decision(ctx: dict) -> dict:
"corr_incident_idle_min": round(_incident_idle_minutes(candidate, now), 1),
"corr_candidates": len(tg_recent),
"corr_fit_signal": fit_signal,
"corr_is_dispatch": is_dispatch,
}
if fit_signal == "unit_overlap" and call_units:
corr_debug["corr_matched_units"] = _matching_units(call_units, candidate.get("units"))
@@ -1720,7 +1695,6 @@ def _call_fits_incident(
call_vehicles: list[str],
call_coords: Optional[dict],
proximity_km: float,
is_dispatch: bool = False,
call_embedding: Optional[list] = None,
now: Optional[datetime] = None,
reassignment: bool = False,
@@ -1730,48 +1704,24 @@ def _call_fits_incident(
the incident; signal names the specific evidence that drove the decision.
fits=True signals: "unit_overlap" | "vehicle_overlap" | "location_proximity"
| "time_fallback" | "tactical_default"
fits=False signals: "unit_loc_conflict" | "content_divergence"
| "location_conflict" | "no_signal" | "tactical_idle"
| "location_conflict" | "no_signal"
Original docstring (logic unchanged):
Return True if this call plausibly belongs to the given incident.
Evaluation order for dispatch channels (is_dispatch=True):
─────────────────────────────────────────────────────────
1. Unit overlap
Same officer = same call. On dispatch channels, also run a location
conflict guard: if both sides carry geocoded coords and they differ
significantly, the officer has moved to a new scene and the unit match
is a false positive.
When the call has NO geocoded coordinates AND the incident is old
(≥ 15 min), use content divergence as a location proxy: an officer at
a genuinely different scene will be talking about clearly different
things. For recent incidents (< 15 min) we skip this proxy — the
officer may simply be giving an update without mentioning the address.
2. Vehicle overlap → True
3. Location proximity
Both sides geocoded and close → True.
Both sides geocoded and far apart (no other positive signal) → False.
4. No positive signals fired → dispatch fallback
a. Conversational continuity: idle < 2 min → True.
A call arriving within 2 minutes of the last incident activity almost
certainly belongs to the same dispatch thread. "Baker, head over
there too" or "copy that" carries no incident-specific vocabulary but
is unambiguously a response to what was just said. We do not require
embedding similarity here — embeddings capture word meaning, not
conversational context, and short operational messages will always
have lower similarity than the incident's accumulated content.
b. Older incident, no positive signals → False.
A shared dispatch channel must not absorb calls by default.
Tactical / working channel (is_dispatch=False):
────────────────────────────────────────────────
Channel is dedicated to one scene. No evidence of separation ≈ same call.
Default → True.
Evaluation order:
1. Unit overlap. Same officer = same call. Also runs a location-conflict
guard: geocoded on both sides and clearly different → the officer has
moved to a new scene, false positive. No geocode on the call (or on
the incident, with a call geocode) AND the incident is old (≥ 15 min)
→ content divergence as a location proxy (embedding similarity < 0.82
→ different scene). Skipped for recent incidents (< 15 min) — an
update without re-stating the address is normal.
2. Vehicle overlap → True.
3. Location proximity. Both geocoded and close → True; far apart with no
other positive signal → False.
4. No positive signal at all → False. A shared channel must not absorb
calls by default (server-26#134 — this used to default True within
20 min on any channel not name-matched as "dispatch"; a talkgroup
named tac/tactical is no less scrutinized in practice than any other).
Thin calls (no units/vehicles/coords) never reach this function —
they are intercepted before it in correlate_call.
@@ -1791,44 +1741,43 @@ def _call_fits_incident(
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:
# Hard location conflict: geocoded on both sides and clearly different.
inc_coords_u = inc.get("location_coords")
if inc_coords_u:
dist_km = _haversine_km(
call_coords["lat"], call_coords["lng"],
inc_coords_u["lat"], inc_coords_u["lng"],
)
if dist_km > proximity_km:
logger.info(f" fits[{inc_id}]: unit_overlap({matched_units}) but location_conflict dist={dist_km:.2f}km → unit_loc_conflict")
return False, "unit_loc_conflict"
elif call_embedding and idle_min >= 15:
# Call has geocode but incident doesn't — fall back to content
# divergence as a location proxy. Without this, stale incidents
# that never geocoded absorb unrelated calls purely on unit
# overlap (e.g. a patrol officer working a second scene 70 min
# after the original call).
inc_emb_u = inc.get("embedding")
if inc_emb_u:
sim = _cosine_similarity(call_embedding, inc_emb_u)
if sim < 0.82:
logger.info(f" fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence (has_call_coords/no_inc_coords) sim={sim:.3f} → content_divergence")
return False, "content_divergence"
if call_coords:
# Hard location conflict: geocoded on both sides and clearly different.
inc_coords_u = inc.get("location_coords")
if inc_coords_u:
dist_km = _haversine_km(
call_coords["lat"], call_coords["lng"],
inc_coords_u["lat"], inc_coords_u["lng"],
)
if dist_km > proximity_km:
logger.info(f" fits[{inc_id}]: unit_overlap({matched_units}) but location_conflict dist={dist_km:.2f}km → unit_loc_conflict")
return False, "unit_loc_conflict"
elif call_embedding and idle_min >= 15:
# No geocode available AND old incident: use content divergence as a
# location-proxy veto. After 15+ minutes an officer at a completely
# different scene will be discussing clearly different content.
# Skip this for recent incidents — an officer updating on the same
# scene without re-stating the address is normal and their update
# won't share much vocabulary with the original dispatch.
# Call has geocode but incident doesn't — fall back to content
# divergence as a location proxy. Without this, stale incidents
# that never geocoded absorb unrelated calls purely on unit
# overlap (e.g. a patrol officer working a second scene 70 min
# after the original call).
inc_emb_u = inc.get("embedding")
if inc_emb_u:
sim = _cosine_similarity(call_embedding, inc_emb_u)
if sim < 0.82:
logger.info(f" fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence sim={sim:.3f} → content_divergence")
logger.info(f" fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence (has_call_coords/no_inc_coords) sim={sim:.3f} → content_divergence")
return False, "content_divergence"
logger.info(f" fits[{inc_id}]: unit_overlap matched={matched_units} is_dispatch={is_dispatch} → unit_overlap")
elif call_embedding and idle_min >= 15:
# No geocode available AND old incident: use content divergence as a
# location-proxy veto. After 15+ minutes an officer at a completely
# different scene will be discussing clearly different content.
# Skip this for recent incidents — an officer updating on the same
# scene without re-stating the address is normal and their update
# won't share much vocabulary with the original dispatch.
inc_emb_u = inc.get("embedding")
if inc_emb_u:
sim = _cosine_similarity(call_embedding, inc_emb_u)
if sim < 0.82:
logger.info(f" fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence sim={sim:.3f} → content_divergence")
return False, "content_divergence"
logger.info(f" fits[{inc_id}]: unit_overlap matched={matched_units} → unit_overlap")
return True, "unit_overlap"
# ── 2. Vehicle overlap ────────────────────────────────────────────────────
@@ -1882,29 +1831,19 @@ def _call_fits_incident(
return False, "location_conflict"
# ── 4. No positive signals ────────────────────────────────────────────────
# Requires at least one positive signal (unit, vehicle, or location match).
# A substantive call with no matching signals is more likely a separate
# incident than a follow-up — two dispatches can arrive within seconds of
# each other on a busy channel. Content-free thin calls are handled before
# this function via the thin path in correlate_call, with a tighter
# 30-second recency window.
logger.info(
f" fits[{inc_id}]: no positive signal — is_dispatch={is_dispatch} idle={idle_min:.1f}min "
f" fits[{inc_id}]: no positive signal — idle={idle_min:.1f}min "
f"inc_units={list(inc_units)} call_units={call_units} "
f"inc_vehicles={list(inc_vehicles)} call_vehicles={call_vehicles} "
f"call_coords={call_coords is not None} inc_coords={inc_coords is not None}"
)
if is_dispatch:
# Dispatch channels require at least one positive signal (unit, vehicle,
# or location match). A substantive call with no matching signals is more
# likely a separate incident than a follow-up to the current one — two
# dispatches can arrive within seconds of each other on a busy channel.
# Content-free thin calls are handled before this function via the thin
# path in correlate_call, with a tighter 30-second recency window.
return False, "no_signal"
# Tactical channel: one scene per channel.
# Within 20 min of the last incident activity, link by default — same
# working channel almost certainly means same scene.
# After 20 min of silence, require at least one positive signal; the same
# frequency can be reused for a new unrelated incident later in the shift.
if idle_min < 20.0:
return True, "tactical_default"
return False, "tactical_idle"
return False, "no_signal"
async def _update_incident(