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

correlator: fix consensus orphan-gate to test call substance, not empty corr_debug (#115)

Browse Source 
The gate added in ca1d8fb checked rules_decision["corr_debug"] for a positive
signal, but that dict is empty at preview time for action=="new" (corr_path is
written at apply time). The check was always False, so the gate fired on real
events — replayed against corr_dump_9-7_pm.json it dropped ~36 linked calls
including a major "extinguishing fire", a moderate fire-alarm, geocoded calls
and pursuit updates.

Gate now runs against ctx (fully populated at preview time). It fires ONLY when
the call is substanceless: routine severity, no vehicle/geocode/tag, and no
incident already running on the same talkgroup. Any of those escalates to the
tiebreak instead. The substance predicate (has_event_substance) is factored out
of incident_correlator's creation gate and shared, so the two cannot diverge.

recorrelation_sweep: a call the gate parked gets a longer link-only retry budget
(10 vs 3) — the gate fires before any incident for the job exists, so the
substantive call that justifies linking can land after the standard ~6 min.
Still create_if_new=False.

incident_correlator location path: evaluate every in-radius candidate and link
the nearest that carries corroboration, instead of the first in an unsorted
`recent`. A unit-overlap location link is now tagged "location_unit_overlap" so
it stops merging into the fast path's bucket in the admin fit-signal histogram.

tests/test_consensus_gate.py: replaced the corr_debug-signal cases with ctx
substance cases (severity, coords, tags, vehicles, same-tg incident); added a
nearest-wins location test; the two location guard tests now assert they reach
the new guard. Full drb-c2-core suite 322 -> 325.

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-07 23:50:41 -04:00
co-authored by Claude Sonnet 5
parent ca1d8fbdae
commit dd426572fc
4 changed files with 290 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
drb-c2-core/tests/test_consensus_gate.py
+151 -50
View File
@@ -1,18 +1,27 @@
"""
server-26#115 — two consensus-quality fixes.
Fix 1 (routers/upload.py): when the cheap LLM says `orphan` and the rules engine
says `new` with NO positive event signal, 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). A genuine event the LLM
misreads as orphan still escalates, because the rules result then carries a real
signal (unit overlap, location proximity, unit-continuity / disambig).
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. 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.
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
@@ -20,8 +29,7 @@ from unittest.mock import AsyncMock, patch
import pytest
from app.routers import upload
from app.routers.upload import _rules_has_positive_event_signal
from app.internal.incident_correlator import _run_decision
from app.internal.incident_correlator import _run_decision, has_event_substance
NOW = datetime(2026, 9, 7, 21, 30, 0, tzinfo=timezone.utc)
@@ -30,7 +38,10 @@ 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):
def _preview(action, corr_debug=None, ctx=None):
base_ctx = {"call_id": "call-1"}
if ctx:
base_ctx.update(ctx)
return {
"decision": {
"action": action,
@@ -38,7 +49,7 @@ def _preview(action, corr_debug=None):
"incident_type": "other" if action == "new" else None,
"corr_debug": {} if corr_debug is None else dict(corr_debug),
},
"ctx": {"call_id": "call-1"},
"ctx": base_ctx,
}
@@ -68,7 +79,7 @@ async def _run_consensus(preview, llm_decision):
return m_apply, m_tiebreak
async def test_llm_orphan_vs_rules_new_no_signal_gates_to_orphan_without_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"),
)
@@ -84,41 +95,96 @@ async def test_llm_orphan_vs_rules_new_no_signal_gates_to_orphan_without_tiebrea
assert dbg["corr_llm_reasoning"] == "unit check-in, not an incident"
@pytest.mark.parametrize("fit_signal", [None, "none", "thin_recency"])
async def test_gate_fires_for_every_non_positive_fit_signal(fit_signal):
dbg = {} if fit_signal is None else {"corr_fit_signal": fit_signal}
m_apply, m_tiebreak = await _run_consensus(_preview("new", dbg), _llm("orphan"))
@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"
@pytest.mark.parametrize("corr_debug", [
{"corr_fit_signal": "unit_overlap"},
{"corr_fit_signal": "location_proximity"},
{"corr_path": "unit-continuity"},
{"corr_path": "fast/disambig"},
])
async def test_positive_rules_signal_still_escalates_to_tiebreak(corr_debug):
async def test_call_with_coords_is_not_gated():
m_apply, m_tiebreak = await _run_consensus(
_preview("new", corr_debug), _llm("orphan"),
_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_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()
def test_rules_has_positive_event_signal_predicate():
assert _rules_has_positive_event_signal({"corr_debug": {"corr_fit_signal": "unit_overlap"}})
assert _rules_has_positive_event_signal({"corr_debug": {"corr_fit_signal": "location_proximity"}})
assert _rules_has_positive_event_signal({"corr_debug": {"corr_path": "unit-continuity"}})
assert _rules_has_positive_event_signal({"corr_debug": {"corr_path": "location"}})
assert not _rules_has_positive_event_signal({"corr_debug": {}})
assert not _rules_has_positive_event_signal({"corr_debug": {"corr_path": "new"}})
assert not _rules_has_positive_event_signal({"corr_debug": {"corr_fit_signal": "thin_recency"}})
assert not _rules_has_positive_event_signal({})
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"})
# ─────────────────────────────────────────────────────────────────────────────
@@ -131,15 +197,18 @@ CALL_COORDS = {"lat": 41.150000, "lng": -73.860000}
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 _loc_ctx(*, inc_coords, inc_units, call_units):
inc = {
"incident_id": "inc-loc",
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": inc_coords,
"units": inc_units,
"location_coords": coords,
"units": units,
"tags": [],
"type": "police",
"updated_at": (NOW - timedelta(minutes=6)).isoformat(),
@@ -147,10 +216,13 @@ def _loc_ctx(*, inc_coords, inc_units, call_units):
"status": "active",
"call_ids": ["c0"],
}
def _loc_ctx(*, incidents, call_units):
return {
"call_id": "call-loc",
"all_active": [inc],
"recent": [inc],
"all_active": list(incidents),
"recent": list(incidents),
"call_doc": {},
"call_embedding": None,
"call_units": call_units,
@@ -172,24 +244,53 @@ def _loc_ctx(*, inc_coords, inc_units, call_units):
}
def test_location_path_shared_area_no_unit_overlap_no_proximity_does_not_link():
ctx = _loc_ctx(inc_coords=FAR_INC_COORDS, inc_units=["7-Adam"], call_units=["3-Boy"])
decision = _run_decision(ctx)
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():
ctx = _loc_ctx(inc_coords=FAR_INC_COORDS, inc_units=["5-Adam"], call_units=["5-Adam"])
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"
assert decision["corr_debug"]["corr_fit_signal"] == "unit_overlap"
# 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(inc_coords=NEAR_INC_COORDS, inc_units=["7-Adam"], call_units=["3-Boy"])
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"