Logan CusanoandClaude Opus 5.5
badfe28823
Owner: traffic stops should show on the portal — otherwise they are only
visible in the archive. In the 09-22 replay (server-26#170) every Ch 1
stop ("45 Adam on a stop, Eastbound Central Express", "CM2 on the stop,
southbound") came back from extraction untyped, untagged and routine —
read as status traffic after #138 — so the creation gate never opened one.
- prompt: a unit reporting its own activity (on a stop, out with a vehicle
or pedestrian) is a real event: police, tagged, at least minor; the plate
lookups for it belong to it.
- deterministic backstop after extraction: stop / "put me out with"
phrasing adds a "traffic-stop" / "self-initiated" tag (the substance the
creation gate counts), police type if none, minor if routine. Negated
phrasing ("not pull the car over") is left alone; nothing is downgraded.
c2-core: 469 pass.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
118 lines
6.1 KiB
Python
118 lines
6.1 KiB
Python
"""
|
|
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, 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")
|