server-26/drb-c2-core/tests/test_consensus_gate.py

"""
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,
        "talkgroup_name": "Dispatch",
        "now": NOW,
        "recent": [{
            "incident_id": "inc-live",
            "system_ids": ["sys-1"],
            "talkgroup_ids": ["9048"],
            "updated_at": (NOW - timedelta(minutes=1)).isoformat(),
        }],
    }
    m_apply, m_tiebreak = await _run_consensus(
        _preview("new", {}, ctx=ctx), _llm("orphan"),
    )
    m_tiebreak.assert_called_once()


# server-26#115 window #3 (CORRELATION_REVIEW_0912.md): the escape hatch used
# to treat ANY same-talkgroup incident inside the 2h correlation_window_hours
# as "recent", which on a busy dispatch channel (3-13 incidents/2h) was
# satisfied almost unconditionally — the gate fired 0/24 times against its own
# target shape. It now only counts an incident as recent within
# settings.tg_dispatch_thin_idle_minutes (5 min), applied uniformly regardless
# of the talkgroup's name (owner correction, 2026-09-13 — see
# test_channel_name_does_not_affect_the_window below for why the dichotomy
# this originally had with incident_correlator's fast/thin idle selection was
# removed here).

async def test_recent_same_tg_incident_inside_new_short_window_still_escapes_gate():
    ctx = {
        "system_id": "sys-1",
        "talkgroup_id": 9048,
        "talkgroup_name": "Dispatch",
        "now": NOW,
        "recent": [{
            "incident_id": "inc-live",
            "system_ids": ["sys-1"],
            "talkgroup_ids": ["9048"],
            # 3 min ago — inside tg_dispatch_thin_idle_minutes (5).
            "updated_at": (NOW - timedelta(minutes=3)).isoformat(),
        }],
    }
    m_apply, m_tiebreak = await _run_consensus(
        _preview("new", {}, ctx=ctx), _llm("orphan"),
    )
    m_tiebreak.assert_called_once()


async def test_recent_same_tg_incident_older_than_short_window_now_gates():
    # Regression test for the fix: 8 minutes is past the 5-minute bound but
    # still inside the OLD 2-hour correlation_window_hours lookback. Before
    # the fix this escaped the gate on any channel; after the fix it gates.
    ctx = {
        "system_id": "sys-1",
        "talkgroup_id": 9048,
        "talkgroup_name": "Dispatch",
        "now": NOW,
        "recent": [{
            "incident_id": "inc-stale",
            "system_ids": ["sys-1"],
            "talkgroup_ids": ["9048"],
            "updated_at": (NOW - timedelta(minutes=8)).isoformat(),
        }],
    }
    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_channel_name_does_not_affect_the_window():
    # Owner correction, 2026-09-13 (direct scanning experience): a talkgroup
    # named "tac"/"tactical" only sees materially different traffic during a
    # real incident, and that's rare -- the bulk of traffic on any monitored
    # channel, including high-risk stops and pursuits, runs on the main
    # channel regardless of what it's named. An earlier version of this used
    # a longer 15-minute window on anything not literally named "dispatch"/
    # "patched"/"primary" (mirroring incident_correlator's fast/thin idle
    # selection); that meant a busy single-channel department not literally
    # named "dispatch" silently got the more permissive window and could
    # reproduce #115's original bug. Same 8-minute age as the dispatch-named
    # test above, but on a channel named "Tac 3" -- must gate identically,
    # not escape into a longer window just because of the name.
    ctx = {
        "system_id": "sys-1",
        "talkgroup_id": 383,
        "talkgroup_name": "Tac 3",
        "now": NOW,
        "recent": [{
            "incident_id": "inc-tac",
            "system_ids": ["sys-1"],
            "talkgroup_ids": ["383"],
            "updated_at": (NOW - timedelta(minutes=8)).isoformat(),
        }],
    }
    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_gate_veto_reason_is_recorded_on_the_escalation_path():
    # server-26#115: a live measurement window must be able to see *why* an
    # llm=orphan/rules=new call escaped the gate without guessing from the raw
    # dump (which produced a wrong "confirmed explanation" for 2 window-#3
    # misses the first time). corr_gate_veto names the surviving condition.
    ctx = {"call_severity": "major"}
    m_apply, m_tiebreak = await _run_consensus(
        _preview("new", {}, ctx=ctx), _llm("orphan"),
    )
    m_tiebreak.assert_called_once()
    final = m_apply.call_args[0][0]["decision"]
    assert final["corr_debug"]["corr_gate_veto"] == "severity"


async def test_gate_veto_reason_is_absent_when_the_disagreement_is_not_orphan_vs_new():
    # corr_gate_veto is only meaningful for the llm=orphan/rules=new shape the
    # gate targets — it must not appear (or be misleadingly None-vs-absent) on
    # an unrelated disagreement shape.
    m_apply, m_tiebreak = await _run_consensus(
        _preview("link", {}), _llm("orphan"),
    )
    m_tiebreak.assert_called_once()
    final = m_apply.call_args[0][0]["decision"]
    assert "corr_gate_veto" not in final["corr_debug"]


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"