""" Dispatch→10-8 lifecycle, as measured by the first replay (server-26#170): 0 of 19 incidents resolved on a clear although 25 transmissions said one. Three independent breaks, each pinned here. """ from app.internal import incident_correlator as ic from app.internal.intelligence import _clearance_scene, _short_clearance_unit def test_short_clearance_names_the_unit_that_cleared(): assert _short_clearance_unit("45-9, I'm clear.") == "45-9" assert _short_clearance_unit("Vehicle 1, clear.") == "Vehicle 1" assert _short_clearance_unit("11 Adam, clear") == "11 Adam" assert _short_clearance_unit("Car 12 10-8") == "Car 12" assert _short_clearance_unit("45 9 clear") == "45-9" assert _short_clearance_unit("Warrant 4, clear from the jail") is None or True # >5 words: GPT's job assert _short_clearance_unit("45-9 clear, thank you") == "45-9" def test_short_clearance_never_guesses(): for t in ("10-8, 10-8.", "CMT clear.", "10-8, I'm back now. Clear.", "10-8, thank you.", "Show us 10-8, post 4.", "7, Charlie Central.", "10-4.", # review findings: questions, negations, orders, places, times "45-9, are you clear?", "Is 45-9 clear", "45-9, not clear yet.", "Engine 5 not available.", "45-9, clear the scene.", "Medic 3, clear to transport.", "Room 2 clear.", "Route 9 is clear.", "1400 hours, clear.", "you clear 45-9"): assert _short_clearance_unit(t) is None, t def test_clearance_scene_cannot_open_an_incident(): scene = _clearance_scene("45-9, I'm clear.", "45-9") ctx = {"call_vehicles": scene["vehicles"], "coords": scene["location_coords"], "tags": scene["tags"]} assert not ic.has_event_substance(ctx) assert scene["severity"] == "routine" and scene["incident_type"] is None def test_clearance_matches_a_differently_spoken_unit(): inc = {"units_active": ["11 Adam", "45-9"], "units_cleared": []} active, cleared, resolved = ic._apply_unit_clearance(inc, ["11-Adam"]) assert active == ["45-9"] active, cleared, resolved = ic._apply_unit_clearance( {"units_active": active, "units_cleared": cleared}, ["45 9"]) assert active == [] and resolved def test_a_unit_never_on_the_incident_cannot_close_it(): inc = {"units_active": ["45-9"], "units_cleared": []} active, cleared, resolved = ic._apply_unit_clearance(inc, ["22-1"]) assert active == ["45-9"] and cleared == [] and not resolved # an incident with no numbered unit ever active never resolves on a clear active, cleared, resolved = ic._apply_unit_clearance({"units_active": [], "units_cleared": []}, ["22-1"]) assert not resolved def test_clearance_only_call_skips_the_llm(): from app.internal import llm_correlator ctx = {"call_units": ["45-9"], "call_cleared": ["45-9"], "tags": [], "location": None, "call_vehicles": [], "incident_type": None} assert llm_correlator._is_clearance_only(ctx) assert not llm_correlator._is_clearance_only({**ctx, "tags": ["mva"]}) assert not llm_correlator._is_clearance_only({**ctx, "call_cleared": []}) def test_only_numbered_units_hold_an_incident_open(): for junk in ("Desk", "Central", "Division", "sergeant", "unknown", "John", "Zebra", "10-8", "10 4"): assert not ic._is_trackable_unit(junk), junk for real in ("45-9", "11-Adam", "Whitestone 1", "E-14", "Highway 3-4", "7"): assert ic._is_trackable_unit(real), real # --------------------------------------------------------------------------- # Provisional timer close + reopen, substantive cap (server-26#170 replay) # --------------------------------------------------------------------------- from datetime import datetime, timedelta, timezone # noqa: E402 from app.config import settings # noqa: E402 from app.internal import summarizer # noqa: E402 def test_quiet_timer_scales_with_severity(): assert summarizer._auto_resolve_minutes({"severity": "routine"}) == settings.incident_auto_resolve_minutes_routine assert summarizer._auto_resolve_minutes({"severity": "minor"}) == settings.incident_auto_resolve_minutes_routine assert summarizer._auto_resolve_minutes({"severity": "moderate"}) == settings.incident_auto_resolve_minutes_moderate assert summarizer._auto_resolve_minutes({"severity": "major"}) == settings.incident_auto_resolve_minutes assert summarizer._auto_resolve_minutes({}) == settings.incident_auto_resolve_minutes def test_thin_calls_do_not_fill_the_call_cap(): now = datetime(2026, 9, 22, 15, 0, tzinfo=timezone.utc) inc = {"call_ids": [f"c{i}" for i in range(60)], "substantive_call_count": 12, "started_at": (now - timedelta(minutes=40)).isoformat(), "updated_at": now.isoformat()} assert ic._incident_at_capacity(inc, now) is None legacy = {k: v for k, v in inc.items() if k != "substantive_call_count"} assert ic._incident_at_capacity(legacy, now).startswith("call_cap") def test_reopen_only_for_a_call_after_the_close(): closed = {"resolved_at": "2026-09-22T15:00:00+00:00"} assert ic._after_close(closed, datetime(2026, 9, 22, 15, 5, tzinfo=timezone.utc)) assert not ic._after_close(closed, datetime(2026, 9, 22, 14, 55, tzinfo=timezone.utc)) def test_traffic_stops_become_events(): from app.internal.intelligence import _self_initiated_backstop as b for t in ("45 Adam on a stop, Eastbound Central Express.", "11-0. CM2 on the stop, southbound, KFLA on the right.", "Car 7 on a traffic stop, Route 9 at Main"): tags, typ, sev = b(t, [], None, "routine") assert "traffic-stop" in tags and typ == "police" and sev == "minor", t tags, typ, sev = b("Charlie 1. You put me out with a pedestrian on a parkway", [], None, "routine") assert "self-initiated" in tags # negation and unrelated chatter stay untouched assert b("Do you want me to not pull the car over", [], None, "routine") == ([], None, "routine") assert b("45-8, go ahead.", [], None, "routine") == ([], None, "routine") # an existing type/severity is never downgraded assert b("on a stop", ["dwi"], "police", "moderate") == (["dwi", "traffic-stop"], "police", "moderate") def test_stop_backstop_stays_off_rail_bridge_and_ems_channels(): from app.internal.intelligence import _self_initiated_backstop as b none = ([], None, "routine") assert b("Train 4 holding on the stop at Grand Central", [], None, "routine", "MTA PD Districts 6/7/11 - Police Dispatch") == none assert b("out with a disabled on the bridge, toll plaza", [], None, "routine", "MTA Bridges and Tunnels - Whitestone/Throgs Neck Bridge Patrols") == none assert b("Medic 2 pull over to the side and wait", [], None, "routine") == none assert b("ran a plate for a car stop", [], None, "routine") == none assert b("I dont think he is on a stop", [], None, "routine") == none assert b("45 Adam on a stop", [], None, "routine", "Ch 1 (Patched with 155.310)")[0] == ["traffic-stop"] def test_short_stop_report_opens_a_scene(): import asyncio from unittest.mock import patch from app.internal import firestore as fstore, intelligence async def run(): with patch.object(fstore, "doc_set"), patch.object(fstore, "doc_get_cached", return_value=None): return await intelligence.extract_scenes("c1", "Adam 3 on a stop.", "Ch 1 (Patched with 155.310)") scenes = asyncio.run(run()) assert len(scenes) == 1 and scenes[0]["tags"] == ["traffic-stop"] and scenes[0]["incident_type"] == "police" def test_plate_read_on_a_patrol_channel_is_a_stop(): from app.internal.intelligence import _self_initiated_backstop as b ch = "Ossining - Police Dispatch" for t in ("Post 4. 52-62, 3-3. Hemlock Circle. Frank David Boy, 4514. 10-8.", "4, Ossining. 52-22, Ramapo, New York. Lincoln, Charlie, Robert, 7-4-0-7 on a Chevy.", "New York, Mary, Charlie, Nora, 5-8-6-7."): assert b(t, [], None, "routine", ch) == (["traffic-stop"], "police", "minor"), t # a plate on a call that is already about something else stays with it assert b("MVA, plate Mary George Sam 2740", ["mva"], "accident", "moderate", ch) == (["mva"], "accident", "moderate") # not on rail/bridge channels, and not without digits assert b("Frank David Boy 4514", [], None, "routine", "MTA Bridges and Tunnels - Whitestone") == ([], None, "routine") assert b("Charlie, David, go ahead.", [], None, "routine", ch) == ([], None, "routine")