correlator: gate LLM-orphan against rules-new instead of escalating to tiebreak (#115) #125
@@ -91,6 +91,13 @@ def _max_severity(current: Optional[str], new: Optional[str]) -> str:
|
|||||||
_MAX_PURSUIT_SPEED_KM_PER_MIN = 8.0 # ~300 km/h, intentionally generous
|
_MAX_PURSUIT_SPEED_KM_PER_MIN = 8.0 # ~300 km/h, intentionally generous
|
||||||
_PURSUIT_PROXIMITY_KM = 20.0 # expanded radius for moving incidents
|
_PURSUIT_PROXIMITY_KM = 20.0 # expanded radius for moving incidents
|
||||||
|
|
||||||
|
# server-26#115 — the location path linked on `location_proximity_km` (0.5 km)
|
||||||
|
# alone, with no unit or content check. In a dense village two unrelated events
|
||||||
|
# routinely geocode that close (a vehicle lockout stitched to a station-restroom
|
||||||
|
# slip; two different churches an hour apart). A location link now needs unit
|
||||||
|
# overlap with the candidate OR a distance under this tighter bar.
|
||||||
|
_LOCATION_TIGHT_PROXIMITY_KM = 0.2
|
||||||
|
|
||||||
_DISPATCH_TG_RE = re.compile(
|
_DISPATCH_TG_RE = re.compile(
|
||||||
r"\bdispatch\b|\bdisp\b"
|
r"\bdispatch\b|\bdisp\b"
|
||||||
r"|\bpatched\b" # patched channels aggregate multiple call streams
|
r"|\bpatched\b" # patched channels aggregate multiple call streams
|
||||||
@@ -239,6 +246,22 @@ def _matching_units(call_units: Optional[list[str]], inc_units: Optional[list[st
|
|||||||
return [u for u in (call_units or []) if _normalize_unit(u) in inc_keys]
|
return [u for u in (call_units or []) if _normalize_unit(u) in inc_keys]
|
||||||
|
|
||||||
|
|
||||||
|
def has_event_substance(ctx: dict) -> bool:
|
||||||
|
"""
|
||||||
|
True when the call carries content beyond who-was-speaking-and-where:
|
||||||
|
a vehicle, a geocode, or a tag.
|
||||||
|
|
||||||
|
This is the substance half of the incident-creation gate (see
|
||||||
|
`_run_decision`, "Severity, not type, decides..."), factored out so the
|
||||||
|
consensus LLM-orphan gate in routers/upload.py mirrors it exactly and can
|
||||||
|
never drop a call the creation gate would have opened. `call_units` and
|
||||||
|
`location` are deliberately excluded — radio protocol puts a unit ID and a
|
||||||
|
place name in almost every transmission, so counting them as substance
|
||||||
|
makes the check trivially true.
|
||||||
|
"""
|
||||||
|
return bool(ctx.get("call_vehicles") or ctx.get("coords") or ctx.get("tags"))
|
||||||
|
|
||||||
|
|
||||||
def _infer_type_from_tags(tags: list[str]) -> Optional[str]:
|
def _infer_type_from_tags(tags: list[str]) -> Optional[str]:
|
||||||
"""Return an incident type inferred from tags, or None if ambiguous."""
|
"""Return an incident type inferred from tags, or None if ambiguous."""
|
||||||
for tag in tags:
|
for tag in tags:
|
||||||
@@ -1173,6 +1196,10 @@ def _run_decision(ctx: dict) -> dict:
|
|||||||
|
|
||||||
# ── 2. Location path: proximity match (time-limited, cross-type) ─────────
|
# ── 2. Location path: proximity match (time-limited, cross-type) ─────────
|
||||||
if not matched_incident and coords:
|
if not matched_incident and coords:
|
||||||
|
# server-26#115 — score every in-radius candidate and link the NEAREST
|
||||||
|
# that carries corroboration, rather than whichever incident happened to
|
||||||
|
# come first in an unsorted `recent`.
|
||||||
|
loc_candidates: list[tuple] = []
|
||||||
for inc in recent:
|
for inc in recent:
|
||||||
inc_coords = inc.get("location_coords")
|
inc_coords = inc.get("location_coords")
|
||||||
if not inc_coords:
|
if not inc_coords:
|
||||||
@@ -1189,18 +1216,49 @@ def _run_decision(ctx: dict) -> dict:
|
|||||||
elapsed_min = max(_incident_idle_minutes(inc, now), 0.1)
|
elapsed_min = max(_incident_idle_minutes(inc, now), 0.1)
|
||||||
if (dist_km / elapsed_min) > _MAX_PURSUIT_SPEED_KM_PER_MIN:
|
if (dist_km / elapsed_min) > _MAX_PURSUIT_SPEED_KM_PER_MIN:
|
||||||
continue # implausible speed — skip this candidate
|
continue # implausible speed — skip this candidate
|
||||||
if dist_km <= radius:
|
if dist_km > radius:
|
||||||
|
continue
|
||||||
|
# server-26#115 — a bare sub-radius distance is not enough on its
|
||||||
|
# own. Require corroboration: unit overlap with the candidate, OR
|
||||||
|
# a much tighter proximity. Pursuit incidents keep their
|
||||||
|
# movement-speed-validated wide radius (they passed the speed
|
||||||
|
# check above), so they are exempt.
|
||||||
|
unit_overlap = bool(
|
||||||
|
_unit_keys(call_units) & _unit_keys(inc.get("units"))
|
||||||
|
)
|
||||||
|
tight_proximity = dist_km <= _LOCATION_TIGHT_PROXIMITY_KM
|
||||||
|
if not (is_pursuit_inc or unit_overlap or tight_proximity):
|
||||||
|
logger.info(
|
||||||
|
f"Correlator location-path skipped: call {call_id} vs "
|
||||||
|
f"{inc['incident_id']} — dist={dist_km:.2f}km within radius "
|
||||||
|
f"but no unit overlap and not tight-proximity "
|
||||||
|
f"(<= {_LOCATION_TIGHT_PROXIMITY_KM}km)"
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
loc_candidates.append((dist_km, unit_overlap, is_pursuit_inc, inc))
|
||||||
|
|
||||||
|
if loc_candidates:
|
||||||
|
loc_candidates.sort(key=lambda c: c[0])
|
||||||
|
dist_km, unit_overlap, is_pursuit_inc, inc = loc_candidates[0]
|
||||||
matched_incident = inc
|
matched_incident = inc
|
||||||
|
# Distinct from the fast path's "unit_overlap" so the admin
|
||||||
|
# corr_fit_signal histogram (routers/admin.py) does not merge a
|
||||||
|
# location-path link into the fast-path bucket (#35).
|
||||||
|
fit_signal = "location_unit_overlap" if unit_overlap else "location_proximity"
|
||||||
corr_debug = {
|
corr_debug = {
|
||||||
"corr_path": "location",
|
"corr_path": "location",
|
||||||
"corr_distance_km": round(dist_km, 3),
|
"corr_distance_km": round(dist_km, 3),
|
||||||
"corr_pursuit_mode": is_pursuit_inc,
|
"corr_pursuit_mode": is_pursuit_inc,
|
||||||
|
"corr_fit_signal": fit_signal,
|
||||||
}
|
}
|
||||||
|
if unit_overlap and call_units:
|
||||||
|
corr_debug["corr_matched_units"] = _matching_units(
|
||||||
|
call_units, inc.get("units")
|
||||||
|
)
|
||||||
logger.info(
|
logger.info(
|
||||||
f"Correlator location-path: call {call_id} → {inc['incident_id']} "
|
f"Correlator location-path: call {call_id} → {inc['incident_id']} "
|
||||||
f"(dist={dist_km:.2f}km, pursuit={is_pursuit_inc})"
|
f"(dist={dist_km:.2f}km, pursuit={is_pursuit_inc}, signal={fit_signal})"
|
||||||
)
|
)
|
||||||
break
|
|
||||||
|
|
||||||
# ── 2.5. Cross-TG path: same department, overlapping units, moderate similarity ──
|
# ── 2.5. Cross-TG path: same department, overlapping units, moderate similarity ──
|
||||||
#
|
#
|
||||||
@@ -1337,7 +1395,7 @@ def _run_decision(ctx: dict) -> dict:
|
|||||||
# each. A vehicle, a geocode, or a tag means the extractor found something
|
# each. A vehicle, a geocode, or a tag means the extractor found something
|
||||||
# beyond who was speaking and where they stood.
|
# beyond who was speaking and where they stood.
|
||||||
if not resolved_type:
|
if not resolved_type:
|
||||||
has_substance = bool(call_vehicles or coords or tags)
|
has_substance = has_event_substance(ctx)
|
||||||
if call_severity in ("minor", "moderate", "major") or has_substance:
|
if call_severity in ("minor", "moderate", "major") or has_substance:
|
||||||
resolved_type = "other"
|
resolved_type = "other"
|
||||||
logger.info(
|
logger.info(
|
||||||
|
|||||||
@@ -20,6 +20,22 @@ from app.internal.logger import logger
|
|||||||
from app.internal import firestore as fstore
|
from app.internal import firestore as fstore
|
||||||
from app.config import settings
|
from app.config import settings
|
||||||
|
|
||||||
|
# Standard link-only retry budget before a call is tombstoned corr_path="unlinked".
|
||||||
|
MAX_SWEEP_ATTEMPTS = 3
|
||||||
|
# server-26#115 — a call the consensus LLM-orphan gate parked (llm=orphan vs
|
||||||
|
# rules=new, no substance) gets a longer budget. The gate fires before any
|
||||||
|
# incident for the job may exist, so the substantive call that would justify
|
||||||
|
# linking can land well after the standard ~6 min. Still link-only: a genuinely
|
||||||
|
# thin call must not mint an incident, and the rules creation gate would re-orphan
|
||||||
|
# it anyway.
|
||||||
|
GATED_ORPHAN_SWEEP_ATTEMPTS = 10
|
||||||
|
|
||||||
|
|
||||||
|
def _max_sweep_attempts(call: dict) -> int:
|
||||||
|
if call.get("corr_consensus") == "llm_orphan_gate":
|
||||||
|
return GATED_ORPHAN_SWEEP_ATTEMPTS
|
||||||
|
return MAX_SWEEP_ATTEMPTS
|
||||||
|
|
||||||
|
|
||||||
async def recorrelation_loop() -> None:
|
async def recorrelation_loop() -> None:
|
||||||
interval = settings.summary_interval_minutes * 60
|
interval = settings.summary_interval_minutes * 60
|
||||||
@@ -46,10 +62,9 @@ async def _run_sweep_pass() -> None:
|
|||||||
("status", "==", "ended"),
|
("status", "==", "ended"),
|
||||||
("ended_at", ">=", cutoff),
|
("ended_at", ">=", cutoff),
|
||||||
])
|
])
|
||||||
# corr_path="unlinked" is written after MAX_SWEEP_ATTEMPTS failures.
|
# corr_path="unlinked" is written after the attempt budget is exhausted.
|
||||||
# Allows a few retries so a welfare-check call can link to an escalation
|
# Allows a few retries so a welfare-check call can link to an escalation
|
||||||
# incident that is created a few minutes later, without sweeping 30× forever.
|
# incident that is created a few minutes later, without sweeping 30× forever.
|
||||||
MAX_SWEEP_ATTEMPTS = 3
|
|
||||||
orphans = [
|
orphans = [
|
||||||
c for c in recent_ended
|
c for c in recent_ended
|
||||||
if not c.get("incident_ids") and not c.get("incident_id")
|
if not c.get("incident_ids") and not c.get("incident_id")
|
||||||
@@ -61,7 +76,7 @@ async def _run_sweep_pass() -> None:
|
|||||||
# the thin path minutes later and attached to whatever was most recent —
|
# the thin path minutes later and attached to whatever was most recent —
|
||||||
# a second route into the over-merge the thin fix above addresses.
|
# a second route into the over-merge the thin fix above addresses.
|
||||||
and not c.get("skip_reason")
|
and not c.get("skip_reason")
|
||||||
and c.get("corr_sweep_count", 0) < MAX_SWEEP_ATTEMPTS
|
and c.get("corr_sweep_count", 0) < _max_sweep_attempts(c)
|
||||||
]
|
]
|
||||||
|
|
||||||
if not orphans:
|
if not orphans:
|
||||||
@@ -120,12 +135,12 @@ async def _recorrelate_orphan(call: dict) -> bool:
|
|||||||
)
|
)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
# Increment the attempt counter. Once MAX_SWEEP_ATTEMPTS is reached the
|
# Increment the attempt counter. Once the budget is reached the orphan filter
|
||||||
# orphan filter above will stop picking this call up, and we write
|
# above will stop picking this call up, and we write corr_path="unlinked" as
|
||||||
# corr_path="unlinked" as a permanent tombstone.
|
# a permanent tombstone.
|
||||||
attempts = call.get("corr_sweep_count", 0) + 1
|
attempts = call.get("corr_sweep_count", 0) + 1
|
||||||
update: dict = {"corr_sweep_count": attempts}
|
update: dict = {"corr_sweep_count": attempts}
|
||||||
if attempts >= 3:
|
if attempts >= _max_sweep_attempts(call):
|
||||||
update["corr_path"] = "unlinked"
|
update["corr_path"] = "unlinked"
|
||||||
await fstore.doc_set("calls", call_id, update)
|
await fstore.doc_set("calls", call_id, update)
|
||||||
return False
|
return False
|
||||||
|
|||||||
@@ -100,6 +100,64 @@ async def upload_call_audio(
|
|||||||
return {"url": gcs_uri}
|
return {"url": gcs_uri}
|
||||||
|
|
||||||
|
|
||||||
|
# server-26#115 — the consensus LLM-orphan gate only fires when the call is
|
||||||
|
# genuinely substanceless. The earlier version tested `rules_decision["corr_debug"]`
|
||||||
|
# for a "positive signal", but corr_debug is EMPTY at preview time for
|
||||||
|
# action=="new" (corr_path:"new" is written at APPLY time), so that test was
|
||||||
|
# always False and the gate dropped real events — a major "extinguishing fire",
|
||||||
|
# geocoded calls, pursuit updates. The substance test now runs against `ctx`,
|
||||||
|
# which is fully populated at preview time.
|
||||||
|
|
||||||
|
|
||||||
|
def _recent_incident_on_same_talkgroup(ctx: dict) -> bool:
|
||||||
|
"""
|
||||||
|
True when one of the already-loaded recent incidents is running on this
|
||||||
|
call's own system + talkgroup. Covers the "unit dispatched on the dispatch
|
||||||
|
channel, thin acknowledgement 10-30s later" case: the ack carries no
|
||||||
|
substance of its own but plainly belongs to the job just opened.
|
||||||
|
|
||||||
|
Reads ctx["recent"] — the same window-filtered candidate list the rules
|
||||||
|
engine already loaded — so this adds no Firestore read.
|
||||||
|
"""
|
||||||
|
tg_id = ctx.get("talkgroup_id")
|
||||||
|
system_id = ctx.get("system_id")
|
||||||
|
if tg_id is None or not system_id:
|
||||||
|
return False
|
||||||
|
tg_str = str(tg_id)
|
||||||
|
for inc in ctx.get("recent") or []:
|
||||||
|
if system_id in (inc.get("system_ids") or []) and tg_str in (inc.get("talkgroup_ids") or []):
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _call_is_substanceless(ctx: dict) -> bool:
|
||||||
|
"""
|
||||||
|
True when the call carries nothing that marks it as a real event:
|
||||||
|
• severity is not moderate/major, AND
|
||||||
|
• no vehicle, geocode or tag (incident_correlator.has_event_substance —
|
||||||
|
the same predicate the incident-creation gate uses), AND
|
||||||
|
• no recent incident already running on the same talkgroup.
|
||||||
|
Only then may the LLM-orphan gate drop the call without a tiebreak.
|
||||||
|
"""
|
||||||
|
from app.internal import incident_correlator
|
||||||
|
|
||||||
|
# The incident-creation gate skips the has_event_substance check entirely
|
||||||
|
# when a type resolved (incident_correlator._run_decision ~:1397), so a
|
||||||
|
# typed call — fire/medical/etc. — opens an incident on substance we do not
|
||||||
|
# re-check here. reassignment=True is dispatch pulling a unit onto a NEW
|
||||||
|
# job (units are blanked at :296 for exactly that reason): the strongest
|
||||||
|
# new-incident signal in the pipeline. Either one means "keep the tiebreak".
|
||||||
|
if ctx.get("incident_type") or ctx.get("reassignment"):
|
||||||
|
return False
|
||||||
|
if (ctx.get("call_severity") or "routine") in ("moderate", "major"):
|
||||||
|
return False
|
||||||
|
if incident_correlator.has_event_substance(ctx):
|
||||||
|
return False
|
||||||
|
if _recent_incident_on_same_talkgroup(ctx):
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
async def _correlate_with_consensus(
|
async def _correlate_with_consensus(
|
||||||
call_id: str,
|
call_id: str,
|
||||||
node_id: str,
|
node_id: str,
|
||||||
@@ -151,6 +209,39 @@ async def _correlate_with_consensus(
|
|||||||
rules_decision["corr_debug"]["corr_llm_reasoning"] = llm_decision.get("reasoning", "")
|
rules_decision["corr_debug"]["corr_llm_reasoning"] = llm_decision.get("reasoning", "")
|
||||||
return await incident_correlator.apply_correlation(preview)
|
return await incident_correlator.apply_correlation(preview)
|
||||||
|
|
||||||
|
# server-26#115 — LLM-orphan gate.
|
||||||
|
# When the cheap LLM says `orphan`, the rules engine says `new`, and the call
|
||||||
|
# is genuinely substanceless (routine severity, no vehicle/geocode/tag, and
|
||||||
|
# no incident already running on this talkgroup), resolve to `orphan` and DO
|
||||||
|
# NOT pay for the smart tiebreaker. A bare rules `new` there means only
|
||||||
|
# "nothing to link to" — trivially true for radio housekeeping (check-ins,
|
||||||
|
# roll call, 10-8/10-98) — and the tiebreaker rubber-stamped it ~21/21 of the
|
||||||
|
# time on exactly this disagreement (CORRELATION_REVIEW_0907b.md). Any real
|
||||||
|
# signal (severity, coords, tags, a live same-talkgroup incident) still
|
||||||
|
# escalates, so an event the LLM misreads as orphan is not lost.
|
||||||
|
if (
|
||||||
|
llm_decision["action"] == "orphan"
|
||||||
|
and rules_decision["action"] == "new"
|
||||||
|
and _call_is_substanceless(ctx)
|
||||||
|
):
|
||||||
|
logger.info(
|
||||||
|
f"Consensus gate for call {call_id}: llm=orphan vs rules=new and call "
|
||||||
|
f"is substanceless — resolving orphan, skipping tiebreak"
|
||||||
|
)
|
||||||
|
gated = {
|
||||||
|
"action": "orphan",
|
||||||
|
"matched_incident": None,
|
||||||
|
"incident_type": None,
|
||||||
|
"corr_debug": dict(rules_decision.get("corr_debug") or {}),
|
||||||
|
}
|
||||||
|
gated["corr_debug"].update({
|
||||||
|
"corr_consensus": "llm_orphan_gate",
|
||||||
|
"corr_rules_action": rules_decision["action"],
|
||||||
|
"corr_llm_action": llm_decision["action"],
|
||||||
|
"corr_llm_reasoning": llm_decision.get("reasoning", ""),
|
||||||
|
})
|
||||||
|
return await incident_correlator.apply_correlation({"decision": gated, "ctx": ctx})
|
||||||
|
|
||||||
# Disagree — escalate to the smarter tiebreaker.
|
# Disagree — escalate to the smarter tiebreaker.
|
||||||
logger.info(
|
logger.info(
|
||||||
f"Consensus disagreement for call {call_id}: "
|
f"Consensus disagreement for call {call_id}: "
|
||||||
|
|||||||
@@ -0,0 +1,315 @@
|
|||||||
|
"""
|
||||||
|
server-26#115 — two consensus-quality fixes.
|
||||||
|
|
||||||
|
Fix 1 (routers/upload.py): when the cheap LLM says `orphan`, the rules engine
|
||||||
|
says `new`, and the call is genuinely SUBSTANCELESS (routine severity, no
|
||||||
|
vehicle/geocode/tag, and no incident already running on the same talkgroup),
|
||||||
|
resolve to `orphan` and DO NOT pay for the smart tiebreaker. Radio housekeeping
|
||||||
|
(unit check-ins, roll call, 10-8/10-98) was being promoted to incidents because
|
||||||
|
the tiebreaker rubber-stamped the rules `new` ~21/21 of the time
|
||||||
|
(CORRELATION_REVIEW_0907b.md).
|
||||||
|
|
||||||
|
The substance test runs against `ctx` (fully populated at preview time), NOT
|
||||||
|
against `rules_decision["corr_debug"]` — that dict is EMPTY at preview time for
|
||||||
|
action=="new" (corr_path:"new" is written at APPLY time), so the first version of
|
||||||
|
this gate fired on real events (a `major` "extinguishing fire", geocoded calls,
|
||||||
|
pursuit updates).
|
||||||
|
|
||||||
|
Fix 2 (incident_correlator.py): the `location` correlation path linked on a bare
|
||||||
|
sub-`location_proximity_km` (0.5 km) distance alone, taking whichever incident
|
||||||
|
came first in an unsorted `recent`. In a dense village two unrelated events
|
||||||
|
routinely geocode that close. A `location` link now needs unit overlap with the
|
||||||
|
candidate OR a distance under a tighter bar, and picks the NEAREST qualifying
|
||||||
|
candidate. A unit-overlap location link is tagged `location_unit_overlap` so it
|
||||||
|
does not merge into the fast path's bucket in the admin fit-signal histogram.
|
||||||
|
"""
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
from unittest.mock import AsyncMock, patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from app.routers import upload
|
||||||
|
from app.internal.incident_correlator import _run_decision, has_event_substance
|
||||||
|
|
||||||
|
NOW = datetime(2026, 9, 7, 21, 30, 0, tzinfo=timezone.utc)
|
||||||
|
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
# Fix 1 — the LLM-orphan gate in _correlate_with_consensus
|
||||||
|
# ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def _preview(action, corr_debug=None, ctx=None):
|
||||||
|
base_ctx = {"call_id": "call-1"}
|
||||||
|
if ctx:
|
||||||
|
base_ctx.update(ctx)
|
||||||
|
return {
|
||||||
|
"decision": {
|
||||||
|
"action": action,
|
||||||
|
"matched_incident": None,
|
||||||
|
"incident_type": "other" if action == "new" else None,
|
||||||
|
"corr_debug": {} if corr_debug is None else dict(corr_debug),
|
||||||
|
},
|
||||||
|
"ctx": base_ctx,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _llm(action, reasoning="—"):
|
||||||
|
md = {"incident_id": "inc-1"} if action == "link" else None
|
||||||
|
return {"action": action, "matched_incident": md, "reasoning": reasoning}
|
||||||
|
|
||||||
|
|
||||||
|
async def _run_consensus(preview, llm_decision):
|
||||||
|
tiebreak_result = {
|
||||||
|
"action": "new", "matched_incident": None, "incident_type": "other",
|
||||||
|
"corr_debug": {}, "reasoning": "tb",
|
||||||
|
}
|
||||||
|
with patch("app.internal.incident_correlator.preview_correlation",
|
||||||
|
new=AsyncMock(return_value=preview)), \
|
||||||
|
patch("app.internal.incident_correlator.apply_correlation",
|
||||||
|
new=AsyncMock(return_value="incident-x")) as m_apply, \
|
||||||
|
patch("app.internal.llm_correlator.decide",
|
||||||
|
new=AsyncMock(return_value=llm_decision)), \
|
||||||
|
patch("app.internal.llm_correlator.tiebreak",
|
||||||
|
new=AsyncMock(return_value=tiebreak_result)) as m_tiebreak:
|
||||||
|
await upload._correlate_with_consensus(
|
||||||
|
call_id="call-1", node_id="n1", system_id="sys-1",
|
||||||
|
talkgroup_id=9048, talkgroup_name="Dispatch", tags=[],
|
||||||
|
incident_type=None, location=None, location_coords=None,
|
||||||
|
)
|
||||||
|
return m_apply, m_tiebreak
|
||||||
|
|
||||||
|
|
||||||
|
async def test_substanceless_no_recent_same_tg_incident_gates_without_tiebreak():
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}), _llm("orphan", "unit check-in, not an incident"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_not_called()
|
||||||
|
m_apply.assert_called_once()
|
||||||
|
gated = m_apply.call_args[0][0]["decision"]
|
||||||
|
assert gated["action"] == "orphan"
|
||||||
|
dbg = gated["corr_debug"]
|
||||||
|
assert dbg["corr_consensus"] == "llm_orphan_gate"
|
||||||
|
assert dbg["corr_consensus"] != "tiebreak"
|
||||||
|
assert dbg["corr_rules_action"] == "new"
|
||||||
|
assert dbg["corr_llm_action"] == "orphan"
|
||||||
|
assert dbg["corr_llm_reasoning"] == "unit check-in, not an incident"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("severity", ["moderate", "major"])
|
||||||
|
async def test_moderate_or_major_severity_is_not_gated(severity):
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}, ctx={"call_severity": severity}), _llm("orphan"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_routine_severity_alone_still_gates():
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}, ctx={"call_severity": "routine"}), _llm("orphan"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_not_called()
|
||||||
|
assert m_apply.call_args[0][0]["decision"]["action"] == "orphan"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_call_with_coords_is_not_gated():
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}, ctx={"coords": {"lat": 41.15, "lng": -73.86}}),
|
||||||
|
_llm("orphan"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_call_with_tags_is_not_gated():
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}, ctx={"tags": ["structure-fire"]}), _llm("orphan"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_call_with_vehicles_is_not_gated():
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}, ctx={"call_vehicles": ["red sedan"]}), _llm("orphan"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_call_with_resolved_incident_type_is_not_gated():
|
||||||
|
# The creation gate skips has_event_substance when a type resolved, so a
|
||||||
|
# typed call (fire/medical/…) opens an incident on substance the gate does
|
||||||
|
# not re-check — it must keep the tiebreak, not be dropped.
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}, ctx={"incident_type": "fire"}), _llm("orphan"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_reassignment_call_is_not_gated():
|
||||||
|
# reassignment=True is dispatch pulling a unit onto a NEW job (units are
|
||||||
|
# blanked for exactly that reason) — the strongest new-incident signal.
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}, ctx={"reassignment": True}), _llm("orphan"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_recent_incident_on_same_talkgroup_is_not_gated():
|
||||||
|
ctx = {
|
||||||
|
"system_id": "sys-1",
|
||||||
|
"talkgroup_id": 9048,
|
||||||
|
"recent": [{
|
||||||
|
"incident_id": "inc-live",
|
||||||
|
"system_ids": ["sys-1"],
|
||||||
|
"talkgroup_ids": ["9048"],
|
||||||
|
}],
|
||||||
|
}
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}, ctx=ctx), _llm("orphan"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_recent_incident_on_a_different_talkgroup_still_gates():
|
||||||
|
ctx = {
|
||||||
|
"system_id": "sys-1",
|
||||||
|
"talkgroup_id": 9048,
|
||||||
|
"recent": [{
|
||||||
|
"incident_id": "inc-other",
|
||||||
|
"system_ids": ["sys-1"],
|
||||||
|
"talkgroup_ids": ["1200"],
|
||||||
|
}],
|
||||||
|
}
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(
|
||||||
|
_preview("new", {}, ctx=ctx), _llm("orphan"),
|
||||||
|
)
|
||||||
|
m_tiebreak.assert_not_called()
|
||||||
|
assert m_apply.call_args[0][0]["decision"]["action"] == "orphan"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_llm_link_vs_rules_new_still_escalates():
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(_preview("new", {}), _llm("link", "same job"))
|
||||||
|
m_tiebreak.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_llm_orphan_vs_rules_link_still_escalates():
|
||||||
|
# Not the gate condition (gate needs rules=="new"); must fall through.
|
||||||
|
m_apply, m_tiebreak = await _run_consensus(_preview("link", {}), _llm("orphan"))
|
||||||
|
m_tiebreak.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
def test_has_event_substance_predicate():
|
||||||
|
assert has_event_substance({"coords": {"lat": 1, "lng": 2}})
|
||||||
|
assert has_event_substance({"tags": ["fire"]})
|
||||||
|
assert has_event_substance({"call_vehicles": ["sedan"]})
|
||||||
|
assert not has_event_substance({})
|
||||||
|
assert not has_event_substance({"coords": None, "tags": [], "call_vehicles": []})
|
||||||
|
# units and location are NOT substance — nearly every transmission has them.
|
||||||
|
assert not has_event_substance({"call_units": ["7-Adam"], "location": "Main St"})
|
||||||
|
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
# Fix 2 — tighten corr_path=location
|
||||||
|
# ─────────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
CALL_COORDS = {"lat": 41.150000, "lng": -73.860000}
|
||||||
|
# ~0.39 km north of the call — inside location_proximity_km (0.5) but well
|
||||||
|
# outside the tight bar (_LOCATION_TIGHT_PROXIMITY_KM, 0.2).
|
||||||
|
FAR_INC_COORDS = {"lat": 41.153500, "lng": -73.860000}
|
||||||
|
# ~0.13 km north of the call — inside the tight bar.
|
||||||
|
NEAR_INC_COORDS = {"lat": 41.151200, "lng": -73.860000}
|
||||||
|
# ~0.28 km north — inside the 0.5 radius, outside the 0.2 tight bar; used as a
|
||||||
|
# second candidate that must lose the nearest-wins sort to NEAR_INC_COORDS.
|
||||||
|
MID_INC_COORDS = {"lat": 41.152500, "lng": -73.860000}
|
||||||
|
|
||||||
|
|
||||||
|
def _inc(incident_id, coords, units):
|
||||||
|
return {
|
||||||
|
"incident_id": incident_id,
|
||||||
|
"system_ids": ["sys-1"],
|
||||||
|
"talkgroup_ids": ["100"], # different TGID → fast path is a no-op
|
||||||
|
"location_coords": coords,
|
||||||
|
"units": units,
|
||||||
|
"tags": [],
|
||||||
|
"type": "police",
|
||||||
|
"updated_at": (NOW - timedelta(minutes=6)).isoformat(),
|
||||||
|
"started_at": (NOW - timedelta(minutes=20)).isoformat(),
|
||||||
|
"status": "active",
|
||||||
|
"call_ids": ["c0"],
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _loc_ctx(*, incidents, call_units):
|
||||||
|
return {
|
||||||
|
"call_id": "call-loc",
|
||||||
|
"all_active": list(incidents),
|
||||||
|
"recent": list(incidents),
|
||||||
|
"call_doc": {},
|
||||||
|
"call_embedding": None,
|
||||||
|
"call_units": call_units,
|
||||||
|
"call_vehicles": [],
|
||||||
|
"call_cleared": [],
|
||||||
|
"call_severity": "routine",
|
||||||
|
"coords": CALL_COORDS,
|
||||||
|
"is_thin_call": False,
|
||||||
|
"now": NOW,
|
||||||
|
"system_id": "sys-1",
|
||||||
|
"talkgroup_id": 999, # not in inc.talkgroup_ids
|
||||||
|
"talkgroup_name": "Tactical",
|
||||||
|
"tags": [],
|
||||||
|
"incident_type": "police",
|
||||||
|
"location": "Main St",
|
||||||
|
"location_coords": CALL_COORDS,
|
||||||
|
"reassignment": True, # suppress the unit-continuity path
|
||||||
|
"create_if_new": True,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_location_path_in_radius_but_no_unit_overlap_no_tight_proximity_does_not_link(caplog):
|
||||||
|
ctx = _loc_ctx(
|
||||||
|
incidents=[_inc("inc-loc", FAR_INC_COORDS, ["7-Adam"])],
|
||||||
|
call_units=["3-Boy"],
|
||||||
|
)
|
||||||
|
with caplog.at_level("INFO", logger="drb-c2-core"):
|
||||||
|
decision = _run_decision(ctx)
|
||||||
|
# Reaches, and is rejected by, the new guard (not an earlier path).
|
||||||
|
assert "location-path skipped" in caplog.text
|
||||||
|
assert decision["action"] != "link"
|
||||||
|
assert (decision.get("corr_debug") or {}).get("corr_path") != "location"
|
||||||
|
|
||||||
|
|
||||||
|
def test_location_path_links_on_unit_overlap_with_distinct_fit_signal():
|
||||||
|
ctx = _loc_ctx(
|
||||||
|
incidents=[_inc("inc-loc", FAR_INC_COORDS, ["5-Adam"])],
|
||||||
|
call_units=["5-Adam"],
|
||||||
|
)
|
||||||
|
decision = _run_decision(ctx)
|
||||||
|
assert decision["action"] == "link"
|
||||||
|
assert decision["corr_debug"]["corr_path"] == "location"
|
||||||
|
# NOT "unit_overlap" — that value belongs to the fast path's histogram bucket.
|
||||||
|
assert decision["corr_debug"]["corr_fit_signal"] == "location_unit_overlap"
|
||||||
|
|
||||||
|
|
||||||
|
def test_location_path_links_on_tight_proximity_without_unit_overlap():
|
||||||
|
ctx = _loc_ctx(
|
||||||
|
incidents=[_inc("inc-loc", NEAR_INC_COORDS, ["7-Adam"])],
|
||||||
|
call_units=["3-Boy"],
|
||||||
|
)
|
||||||
|
decision = _run_decision(ctx)
|
||||||
|
assert decision["action"] == "link"
|
||||||
|
assert decision["corr_debug"]["corr_path"] == "location"
|
||||||
|
assert decision["corr_debug"]["corr_fit_signal"] == "location_proximity"
|
||||||
|
|
||||||
|
|
||||||
|
def test_location_path_picks_nearest_in_radius_candidate():
|
||||||
|
# `recent` order puts the farther tight-proximity incident first; the guard
|
||||||
|
# must still select the nearest one.
|
||||||
|
ctx = _loc_ctx(
|
||||||
|
incidents=[
|
||||||
|
_inc("inc-mid", MID_INC_COORDS, ["3-Boy"]), # ~0.28 km, tight-fail
|
||||||
|
_inc("inc-near", NEAR_INC_COORDS, ["3-Boy"]), # ~0.13 km, tight-pass
|
||||||
|
],
|
||||||
|
call_units=["3-Boy"],
|
||||||
|
)
|
||||||
|
decision = _run_decision(ctx)
|
||||||
|
assert decision["action"] == "link"
|
||||||
|
assert decision["matched_incident"]["incident_id"] == "inc-near"
|
||||||
|
assert decision["corr_debug"]["corr_path"] == "location"
|
||||||
Reference in New Issue
Block a user