Let severity, not incident_type, decide what becomes an incident
The 2026-08-16 correlation dump showed two failures that looked unrelated and were the same bug. TG 9048 held one incident of 28 calls spanning 49 minutes -- a prisoner transport, a drone retrieval, a records lookup and a canvass, glued together -- while 32 other calls on that same channel stayed permanently orphaned. Creating an incident required a concrete incident_type. Nothing on a transit police channel produced one: the extraction prompt said to prefer "other" when uncertain, extraction then collapsed "other" to None, and the tag-based fallback had no tags to work with because administrative traffic carries none. So the channel could never open a SECOND incident. Every later call funnelled into whichever incident happened to exist first, and every call too substantial for the thin path had nowhere to go at all. The two symptoms were the same missing value seen from opposite ends. Severity now decides incident-worthiness. It is a better fit for the question being asked -- "is this a real event?" -- than a service label ever was, and unlike incident_type it is always present. The prompt defines four levels with no escape hatch (routine/minor/moderate/major, "unknown" is gone) and calls skipped for a too-short transcript are still recorded as routine, because downstream code reads a missing severity as "not processed yet" rather than "nothing happened". Anything above routine, or carrying any extracted content, opens an incident under the neutral "other" type. "other" is also kept as a real classification now -- rail operations and public works genuinely are not police, fire or EMS. Separately, thin calls no longer refresh updated_at; they write last_thin_at. updated_at drives every recency gate in the fast path, so each "10-4" was resetting the idle clock on whatever it attached to, keeping that incident inside the gate for as long as anyone kept acknowledging. An incident now ages from its last substantive call. This is what made the 49-minute incident possible even once buckets existed, so it is fixed independently rather than being left to the gate change. The re-correlation sweep also now honours skip_reason. /upload has always refused to correlate garbage and too-short transcripts, but the sweep did not apply the same filter, so those fragments came back minutes later through the thin path and attached to whatever was most recent -- a second, quieter route into the same over-merge. Adds tests/test_correlator_gate.py (15 cases), the first tests against incident_correlator.py in its 1,517-line history. tests/conftest.py stubs firebase-admin only when it is genuinely absent, so the container's real SDK is never shadowed; this is what makes the correlator importable in the dev venv. That stub also made test_mqtt_handler and test_node_sweeper collectable for the first time, revealing 10 pre-existing failures in them -- test-vs-code drift, untouched here and catalogued in DEFERRED.md. No new environment variables, so CI deploys this without an ansible run.
This commit is contained in:
Logan Cusano
@@ -287,7 +287,7 @@ async def _build_context(
|
||||
call_units = units if units is not None else (call_doc.get("units") or [])
|
||||
call_vehicles = vehicles if vehicles is not None else (call_doc.get("vehicles") or [])
|
||||
call_cleared = cleared_units if cleared_units is not None else (call_doc.get("cleared_units") or [])
|
||||
call_severity = call_doc.get("severity") or "unknown"
|
||||
call_severity = call_doc.get("severity") or "routine"
|
||||
coords = location_coords or call_doc.get("location_coords")
|
||||
is_thin_call = not call_units and not call_vehicles and not coords
|
||||
|
||||
@@ -326,6 +326,7 @@ def _run_decision(ctx: dict) -> dict:
|
||||
call_embedding = ctx["call_embedding"]
|
||||
call_units = ctx["call_units"]
|
||||
call_vehicles = ctx["call_vehicles"]
|
||||
call_severity = ctx["call_severity"]
|
||||
coords = ctx["coords"]
|
||||
is_thin_call = ctx["is_thin_call"]
|
||||
now = ctx["now"]
|
||||
@@ -718,6 +719,27 @@ def _run_decision(ctx: dict) -> dict:
|
||||
f"Correlator: inferred incident_type={resolved_type!r} from tags {tags} for call {call_id}"
|
||||
)
|
||||
|
||||
# Severity, not type, decides whether a call is incident-worthy.
|
||||
#
|
||||
# Requiring a concrete incident_type here meant a channel whose traffic never
|
||||
# classifies — transit/rail administration, records lookups, prisoner
|
||||
# transports — could never open a SECOND incident. Every later call on that
|
||||
# talkgroup then funnelled into whichever incident happened to be created
|
||||
# first, producing hour-long incidents made of unrelated transmissions
|
||||
# (2026-08-16: TG 9048, 28 calls / 49 min) alongside 30+ permanent orphans.
|
||||
#
|
||||
# Anything the extractor judged a real event, or that carries any concrete
|
||||
# content, now opens an incident under the neutral "other" type. Only
|
||||
# content-free routine traffic is still left for the thin path to attach.
|
||||
if not resolved_type:
|
||||
has_substance = bool(call_units or call_vehicles or coords or location or tags)
|
||||
if call_severity in ("minor", "moderate", "major") or has_substance:
|
||||
resolved_type = "other"
|
||||
logger.info(
|
||||
f"Correlator: call {call_id} has no incident_type — opening an 'other' "
|
||||
f"incident (severity={call_severity}, substance={has_substance})"
|
||||
)
|
||||
|
||||
if not resolved_type:
|
||||
return {"action": "orphan", "matched_incident": None, "incident_type": None, "corr_debug": corr_debug}
|
||||
|
||||
@@ -772,11 +794,13 @@ async def _apply_decision(decision: dict, ctx: dict) -> Optional[str]:
|
||||
|
||||
if action == "link":
|
||||
matched_incident = decision["matched_incident"]
|
||||
# A thin call attaches for context but does not count as incident activity.
|
||||
thin_link = (decision.get("corr_debug") or {}).get("corr_path") == "fast/thin"
|
||||
await _update_incident(
|
||||
matched_incident, call_id, talkgroup_id, system_id, tags,
|
||||
location, location_coords, call_units, call_vehicles, call_embedding, now,
|
||||
talkgroup_name=talkgroup_name, incident_type=incident_type,
|
||||
cleared_units=call_cleared,
|
||||
cleared_units=call_cleared, refresh_activity=not thin_link,
|
||||
)
|
||||
return matched_incident["incident_id"]
|
||||
|
||||
@@ -1149,6 +1173,7 @@ async def _update_incident(
|
||||
talkgroup_name: Optional[str] = None,
|
||||
incident_type: Optional[str] = None,
|
||||
cleared_units: Optional[list[str]] = None,
|
||||
refresh_activity: bool = True,
|
||||
) -> None:
|
||||
incident_id = inc["incident_id"]
|
||||
|
||||
@@ -1200,10 +1225,20 @@ async def _update_incident(
|
||||
"units_active": units_active,
|
||||
"units_cleared": units_cleared,
|
||||
"location_mentions": location_mentions,
|
||||
"updated_at": now.isoformat(),
|
||||
"summary_stale": True,
|
||||
**embedding_updates,
|
||||
}
|
||||
|
||||
# `updated_at` drives every recency gate in the fast path, so a content-free
|
||||
# status call must NOT refresh it. When it did, each "10-4" reset the idle
|
||||
# clock on the incident it attached to, which kept that incident permanently
|
||||
# "recent" and made it absorb the entire channel for as long as anyone kept
|
||||
# acknowledging. The incident now ages from its last SUBSTANTIVE call, and
|
||||
# thin traffic rides along without extending its life.
|
||||
if refresh_activity:
|
||||
updates["updated_at"] = now.isoformat()
|
||||
else:
|
||||
updates["last_thin_at"] = now.isoformat()
|
||||
if best_location:
|
||||
updates["location"] = best_location
|
||||
if best_coords:
|
||||
|
||||
@@ -42,7 +42,7 @@ Response format — a JSON object with a "scenes" array. Each scene:
|
||||
vehicles: list of vehicle descriptions mentioned
|
||||
units: list of unit IDs or officer numbers explicitly mentioned
|
||||
cleared_units: list of unit IDs that explicitly signal back-in-service or available in this recording
|
||||
severity: one of "minor" | "moderate" | "major" | "unknown"
|
||||
severity: one of "routine" | "minor" | "moderate" | "major"
|
||||
resolved: true if this scene explicitly signals incident closure, false otherwise
|
||||
reassignment: true if a unit is breaking from their current scene to respond to a completely different call — whether dispatch-initiated ("Baker, can you clear and respond to...", "Adam, break from that and go to...") OR unit-initiated ("Show me headed to the vehicle complaint", "Can you show me to that call", a unit going 10-8 and self-requesting a new assignment). False if the unit is reporting in on their current scene, giving a status update, or requesting information about their existing call.
|
||||
transcript_corrected: corrected text for this scene's transmissions only, or null
|
||||
@@ -52,7 +52,12 @@ Rules:
|
||||
- tags: describe WHAT happened, not WHERE. Specific, lowercase, hyphenated. Do not use location names, road names, talkgroup names, or place names as tags (wrong: "lower-macy's", "canvas-route-6", "route-202"; right: "suspect-search", "shoplifting", "vehicle-pursuit"). Do not repeat incident_type as a tag.
|
||||
- units: ONLY identifiers that appear verbatim in the transcript. Use speaker role inference to distinguish units being dispatched from units acknowledging — both should be included. Never infer or guess unit IDs not present in the text.
|
||||
- Do not invent details not present in the transcript.
|
||||
- incident_type: let the talkgroup channel be your primary signal. Use "fire" ONLY if the talkgroup is clearly a fire/rescue channel OR the transcript explicitly describes active fire, smoke, flames, or structure fire activation. Police or EMS referencing a fire scene → use "police" or "ems". When uncertain, prefer "other" over "fire".
|
||||
- incident_type: let the talkgroup channel be your primary signal. Use "fire" ONLY if the talkgroup is clearly a fire/rescue channel OR the transcript explicitly describes active fire, smoke, flames, or structure fire activation. Police or EMS referencing a fire scene → use "police" or "ems". When the channel is a police channel and nothing in the transcript contradicts it, return "police" — do NOT fall back to "other" merely because the transmission is administrative. Reserve "other" for traffic that genuinely belongs to no emergency service (rail operations, public works, utility coordination). Reserve "unknown" for transcripts too garbled to place at all.
|
||||
- severity: ALWAYS return one of the four values. Judge the underlying event, not how dramatic the words sound.
|
||||
"routine" — administrative/status traffic with no incident behind it: mileage and transport logging, radio checks, acknowledgements, shift changes, track block/power requests, records lookups.
|
||||
"minor" — a real but low-stakes call: lift assist, parking complaint, past-tense larceny report, noise complaint, welfare check.
|
||||
"moderate" — an active call needing a response now: MVA, alarm activation, disturbance in progress, medical call, suspicious person, road closure.
|
||||
"major" — life safety or major property loss: structure fire, vehicle pursuit, shots fired, entrapment, cardiac arrest, officer needing assistance.
|
||||
- ten_codes: interpret radio codes using the department reference provided below. Do not guess codes not listed.
|
||||
- resolved: true only when the scene explicitly signals "Code 4", "all clear", "10-42", "in custody", "patient transported", "fire out", "GOA", "negative contact", "scene clear".
|
||||
- cleared_units: only include units that explicitly stated their own back-in-service status in this recording (e.g. "Unit 7, 10-8", "Baker-1 available", "E-14 back in service", or the department ten-code for available/back-in-service listed above). Silence or absence of a unit is NOT clearance. A scene-wide Code 4 belongs in resolved=true, not here — cleared_units is for individual unit availability signals only.
|
||||
@@ -183,7 +188,13 @@ async def extract_scenes(
|
||||
f"({len(transcript.split())} words), skipping"
|
||||
)
|
||||
try:
|
||||
await fstore.doc_set("calls", call_id, {"skip_reason": "transcript_too_short"})
|
||||
# Severity is still recorded: a five-word acknowledgement is genuinely
|
||||
# routine traffic, and downstream code treats a missing severity as
|
||||
# "not yet processed" rather than "nothing happened".
|
||||
await fstore.doc_set("calls", call_id, {
|
||||
"skip_reason": "transcript_too_short",
|
||||
"severity": "routine",
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
return []
|
||||
@@ -213,13 +224,23 @@ async def extract_scenes(
|
||||
vehicles: list[str] = scene.get("vehicles") or []
|
||||
units: list[str] = scene.get("units") or []
|
||||
cleared_units: list[str] = scene.get("cleared_units") or []
|
||||
severity: str = scene.get("severity") or "unknown"
|
||||
# Every call carries a severity — it is the signal the correlator uses to
|
||||
# decide whether a call is incident-worthy at all, so it must never be
|
||||
# absent. "unknown" is a legacy value from before the prompt guaranteed
|
||||
# one of the four levels; normalise it to the bottom rung.
|
||||
severity: str = scene.get("severity") or "routine"
|
||||
if severity == "unknown":
|
||||
severity = "routine"
|
||||
resolved: bool = bool(scene.get("resolved", False))
|
||||
reassignment: bool = bool(scene.get("reassignment", False))
|
||||
transcript_corrected: Optional[str]= scene.get("transcript_corrected") or None
|
||||
segment_indices: Optional[list] = scene.get("segment_indices")
|
||||
|
||||
if incident_type in ("unknown", "other", ""):
|
||||
# "other" is a real classification (rail ops, public works, utility work)
|
||||
# and is kept. Collapsing it to None used to make the call untypeable,
|
||||
# and an untypeable call could never open an incident — see the creation
|
||||
# gate in incident_correlator._run_decision().
|
||||
if incident_type in ("unknown", ""):
|
||||
incident_type = None
|
||||
|
||||
# Geocode this scene's location.
|
||||
|
||||
@@ -55,6 +55,12 @@ async def _run_sweep_pass() -> None:
|
||||
if not c.get("incident_ids") and not c.get("incident_id")
|
||||
and not c.get("corr_path") # skip calls already exhausted
|
||||
and not c.get("duplicate_of") # another node's copy — never processed by design
|
||||
# /upload deliberately skips correlation for garbage and too-short
|
||||
# transcripts (routers/upload.py) because they carry no signal. The sweep
|
||||
# was not applying the same guard, so those fragments came back in through
|
||||
# the thin path minutes later and attached to whatever was most recent —
|
||||
# a second route into the over-merge the thin fix above addresses.
|
||||
and not c.get("skip_reason")
|
||||
and c.get("corr_sweep_count", 0) < MAX_SWEEP_ATTEMPTS
|
||||
]
|
||||
|
||||
|
||||
@@ -1,2 +1,61 @@
|
||||
# All C2 core settings have defaults — no env setup needed.
|
||||
# Add any shared fixtures here if required in the future.
|
||||
#
|
||||
# firebase-admin and google-cloud-firestore are runtime-only dependencies: they
|
||||
# are installed in the container but not in the local dev venv, and
|
||||
# app/internal/firestore.py calls _init_firebase() at import time. Without the
|
||||
# stubs below, importing ANY module that reaches Firestore fails at collection
|
||||
# time, which is why test_mqtt_handler and test_node_sweeper could not be run
|
||||
# outside the container.
|
||||
#
|
||||
# The stubs are installed only when the real packages are absent, so the
|
||||
# container's real SDK is never shadowed.
|
||||
import sys
|
||||
from types import ModuleType
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
try: # pragma: no cover - exercised only by which packages are installed
|
||||
import firebase_admin # noqa: F401
|
||||
except ModuleNotFoundError:
|
||||
_firebase = ModuleType("firebase_admin")
|
||||
# Falsy so _init_firebase() takes the initialize_app() branch rather than the
|
||||
# already-initialised branch, which is itself broken (see DEFERRED.md).
|
||||
_firebase._apps = {}
|
||||
_firebase.initialize_app = MagicMock()
|
||||
_firebase.credentials = MagicMock()
|
||||
_firebase.firestore = MagicMock()
|
||||
|
||||
_credentials = ModuleType("firebase_admin.credentials")
|
||||
_credentials.Certificate = MagicMock()
|
||||
_credentials.ApplicationDefault = MagicMock()
|
||||
|
||||
_fs = ModuleType("firebase_admin.firestore")
|
||||
_fs.client = MagicMock()
|
||||
# A distinct sentinel rather than a MagicMock: production code writes this
|
||||
# into dicts that tests compare against, and a MagicMock compares unequal
|
||||
# to itself across attribute accesses.
|
||||
_fs.SERVER_TIMESTAMP = "__SERVER_TIMESTAMP__"
|
||||
|
||||
_auth = ModuleType("firebase_admin.auth")
|
||||
_auth.verify_id_token = MagicMock()
|
||||
_auth.set_custom_user_claims = MagicMock()
|
||||
_auth.get_user_by_email = MagicMock()
|
||||
_auth.get_user = MagicMock()
|
||||
|
||||
_firebase.auth = _auth
|
||||
_firebase.credentials = _credentials
|
||||
_firebase.firestore = _fs
|
||||
|
||||
sys.modules["firebase_admin"] = _firebase
|
||||
sys.modules["firebase_admin.credentials"] = _credentials
|
||||
sys.modules["firebase_admin.firestore"] = _fs
|
||||
sys.modules["firebase_admin.auth"] = _auth
|
||||
|
||||
try: # pragma: no cover
|
||||
from google.cloud.firestore_v1.base_query import FieldFilter # noqa: F401
|
||||
except ModuleNotFoundError:
|
||||
for _name in (
|
||||
"google", "google.cloud", "google.cloud.firestore_v1",
|
||||
"google.cloud.firestore_v1.base_query",
|
||||
):
|
||||
sys.modules.setdefault(_name, ModuleType(_name))
|
||||
sys.modules["google.cloud.firestore_v1.base_query"].FieldFilter = MagicMock()
|
||||
|
||||
@@ -0,0 +1,148 @@
|
||||
"""
|
||||
Unit tests for the incident-creation gate and the thin-call activity rule.
|
||||
|
||||
Both behaviours come from the 2026-08-16 correlation dump, where TG 9048
|
||||
produced one 28-call / 49-minute incident alongside 32 permanent orphans:
|
||||
|
||||
* Requiring a concrete incident_type to create an incident meant a channel
|
||||
whose traffic never classifies could never open a second incident, so every
|
||||
later call funnelled into whichever incident existed first.
|
||||
* Thin ("10-4") calls refreshed updated_at, which kept that incident
|
||||
permanently inside the fast-path recency gate.
|
||||
|
||||
_run_decision is pure — it reads only the context dict — so these cases need no
|
||||
Firestore. _update_incident writes, so its test patches fstore.
|
||||
"""
|
||||
import pytest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import AsyncMock, patch
|
||||
from app.internal.incident_correlator import _run_decision, _update_incident
|
||||
|
||||
NOW = datetime(2026, 8, 16, 21, 0, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _ctx(**overrides) -> dict:
|
||||
"""Context with no active incidents, so the decision reaches the creation gate."""
|
||||
base = {
|
||||
"call_id": "call-1",
|
||||
"all_active": [],
|
||||
"recent": [],
|
||||
"call_doc": {},
|
||||
"call_embedding": None,
|
||||
"call_units": [],
|
||||
"call_vehicles": [],
|
||||
"call_cleared": [],
|
||||
"call_severity": "routine",
|
||||
"coords": None,
|
||||
"is_thin_call": True,
|
||||
"now": NOW,
|
||||
"system_id": "sys-1",
|
||||
"talkgroup_id": 9048,
|
||||
"talkgroup_name": "MTA PD Districts 6/7/11 - Police Dispatch",
|
||||
"tags": [],
|
||||
"incident_type": None,
|
||||
"location": None,
|
||||
"location_coords": None,
|
||||
"reassignment": False,
|
||||
"create_if_new": True,
|
||||
}
|
||||
base.update(overrides)
|
||||
return base
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Creation gate — severity decides incident-worthiness, not incident_type
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_routine_status_traffic_stays_orphaned():
|
||||
"""A content-free acknowledgement must not open an incident of its own."""
|
||||
assert _run_decision(_ctx())["action"] == "orphan"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("severity", ["minor", "moderate", "major"])
|
||||
def test_any_real_severity_opens_an_untyped_incident(severity):
|
||||
decision = _run_decision(_ctx(call_severity=severity))
|
||||
assert decision["action"] == "new"
|
||||
assert decision["incident_type"] == "other"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("field,value", [
|
||||
("call_units", ["6 Adam"]),
|
||||
("call_vehicles", ["RMP 22146"]),
|
||||
("coords", {"lat": 41.0, "lng": -73.8}),
|
||||
("location", "District 6"),
|
||||
("tags", ["prisoner-transport"]),
|
||||
])
|
||||
def test_concrete_content_opens_an_untyped_incident(field, value):
|
||||
"""Routine severity is overridden by anything the extractor actually found."""
|
||||
decision = _run_decision(_ctx(**{field: value}))
|
||||
assert decision["action"] == "new"
|
||||
assert decision["incident_type"] == "other"
|
||||
|
||||
|
||||
def test_explicit_type_is_never_downgraded_to_other():
|
||||
decision = _run_decision(_ctx(incident_type="police", call_severity="moderate"))
|
||||
assert decision["action"] == "new"
|
||||
assert decision["incident_type"] == "police"
|
||||
|
||||
|
||||
def test_other_survives_extraction_and_creates_an_incident():
|
||||
""""other" is a real classification now, not a synonym for unclassifiable."""
|
||||
decision = _run_decision(_ctx(incident_type="other"))
|
||||
assert decision["action"] == "new"
|
||||
assert decision["incident_type"] == "other"
|
||||
|
||||
|
||||
def test_sweep_never_creates_incidents():
|
||||
"""The re-correlation sweep passes create_if_new=False — it may only link."""
|
||||
decision = _run_decision(_ctx(call_severity="major", create_if_new=False))
|
||||
assert decision["action"] == "orphan"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Thin calls attach for context but do not count as incident activity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _incident(idle_minutes: float) -> dict:
|
||||
updated = NOW - timedelta(minutes=idle_minutes)
|
||||
return {
|
||||
"incident_id": "inc-1",
|
||||
"system_ids": ["sys-1"],
|
||||
"talkgroup_ids": ["9048"],
|
||||
"updated_at": updated.isoformat(),
|
||||
"started_at": updated.isoformat(),
|
||||
"status": "active",
|
||||
}
|
||||
|
||||
|
||||
def test_thin_call_links_to_the_active_incident_on_its_talkgroup():
|
||||
inc = _incident(0.2)
|
||||
decision = _run_decision(_ctx(all_active=[inc], recent=[inc]))
|
||||
assert decision["action"] == "link"
|
||||
assert decision["corr_debug"]["corr_path"] == "fast/thin"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_thin_link_does_not_refresh_updated_at():
|
||||
with patch("app.internal.incident_correlator.fstore") as mock_fstore:
|
||||
mock_fstore.doc_set = AsyncMock()
|
||||
await _update_incident(
|
||||
_incident(5), "call-1", 9048, "sys-1", [], None, None, [], [], None, NOW,
|
||||
refresh_activity=False,
|
||||
)
|
||||
updates = mock_fstore.doc_set.await_args.args[2]
|
||||
assert "updated_at" not in updates, "a '10-4' must not reset the incident idle clock"
|
||||
assert updates["last_thin_at"] == NOW.isoformat()
|
||||
assert updates["summary_stale"] is True, "the call still belongs in the summary"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_substantive_link_does_refresh_updated_at():
|
||||
with patch("app.internal.incident_correlator.fstore") as mock_fstore:
|
||||
mock_fstore.doc_set = AsyncMock()
|
||||
await _update_incident(
|
||||
_incident(5), "call-1", 9048, "sys-1", [], None, None, ["6 Adam"], [], None, NOW,
|
||||
)
|
||||
updates = mock_fstore.doc_set.await_args.args[2]
|
||||
assert updates["updated_at"] == NOW.isoformat()
|
||||
assert "last_thin_at" not in updates
|
||||
Reference in New Issue
Block a user