diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 712043e..90288c8 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -65,6 +65,41 @@ class Settings(BaseSettings): # one was >= 8.2, so 5 separates them with room on both sides. Genuine # back-and-forth is handled by the 30-second tier-1 path above this. tg_dispatch_thin_idle_minutes: int = 5 + # Every other channel: tier-2 thin calls attach to a lone candidate idle < this. + # Non-dispatch talkgroups previously had NO tier-2 bound at all — they used the + # whole 90-minute tg_fast_path_idle_minutes window with no single-candidate + # requirement and no fit test, which is the widest version of the 2026-08-20 + # over-merge. A tactical channel really is dedicated to one scene, so it earns + # a longer window than a dispatch backbone, but not an unbounded one: 15 sits + # inside the 20-minute tactical-default window in _call_fits_incident, so the + # no-evidence thin path is never more permissive than the fit-tested path on + # the same channel. + tg_thin_idle_minutes: int = 15 + + # ── Hard caps: an incident past either of these stops accepting calls ────── + # Enforced on every correlation path (see _incident_at_capacity). Pairwise fit + # tests judge one call against one incident and cannot see the shape of the + # chain they are building, so these are the only guard against a "work shift" + # incident regardless of how individually plausible each link looked. + # + # 120 minutes: the one incident in the 2026-08-20 dump that was genuinely a + # single event ran 63 minutes (06:15 wrong-way driver → 07:18 closeout), so + # the cap has to clear an hour with real headroom. The four junk chains ran + # 3h41m, 3h43m, 4h05m and 4h09m, so it has to sit well under three hours. + # 120 also equals correlation_window_hours: the location and slow paths + # already refuse to consider a candidate older than that, and the fast path + # was the only one exempt. Making it agree removes that inconsistency rather + # than inventing a new number. + incident_max_duration_minutes: int = 120 + # 40 calls: a backstop for a burst that fills up inside the duration cap + # rather than the primary bound. The worst observed chain averaged ~16 + # calls/hour while absorbing an ENTIRE dispatch backbone, so 40 calls in + # under two hours means the incident is eating most of the channel — that is + # a chain, not an event. Set deliberately above any plausible single-incident + # call volume (a multi-alarm fire on its own tactical channel) so this cap + # errs toward keeping real incidents whole and lets the duration cap do the + # cutting. + incident_max_calls: int = 40 # Vocabulary learning vocabulary_induction_interval_hours: int = 24 # how often the induction loop runs diff --git a/drb-c2-core/app/internal/incident_correlator.py b/drb-c2-core/app/internal/incident_correlator.py index c4ceb8f..03ee7a2 100644 --- a/drb-c2-core/app/internal/incident_correlator.py +++ b/drb-c2-core/app/internal/incident_correlator.py @@ -21,9 +21,16 @@ Matching priority (in order): or location proximity) to link. Without this, every ungeocoded call on a dispatch backbone would link to the one active incident on that channel. - Thin calls (no units/vehicles/coords) skip scene verification and link to the - most recently updated incident on this TGID — but only if that incident is - within the recency window. + Thin calls — genuinely content-free housekeeping, see `_is_thin_call` — skip + scene verification and link to the most recently updated incident on this + TGID, but only inside a tight conversational window (30s any-candidate, then + single-candidate up to the channel's thin window). They are the one class of + call that links without a fit test, so that window is the whole guard. + +0. Hard caps (`_incident_at_capacity`) — an incident past + `incident_max_duration_minutes` or `incident_max_calls` stops being a + candidate on every path below, including the LLM tier. Pairwise fit tests + cannot see the shape of a chain; only a cap can. 2. Location path — geocoded coords within `location_proximity_km` (time-limited) Primary mutual-aid signal: EMS + police at the same scene. @@ -253,6 +260,153 @@ def _incident_idle_minutes(inc: dict, now: datetime) -> float: return 9999.0 +def _idle_gate_minutes(inc: dict, now: datetime) -> float: + """ + Absolute distance in minutes between `now` and the incident's last activity — + the value every recency gate must compare against. + + `_incident_idle_minutes` is signed and can go NEGATIVE: the re-correlation + sweep anchors `now` to the call's own `started_at`, so a back-dated call is + routinely evaluated against an incident that was updated later. Observed + 2026-08-20 on incident `9d376ffe`: `corr_incident_idle_min: -4.1`. Every + `idle <= window` gate in this module reads True for a negative number, so + those gates silently stopped bounding anything for exactly the calls the + sweep re-examines. Distance in either direction is what the gates actually + mean, so they compare against the magnitude and the signed value is kept + only for the debug field. + """ + return abs(_incident_idle_minutes(inc, now)) + + +def _incident_span_minutes(inc: dict, now: datetime) -> float: + """ + Wall-clock minutes the incident has been open: from its `started_at` to the + later of `now` and its last update. The `max` matters because the + re-correlation sweep passes a back-dated `now` (the call's own started_at), + and a raw `now - started_at` would understate — or invert — the span for + exactly the calls most likely to be force-attached to an old chain. + + Returns 0.0 when `started_at` is unparseable, so a malformed timestamp can + never be the sole reason an incident is capped. + """ + try: + raw = inc.get("started_at") or "" + started = datetime.fromisoformat(str(raw).replace("Z", "+00:00")) + if started.tzinfo is None: + started = started.replace(tzinfo=timezone.utc) + except Exception: + return 0.0 + end = now + try: + raw_u = inc.get("updated_at") or "" + updated = datetime.fromisoformat(str(raw_u).replace("Z", "+00:00")) + if updated.tzinfo is None: + updated = updated.replace(tzinfo=timezone.utc) + # The sweep evaluates old calls against incidents that have since moved + # on, so `now` alone understates the span. Take whichever is later. + end = max(now, updated) + except Exception: + pass + return max((end - started).total_seconds() / 60, 0.0) + + +def _incident_at_capacity(inc: dict, now: datetime) -> Optional[str]: + """ + Return a reason string when this incident has grown past the hard caps and + must not absorb any more calls, or None when it is still open for business. + + These caps are deliberately path-independent. Every fit test in this module + is a *pairwise* judgement — does THIS call belong with THAT incident — and + each one can be individually defensible while the chain they build is not. + `f5190670` (2026-08-20 dump) is the proof: 68 calls, 4h09m, 44 units, 12 + tags and at least 13 genuinely distinct events, every link arrived at one + call at a time. No pairwise rule catches that, because the mistake is the + accumulated shape, not any single link. A cap is the only guard that can + see the shape, so it is applied once, to the candidate pool itself, before + any path gets to choose — which is also why it binds the LLM tier, since + that reads the same ctx["recent"] / ctx["all_active"] lists. + + Capping does not delete or truncate anything: the incident keeps the calls + it has and still auto-resolves on the normal idle sweep. It just stops + being a candidate, so the next call opens a fresh incident. + """ + call_count = len(inc.get("call_ids") or []) + if call_count >= settings.incident_max_calls: + return f"call_cap:{call_count}" + span = _incident_span_minutes(inc, now) + if span > settings.incident_max_duration_minutes: + return f"duration_cap:{span:.0f}min" + return None + + +def _drop_capped(incidents: list[dict], now: datetime) -> list[dict]: + """Remove over-cap incidents from a candidate pool (see _incident_at_capacity).""" + kept: list[dict] = [] + for inc in incidents: + reason = _incident_at_capacity(inc, now) + if reason: + logger.info( + f"Correlator: incident {inc.get('incident_id', '?')} is at capacity " + f"({reason}) — excluded as a correlation candidate" + ) + continue + kept.append(inc) + return kept + + +def _is_thin_call( + call_units: list, + call_vehicles: list, + coords: Optional[dict], + tags: Optional[list], + location: Optional[str], + call_severity: Optional[str], + reassignment: bool, +) -> bool: + """ + True only for genuinely content-free radio housekeeping — "10-4", "Copy", + "En route" — the traffic that legitimately has no evidence of its own and + can only be placed by conversational context. + + This test used to be `not units and not vehicles and not coords`, which is + where the 2026-08-20 junk chains came from. Thin calls are the ONE class + that attaches without a `_call_fits_incident` check, so anything wrongly + called thin is force-merged into whatever was most recent. Six substantive + dispatches in that dump qualified as thin purely because no unit ID parsed + and the geocode failed — including *"All units head over to the powerhouse, + 55 Hyman Hills Road … she's 87 years old"*, a brand-new job that attached to + a four-hour chain and then overwrote its location and title. + + So the test now also rejects a call as thin when it carries: + • tags — the extractor classified the event, so there IS content + • a location — the call names a place; that is a claim to be checked, + not context-free chatter + • real severity (minor/moderate/major) — the extractor judged it an event + • reassignment — dispatch pulling a unit to a NEW job, by definition not + a continuation. `upload.py` blanks `units` on these to + stop unit-overlap chaining, which used to make the call + thin and route it to the one path with no fit check — + the guard produced the merge it existed to prevent. + + Such calls now go through the normal fit-tested path. On a dispatch channel + with no positive signal that means they open their own incident or orphan, + which is the correct direction: a wrongly-separate incident is visibly + wrong and can be merged later, a wrongly-merged one silently corrupts + every unit, tag, severity and map pin on the incident it joined. + """ + if call_units or call_vehicles or coords: + return False + if tags: + return False + if location and str(location).strip(): + return False + if call_severity in ("minor", "moderate", "major"): + return False + if reassignment: + return False + return True + + # ───────────────────────────────────────────────────────────────────────────── # Public API # ───────────────────────────────────────────────────────────────────────────── @@ -376,6 +530,10 @@ async def _build_context( all_active = await fstore.collection_list("incidents", status="active", org_id=org_id) else: all_active = await fstore.collection_list("incidents", status="active") + # Incidents past the hard caps are removed from the candidate pool here, so + # neither the rules engine nor the LLM tier (which reads ctx["recent"] / + # ctx["all_active"]) can propose linking into one. + all_active = _drop_capped(all_active, now) recent = [inc for inc in all_active if _within_window_of(inc, now, window)] call_embedding = call_doc.get("embedding") @@ -384,7 +542,9 @@ async def _build_context( call_cleared = cleared_units if cleared_units is not None else (call_doc.get("cleared_units") or []) 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 + is_thin_call = _is_thin_call( + call_units, call_vehicles, coords, tags, location, call_severity, reassignment + ) return { "call_id": call_id, "org_id": org_id, "all_active": all_active, "recent": recent, @@ -415,8 +575,23 @@ def _run_decision(ctx: dict) -> dict: incident_type resolved type (action == "new") corr_debug fields to write to the call doc """ - all_active = ctx["all_active"] - recent = ctx["recent"] + now = ctx["now"] + # Hard caps, enforced regardless of which path would have matched. Normally + # a no-op because _build_context already filtered these out; re-applied here + # so the guarantee holds for any caller that assembles a ctx directly + # (tests, the LLM tiebreak path) rather than through _build_context. + _capped_ids = { + inc.get("incident_id") + for inc in (list(ctx["all_active"]) + list(ctx["recent"])) + if _incident_at_capacity(inc, now) + } + if _capped_ids: + logger.info( + f"Correlator: {len(_capped_ids)} incident(s) at capacity, excluded as " + f"candidates for call {ctx['call_id']}: {sorted(_capped_ids)}" + ) + all_active = [inc for inc in ctx["all_active"] if inc.get("incident_id") not in _capped_ids] + recent = [inc for inc in ctx["recent"] if inc.get("incident_id") not in _capped_ids] call_doc = ctx["call_doc"] call_embedding = ctx["call_embedding"] call_units = ctx["call_units"] @@ -424,7 +599,6 @@ def _run_decision(ctx: dict) -> dict: call_severity = ctx["call_severity"] coords = ctx["coords"] is_thin_call = ctx["is_thin_call"] - now = ctx["now"] system_id = ctx["system_id"] talkgroup_id = ctx["talkgroup_id"] talkgroup_name = ctx["talkgroup_name"] @@ -469,7 +643,7 @@ def _run_decision(ctx: dict) -> dict: # Apply recency gate — only incidents active within the rolling window. tg_recent = [ inc for inc in tg_matches - if _incident_idle_minutes(inc, now) <= settings.tg_fast_path_idle_minutes + if _idle_gate_minutes(inc, now) <= settings.tg_fast_path_idle_minutes ] if tg_matches and not tg_recent: @@ -479,43 +653,58 @@ def _run_decision(ctx: dict) -> dict: ) if tg_recent and is_thin_call: - # Content-free status calls ("10-4", "Copy", "En route") — two tiers: + # Content-free status calls ("10-4", "Copy", "En route") — the only + # class of call that links without a `_call_fits_incident` check, + # because by construction it has no unit, vehicle, coordinate, tag, + # location or severity to test. There is therefore nothing to fit; + # the only honest evidence is "someone just said something on this + # channel and this is the reply". That is a claim about SECONDS, + # so the window is the whole guard and it has to be tight. # - # Tier 1 — ≤30 seconds idle: this is a direct conversational reply to - # whatever was just transmitted. Attach to the most recently updated - # incident regardless of how many are active; within 30 seconds, the - # "most recently updated" IS the active thread. + # Tier 1 — ≤30 seconds idle: a direct conversational reply. Attach + # to the most recently updated incident regardless of how many are + # active; within 30 seconds, "most recently updated" IS the live + # thread on the channel. # - # Tier 2 — 30 seconds to tg_dispatch_thin_idle_minutes: channel context - # is less clear. Only attach when there is exactly ONE candidate to - # avoid guessing on a busy multi-incident channel. - if is_dispatch: - THIN_CONVERSATIONAL_SECS = 30 - very_recent = [ - inc for inc in tg_recent - if _incident_idle_minutes(inc, now) * 60 <= THIN_CONVERSATIONAL_SECS - ] - if very_recent: - # Tier 1: direct conversational reply — most recent wins. - thin_pool = [max(very_recent, key=lambda inc: inc.get("updated_at", ""))] - logger.info( - f"Correlator fast-path thin (tier-1, ≤{THIN_CONVERSATIONAL_SECS}s): " - f"using most-recent of {len(very_recent)} candidate(s) for call {call_id}" - ) - else: - # Tier 2: less certain — require single candidate. - thin_pool = [ - inc for inc in tg_recent - if _incident_idle_minutes(inc, now) <= settings.tg_dispatch_thin_idle_minutes - ] - if len(thin_pool) > 1: - logger.info( - f"Correlator fast-path thin (tier-2): {len(thin_pool)} active incidents " - f"on dispatch channel — ambiguous, skipping thin call {call_id}" - ) - thin_pool = [] + # Tier 2 — 30 seconds up to the channel's thin window: context is + # less clear. Only attach when there is exactly ONE candidate, so + # we never guess on a busy multi-incident channel. + # + # This bounding used to apply to dispatch channels only; every other + # talkgroup fell through to `thin_pool = tg_recent`, i.e. any + # incident idle up to tg_fast_path_idle_minutes (90) with no + # single-candidate requirement and no fit test of any kind. Four + # hours is not a bound, and neither is ninety minutes. + THIN_CONVERSATIONAL_SECS = 30 + thin_window_min = ( + settings.tg_dispatch_thin_idle_minutes if is_dispatch + else settings.tg_thin_idle_minutes + ) + very_recent = [ + inc for inc in tg_recent + if _idle_gate_minutes(inc, now) * 60 <= THIN_CONVERSATIONAL_SECS + ] + if very_recent: + # Tier 1: direct conversational reply — most recent wins. + thin_pool = [max(very_recent, key=lambda inc: inc.get("updated_at", ""))] + logger.info( + f"Correlator fast-path thin (tier-1, ≤{THIN_CONVERSATIONAL_SECS}s): " + f"using most-recent of {len(very_recent)} candidate(s) for call {call_id}" + ) else: - thin_pool = tg_recent + # Tier 2: less certain — require a single candidate inside the + # channel's thin window. + thin_pool = [ + inc for inc in tg_recent + if _idle_gate_minutes(inc, now) <= thin_window_min + ] + if len(thin_pool) > 1: + logger.info( + f"Correlator fast-path thin (tier-2): {len(thin_pool)} active incidents " + f"on {'dispatch' if is_dispatch else 'tactical'} channel — " + f"ambiguous, skipping thin call {call_id}" + ) + thin_pool = [] if not thin_pool: logger.info( @@ -627,7 +816,7 @@ def _run_decision(ctx: dict) -> dict: # has almost certainly moved on or the incident closed. unit_candidates = [ inc for inc in unit_candidates - if _incident_idle_minutes(inc, now) <= settings.unit_continuity_max_idle_minutes + if _idle_gate_minutes(inc, now) <= settings.unit_continuity_max_idle_minutes ] if unit_candidates: best_unit_inc = max(unit_candidates, key=lambda i: i.get("updated_at", "")) diff --git a/drb-c2-core/tests/test_correlator_merge_caps.py b/drb-c2-core/tests/test_correlator_merge_caps.py new file mode 100644 index 0000000..c0beadc --- /dev/null +++ b/drb-c2-core/tests/test_correlator_merge_caps.py @@ -0,0 +1,470 @@ +""" +Over-merge guards — server-26#22. + +All of this comes from the 2026-08-20 production dump (CORRELATION_REVIEW_0820.md), +where 4 of 6 sampled incidents were junk chains and the worst, `f5190670`, was +68 calls over 4h09m carrying 44 units, 12 tags and at least 13 genuinely distinct +events. That is a work shift filed as one incident. + +Three defects combined to produce it, and each has cases below: + + 1. `is_thin_call` was `not units and not vehicles and not coords`, so a real + dispatch with tags and a street address counted as thin whenever no unit ID + parsed and the geocode failed. Thin calls are the one class that links with + NO `_call_fits_incident` check, so those dispatches were force-merged. + 2. The thin path was bounded only on dispatch channels. Everywhere else it + used the full 90-minute fast-path window, any number of candidates, no fit. + 3. Nothing capped an incident's total size. Every fit test in the correlator + is pairwise, so each individual link can be defensible while the chain they + accumulate is not — no pairwise rule can see the shape. + +`_run_decision` and `_is_thin_call` are pure, so none of this needs Firestore. +""" +import pytest +from datetime import datetime, timedelta, timezone +from app.config import settings +from app.internal.incident_correlator import ( + _run_decision, _is_thin_call, _idle_gate_minutes, + _incident_at_capacity, _incident_span_minutes, +) + +NOW = datetime(2026, 8, 20, 7, 0, 0, tzinfo=timezone.utc) + +# TG 383 from the dump: "Ch 1 (Patched with 155.310)". _DISPATCH_TG_RE matches +# "patched", so this is a shared dispatch backbone carrying the whole department. +DISPATCH_TG = "Ch 1 (Patched with 155.310)" +TACTICAL_TG = "Fireground 2" + + +def _ctx(**overrides) -> dict: + 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": 383, + "talkgroup_name": DISPATCH_TG, + "tags": [], + "incident_type": None, + "location": None, + "location_coords": None, + "reassignment": False, + "create_if_new": True, + } + base.update(overrides) + return base + + +def _incident(idle_minutes: float = 0.2, **overrides) -> dict: + updated = NOW - timedelta(minutes=idle_minutes) + inc = { + "incident_id": "inc-1", + "system_ids": ["sys-1"], + "talkgroup_ids": ["383"], + "updated_at": updated.isoformat(), + "started_at": updated.isoformat(), + "status": "active", + "call_ids": ["seed-call"], + } + inc.update(overrides) + return inc + + +# --------------------------------------------------------------------------- +# 1. What counts as thin — the misclassification that drove the chains +# --------------------------------------------------------------------------- + +def test_content_free_acknowledgement_is_thin(): + """"10-4." — no unit, no vehicle, no coords, no tags, no place, routine.""" + assert _is_thin_call([], [], None, [], None, "routine", False) is True + + +@pytest.mark.parametrize("field,value", [ + ("tags", ["welfare-check"]), + ("location", "55 Hyman Hills Road"), + ("call_severity", "minor"), + ("call_severity", "moderate"), + ("call_severity", "major"), +]) +def test_extracted_content_makes_a_call_substantive(field, value): + """ + The 07:08 call in `f5190670`: "All units head over to the powerhouse, 55 + Hyman Hills Road … she's 87 years old" — a brand new job that was called + thin purely because no unit ID parsed and the geocode failed. It attached + with no fit check and then overwrote the four-hour chain's title and pin. + """ + kwargs = {"tags": [], "location": None, "call_severity": "routine"} + kwargs[field] = value + assert _is_thin_call([], [], None, kwargs["tags"], kwargs["location"], + kwargs["call_severity"], False) is False + + +def test_reassignment_is_never_thin(): + """ + upload.py blanks `units` when dispatch pulls a unit onto a NEW job, to stop + unit-overlap chaining. That made the call thin and routed it to the only + path with no fit check — the guard produced the merge it existed to prevent. + """ + assert _is_thin_call([], [], None, [], None, "routine", True) is False + + +@pytest.mark.parametrize("units,vehicles,coords", [ + (["6-Adam"], [], None), + ([], ["black Toyota Camry"], None), + ([], [], {"lat": 41.08, "lng": -73.81}), +]) +def test_original_thinness_signals_still_apply(units, vehicles, coords): + assert _is_thin_call(units, vehicles, coords, [], None, "routine", False) is False + + +def test_blank_location_string_does_not_make_a_call_substantive(): + assert _is_thin_call([], [], None, [], " ", "routine", False) is True + + +# --------------------------------------------------------------------------- +# 2. A thin call needs a tight window; a real one needs a fit signal +# --------------------------------------------------------------------------- + +def test_thin_call_with_no_overlap_does_not_attach_on_a_dispatch_channel(): + inc = _incident(idle_minutes=40) + assert _run_decision(_ctx(all_active=[inc], recent=[inc]))["action"] == "orphan" + + +def test_thin_call_with_no_overlap_does_not_attach_on_a_tactical_channel(): + """ + The widest version of the bug: non-dispatch talkgroups skipped the tiering + entirely and used the whole 90-minute fast-path window with no + single-candidate requirement, so ANY thin call joined whatever was newest. + """ + inc = _incident(idle_minutes=40) + decision = _run_decision(_ctx( + all_active=[inc], recent=[inc], talkgroup_name=TACTICAL_TG, + )) + assert decision["action"] == "orphan" + + +def test_tactical_thin_call_still_attaches_inside_its_own_window(): + """Bounded, not removed — a "10-4" on a working channel is still context.""" + inc = _incident(idle_minutes=settings.tg_thin_idle_minutes - 1) + decision = _run_decision(_ctx( + all_active=[inc], recent=[inc], talkgroup_name=TACTICAL_TG, + )) + assert decision["action"] == "link" + assert decision["corr_debug"]["corr_path"] == "fast/thin" + + +def test_tactical_thin_call_is_ambiguous_with_two_candidates(): + a = _incident(idle_minutes=3.0, incident_id="inc-a") + b = _incident(idle_minutes=4.0, incident_id="inc-b") + decision = _run_decision(_ctx( + all_active=[a, b], recent=[a, b], talkgroup_name=TACTICAL_TG, + )) + assert decision["action"] == "orphan" + + +def test_call_with_unit_overlap_does_attach(): + """ + Positive control: real evidence still links. Carrying units also means the + call is not thin, so it reaches _call_fits_incident and passes on + unit_overlap rather than being force-attached. + """ + inc = _incident(idle_minutes=3.0, units=["6-Adam", "K-9A2"]) + assert _is_thin_call(["6-Adam"], [], None, [], None, "routine", False) is False + decision = _run_decision(_ctx( + all_active=[inc], recent=[inc], call_units=["6-Adam"], is_thin_call=False, + )) + assert decision["action"] == "link" + assert decision["corr_debug"]["corr_fit_signal"] == "unit_overlap" + + +def test_substantive_call_with_no_signal_opens_its_own_incident(): + """ + A tagged dispatch on a shared backbone with no unit/vehicle/geocode match + is a separate job, not a follow-up. Under the old thinness test this exact + call took fast/thin and merged. + """ + inc = _incident(idle_minutes=2.0) + decision = _run_decision(_ctx( + all_active=[inc], recent=[inc], is_thin_call=False, + tags=["welfare-check"], location="55 Hyman Hills Road", + )) + assert decision["action"] == "new" + assert decision["incident_type"] == "other" + + +# --------------------------------------------------------------------------- +# 3. Negative idle — the sweep anchors `now` to the call's own started_at +# --------------------------------------------------------------------------- + +def test_idle_gate_uses_distance_not_sign(): + """ + Observed on `9d376ffe`: corr_incident_idle_min -4.1, because the sweep + evaluated a 02:45 call against an incident updated at 02:50. Every + `idle <= window` gate reads True for a negative number, so the gates + stopped bounding anything for precisely the calls the sweep re-examines. + """ + future = _incident(idle_minutes=-45) + assert _idle_gate_minutes(future, NOW) == pytest.approx(45.0) + + +def test_back_dated_thin_call_does_not_sail_through_the_recency_gate(): + future = _incident(idle_minutes=-45) + assert _run_decision(_ctx(all_active=[future], recent=[future]))["action"] == "orphan" + + +# --------------------------------------------------------------------------- +# 4. Hard caps — path-independent, because pairwise fit tests can't see shape +# --------------------------------------------------------------------------- + +def _long_running(minutes: float) -> dict: + started = NOW - timedelta(minutes=minutes) + return _incident(idle_minutes=0.2, started_at=started.isoformat()) + + +def test_duration_cap_forces_a_new_incident(): + """ + `f5190670` ran 4h09m. The one real incident in the dump ran 63 minutes. + The unit here overlaps the incident's own roster, so without the cap this + links on unit_overlap — the cap is the only thing separating them. + """ + inc = _long_running(settings.incident_max_duration_minutes + 30) + inc["units"] = ["6-Adam"] + decision = _run_decision(_ctx( + all_active=[inc], recent=[inc], is_thin_call=False, + call_units=["6-Adam"], tags=["welfare-check"], + )) + assert decision["action"] == "new" + + +def test_duration_cap_blocks_the_thin_path_too(): + """The cap is checked before any path runs, so 'no fit test' is no escape.""" + inc = _long_running(settings.incident_max_duration_minutes + 30) + assert _run_decision(_ctx(all_active=[inc], recent=[inc]))["action"] == "orphan" + + +def test_incident_just_under_the_duration_cap_still_accepts_calls(): + inc = _long_running(settings.incident_max_duration_minutes - 10) + inc["units"] = ["6-Adam"] + decision = _run_decision(_ctx( + all_active=[inc], recent=[inc], is_thin_call=False, call_units=["6-Adam"], + )) + assert decision["action"] == "link" + + +def test_call_count_cap_forces_a_new_incident(): + inc = _incident(idle_minutes=0.2, units=["6-Adam"]) + inc["call_ids"] = [f"c{i}" for i in range(settings.incident_max_calls)] + decision = _run_decision(_ctx( + all_active=[inc], recent=[inc], is_thin_call=False, + call_units=["6-Adam"], tags=["vehicle-accident"], + )) + assert decision["action"] == "new" + + +def test_incident_one_call_under_the_count_cap_still_accepts_calls(): + inc = _incident(idle_minutes=0.2, units=["6-Adam"]) + inc["call_ids"] = [f"c{i}" for i in range(settings.incident_max_calls - 1)] + decision = _run_decision(_ctx( + all_active=[inc], recent=[inc], is_thin_call=False, call_units=["6-Adam"], + )) + assert decision["action"] == "link" + + +@pytest.mark.parametrize("inc,expect", [ + (_incident(), None), + (_long_running(9999), "duration"), +]) +def test_capacity_reason_names_the_cap_that_fired(inc, expect): + reason = _incident_at_capacity(inc, NOW) + assert (reason is None) if expect is None else reason.startswith(expect) + + +def test_span_survives_a_back_dated_reference_time(): + """ + The sweep passes the call's own started_at as `now`, which can precede the + incident's last update — the span must still reflect what the incident has + actually accumulated, not go negative and defeat the cap. + """ + started = NOW - timedelta(hours=5) + inc = {"started_at": started.isoformat(), "updated_at": NOW.isoformat()} + past = NOW - timedelta(hours=4) + assert _incident_span_minutes(inc, past) == pytest.approx(300.0, abs=0.1) + + +def test_unparseable_started_at_is_never_capped_on_duration(): + assert _incident_span_minutes({"started_at": "not-a-date"}, NOW) == 0.0 + + +# --------------------------------------------------------------------------- +# 5. Regression: the `f5190670` shape must not reassemble +# --------------------------------------------------------------------------- + +# The 13 distinct events visible in `f5190670`, at their real offsets from the +# 03:01 opener. Each arrived exactly as reproduced here: tags and often a place +# name, but no parsed unit and no successful geocode — which is what made the +# old thinness test classify them as chatter. +_F5190670_EVENTS = [ + (0, ["vehicle-accident", "telephone-pole-strike"], "Airport Road traffic circle"), + (2, ["uber-passenger", "phone-pinging"], "traffic circle near New King Street"), + (13, ["sign-down"], None), + (26, ["burglary-alarm"], "34 Carlton Drive"), + (67, ["inspection"], "2 Filno River Road"), + (108, ["disabled-vehicle"], "Yonkers Ave"), + (119, ["altercation"], "137 East Main Street"), + (131, ["inspection"], "80 North Grasslands Road"), + (156, ["foot-patrol"], "Tanzania Road"), + (211, [], None), # unit roll call — pure noise + (222, ["vehicle-off-roadway"], None), + (240, ["premises-check"], "Hillcrest Drive"), + (247, ["welfare-check"], "55 Hyman Hills Road"), +] + + +# The department roster heard on that channel. `f5190670` accumulated 44 units, +# partly from the nine phonetic-alphabet roll calls it absorbed, and once an +# incident holds most of the roster essentially every later call overlaps it — +# mechanism B in the review, unit-overlap positive feedback. Reproduced here so +# the chain has a real engine driving it, not just the thin path. +_ROSTER = ["6-Adam", "7-Baker", "11-Victor", "45-Charlie", "K-9A2", "22-47"] + +_START = datetime(2026, 5, 24, 3, 1, 0, tzinfo=timezone.utc) + + +def _overnight_traffic(): + """ + The real shape of that channel: a transmission roughly every two minutes for + 4h07m — 13 dispatched jobs, routine unit traffic drawn from one roster, and + acknowledgements in between. Yields (offset_min, units, tags, location). + """ + events = {off: (tags, loc) for off, tags, loc in _F5190670_EVENTS} + # The dispatches, at their real offsets, exactly as they arrived: tags and + # usually a place name, but no parsed unit and no successful geocode. + traffic = [(off, [], tags, loc) for off, (tags, loc) in events.items()] + + # Mechanism B, the engine that kept the real chain alive for four hours: one + # job that legitimately opens with units, then keeps producing unit traffic + # all shift. Each follow-up genuinely overlaps on unit, so each link is + # individually defensible and each one refreshes updated_at — which keeps + # the incident permanently inside every recency gate. No pairwise fit test + # can refuse these; only a cap can stop the accumulation. + traffic.append((1, [_ROSTER[0]], ["prisoner-transport"], "Medical Center")) + traffic += [(m, [_ROSTER[0]], [], None) for m in range(5, 249, 4)] + + # Everything else on the channel — one transmission every two minutes. + for n, minute in enumerate(range(0, 249, 2)): + if minute in events: + continue + if n % 3 == 0: + traffic.append((minute, [_ROSTER[1 + n % (len(_ROSTER) - 1)]], [], None)) + else: + traffic.append((minute, [], [], None)) # "10-4" + return traffic + + +def _simulate(traffic): + """ + Replay traffic through the real decision engine, applying the same incident + mutations the commit layer would: a link appends the call, merges units, and + (unless thin) refreshes updated_at; "new" opens a doc. Returns + (incidents, placement) where placement maps call_id → incident_id. + """ + incidents: list[dict] = [] + placement: dict[str, str] = {} + for i, (offset, units, tags, location) in enumerate(sorted(traffic)): + now = _START + timedelta(minutes=offset) + call_id = f"call-{i}" + thin = _is_thin_call(units, [], None, tags, location, "routine", False) + active = [inc for inc in incidents if inc["status"] == "active"] + decision = _run_decision(_ctx( + call_id=call_id, now=now, all_active=active, recent=active, + tags=tags, location=location, call_units=units, is_thin_call=thin, + )) + if decision["action"] == "link": + inc = decision["matched_incident"] + inc["call_ids"].append(call_id) + inc["units"] = list(dict.fromkeys(inc["units"] + units)) + inc["_last_call_at"] = now + if decision["corr_debug"].get("corr_path") != "fast/thin": + inc["updated_at"] = now.isoformat() + placement[call_id] = inc["incident_id"] + elif decision["action"] == "new": + incidents.append({ + "incident_id": f"inc-{i}", "system_ids": ["sys-1"], + "talkgroup_ids": ["383"], "status": "active", + "started_at": now.isoformat(), "updated_at": now.isoformat(), + "call_ids": [call_id], "units": list(units), + "_started": now, "_last_call_at": now, "_offset": offset, + }) + placement[call_id] = incidents[-1]["incident_id"] + return incidents, placement + + +def _live_span_minutes(inc: dict) -> float: + """Minutes from an incident's first call to the last one it actually took.""" + return (inc["_last_call_at"] - inc["_started"]).total_seconds() / 60 + + +def test_f5190670_does_not_become_one_incident(): + """ + The headline regression: a full overnight shift on one patched dispatch + backbone must not end up as a single incident. The real one was 68 calls, + 4h09m, 44 units, 12 tags and at least 13 distinct events. + """ + traffic = _overnight_traffic() + incidents, placement = _simulate(traffic) + + assert len(incidents) >= 12, f"the shift merged into {len(incidents)} incident(s)" + + # Without the caps this same traffic produces a 125-call incident spanning + # 244 minutes; with the old thinness test on top of that, 153 calls over + # 247 minutes in 3 incidents — the `f5190670` shape, reproduced. + biggest = max(len(inc["call_ids"]) for inc in incidents) + assert biggest <= settings.incident_max_calls, ( + f"one incident holds {biggest} calls, past the " + f"{settings.incident_max_calls}-call cap" + ) + + longest = max(_live_span_minutes(inc) for inc in incidents) + assert longest <= settings.incident_max_duration_minutes, ( + f"an incident took calls across {longest:.0f}min, past the " + f"{settings.incident_max_duration_minutes}min cap" + ) + + +def test_each_dispatched_job_gets_its_own_incident(): + """ + The 13 events are unrelated jobs — a pole strike, a burglar alarm, two + inspections, an altercation, a welfare check. On a dispatch backbone with no + unit or geocode tying them together, none of them may join another's + incident. Under the old thinness test every one of these was "thin" and + force-attached to whatever was most recent. + """ + traffic = _overnight_traffic() + incidents, placement = _simulate(traffic) + dispatch_offsets = {off for off, _, tags, _ in traffic if tags} + dispatch_incidents = { + placement[f"call-{i}"] + for i, (off, _, tags, _) in enumerate(sorted(traffic)) + if tags and f"call-{i}" in placement + } + assert len(dispatch_incidents) == len(dispatch_offsets), ( + f"{len(dispatch_offsets)} jobs landed in {len(dispatch_incidents)} incident(s)" + ) + + +def test_a_long_run_of_pure_chatter_never_builds_an_incident(): + """ + Acknowledgements alone carry no content, so they cannot open an incident and + — with nothing recent to reply to — must not accrete into one either. + """ + incidents, _ = _simulate([(m, [], [], None) for m in range(0, 240, 6)]) + assert incidents == []