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` 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 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.
"""
from datetime import datetime, timedelta, timezone
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

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):
    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": {"call_id": "call-1"},
    }


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_llm_orphan_vs_rules_new_no_signal_gates_to_orphan_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("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"))
    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):
    m_apply, m_tiebreak = await _run_consensus(
        _preview("new", corr_debug), _llm("orphan"),
    )
    m_tiebreak.assert_called_once()


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({})


# ─────────────────────────────────────────────────────────────────────────────
# 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}


def _loc_ctx(*, inc_coords, inc_units, call_units):
    inc = {
        "incident_id": "inc-loc",
        "system_ids": ["sys-1"],
        "talkgroup_ids": ["100"],           # different TGID → fast path is a no-op
        "location_coords": inc_coords,
        "units": inc_units,
        "tags": [],
        "type": "police",
        "updated_at": (NOW - timedelta(minutes=6)).isoformat(),
        "started_at": (NOW - timedelta(minutes=20)).isoformat(),
        "status": "active",
        "call_ids": ["c0"],
    }
    return {
        "call_id": "call-loc",
        "all_active": [inc],
        "recent": [inc],
        "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_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)
    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"])
    decision = _run_decision(ctx)
    assert decision["action"] == "link"
    assert decision["corr_debug"]["corr_path"] == "location"
    assert decision["corr_debug"]["corr_fit_signal"] == "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"])
    decision = _run_decision(ctx)
    assert decision["action"] == "link"
    assert decision["corr_debug"]["corr_path"] == "location"
    assert decision["corr_debug"]["corr_fit_signal"] == "location_proximity"