From 3246fc97b0ba6c69f87d11f611bb3df2addbdaee Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Tue, 27 Oct 2026 20:26:38 -0400 Subject: [PATCH] correlation: rules decide, one cheap LLM call only when ambiguous; drop the consensus tiebreak (server-26#299) The rules engine now commits every confident decision alone. _run_decision marks a decision ambiguous (new_with_candidates, unfit_tg_candidate, unverified_unit_link) and only then does upload._correlate make ONE llm_correlator.decide call; a usable answer is committed, otherwise the rules decision stands. Removed: _correlate_with_consensus, llm_correlator.tiebreak / decisions_agree / tiebreak prompt, corr_smart_model, the correlation_smart health tier, the LLM-orphan gate (_call_is_substanceless, effectively dead: it could only fire on minor-severity untyped calls) and its sweep budget. corr_consensus is now rules_only | llm | rules_fallback | rules_guard, with corr_band naming why; admin debug and replay metrics read the new values. Section fixes in the same pass: - #115: no tiebreaker left to manufacture incidents; an LLM orphan of a rules "new" is honoured only for untyped, substanceless, routine/minor calls. - #86: divergence veto bar moved to config (embedding_divergence_threshold); the band between it and the match bar is ambiguous, not a silent merge. - #103: a unit link with no embedding is "unit_overlap_unverified" -> ambiguous, instead of skipping the veto and linking as verified. - #189: unit-only and LLM links refused once an incident is open 90 min; unit links on incidents open >30 min are ambiguous; LLM links distance- guarded (incident_max_spread_km) and only to shown candidates. - #190: severity steps down on an explicit stand-down phrase; new severity_peak keeps the monotonic max. - #88: parse_llm_bool fails closed on "false"; scene closures honoured only on confident links or the incident the scene opened. - #27: sweep budget is the scan window (wall clock), not 3 attempts. - LLM prompt: "Reserve new" bullet no longer contradicts the #189 EXCEPT. - updated_at never moves backward on a back-dated link. Unmeasured: no live AI or replay (credits unfunded); band frequency, F1 and thresholds are unverified. Must be re-scored on both answer keys in Replay before merge. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/config.py | 55 +- drb-c2-core/app/internal/ai_health.py | 2 +- .../app/internal/chatter_classifier.py | 4 +- .../app/internal/incident_correlator.py | 499 ++++++++++++------ drb-c2-core/app/internal/llm_correlator.py | 209 +++----- .../app/internal/recorrelation_sweep.py | 52 +- drb-c2-core/app/internal/replay.py | 9 +- drb-c2-core/app/routers/admin.py | 34 +- drb-c2-core/app/routers/upload.py | 416 ++++++--------- drb-c2-core/tests/test_ai_flag_gating.py | 2 +- drb-c2-core/tests/test_ai_health.py | 6 +- drb-c2-core/tests/test_ai_timeouts.py | 25 +- drb-c2-core/tests/test_clearance_tracking.py | 7 +- drb-c2-core/tests/test_consensus_gate.py | 430 --------------- .../tests/test_correlation_decision_flow.py | 367 +++++++++++++ .../tests/test_correlator_location_path.py | 141 +++++ .../tests/test_correlator_merge_caps.py | 9 +- ...est_upload_short_transcript_correlation.py | 2 +- 18 files changed, 1207 insertions(+), 1062 deletions(-) delete mode 100644 drb-c2-core/tests/test_consensus_gate.py create mode 100644 drb-c2-core/tests/test_correlation_decision_flow.py create mode 100644 drb-c2-core/tests/test_correlator_location_path.py diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 922aeed..b796974 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -70,20 +70,16 @@ class Settings(BaseSettings): # Gemini (intelligence extraction, embeddings, incident summaries) gemini_api_key: Optional[str] = None - # Correlation consensus models - # corr_cheap_model — first-pass LLM correlator (runs on every call) - # corr_smart_model — tiebreaker (only fires when rules and cheap LLM disagree) - # Both IDs below were retired by Google and returned 404 on every call from - # some point before 2026-08-18 until they were corrected. Because a failed - # LLM call falls back to the rules decision, nothing broke loudly -- the - # entire LLM tier and the consensus tiebreak were simply dead in production - # while correlation behaviour was being tuned against rules-only output. - # Verify against https://ai.google.dev/gemini-api/docs/models before changing. + # Correlation LLM (server-26#299). The rules engine decides every confident + # case; this model is called at most ONCE per scene, and only when the rules + # flag the decision as ambiguous (incident_correlator._ambiguity). There is + # no second "smart" tier any more: the cheap/smart consensus tiebreak ran on + # ~15% of scenes and resolved nearly every disagreement as "new" (#115). + # The model ID was once retired by Google and 404'd on every call for days; + # a failed LLM call falls back to the rules decision, so nothing broke + # loudly. Verify against https://ai.google.dev/gemini-api/docs/models + # before changing. corr_cheap_model: str = "gemini-3.6-flash" # was gemini-2.0-flash (shut down) - # gemini-2.5-pro was closed to new projects by 2026-09 (every tiebreak 404'd - # in the first replay run, server-26#170); Google lists no stable Pro model, - # so the smart tier is the newest stable Flash instead. - corr_smart_model: str = "gemini-3.8-flash" # was gemini-2.5-pro, gemini-1.5-pro # Transcript correction (server-26#36). Runs inside transcription, once per # transcribed call above MIN_WORDS_FOR_CORRECTION, so it is priced like STT # rather than like the correlation tier — cheap model on purpose. @@ -122,6 +118,35 @@ class Settings(BaseSettings): embedding_similarity_threshold: float = 0.93 # slow-path: requires location corroboration embedding_no_location_threshold: float = 0.97 # slow-path: match without location (very high bar) embedding_cross_tg_threshold: float = 0.85 # cross-TG path: same dept + 2+ shared units + # server-26#86: the content-divergence veto on a unit-overlap link. This is + # a DIVERGENCE bar, not a match bar, so it sits deliberately below + # embedding_similarity_threshold: two calls about different jobs on one + # dispatch channel still share ten-codes, unit IDs and address phrasing, so + # "not a match" (< 0.93) is not evidence of "a different job". Below this + # bar the link is vetoed; between the two bars the link is neither + # confirmed nor refused, so it is AMBIGUOUS and goes to the LLM (#299) + # instead of silently merging. 0.82 is the historical literal, unmeasured + # against real score distributions. + embedding_divergence_threshold: float = 0.82 + # A unit-overlap link is only a confident one while the incident is fresh: + # idle under this many minutes (the old hard-coded 15 of the divergence + # veto) AND open under unit_link_confident_max_span_minutes. Past either, + # shared units alone are mechanism B (CORRELATION_REVIEW_0820.md) — patrol + # units work job after job on one channel — so the link is ambiguous. + unit_link_fresh_idle_minutes: int = 15 + unit_link_confident_max_span_minutes: int = 30 + # server-26#189: an incident open this long accepts no more WEAK links + # (unit overlap without location/content agreement, or an LLM link) — only + # vehicle, location or embedding evidence, until the hard + # incident_max_duration_minutes cap. The 09-27 "Rollover Entrapment" + # sink absorbed unrelated Ch 1 jobs for 2h34m this way. Thin replies are + # exempt: they never refresh updated_at, so they cannot keep a sink alive. + incident_weak_link_max_span_minutes: int = 90 + # server-26#189: an LLM link may not pull a call geocoded farther than + # this from a non-pursuit incident's verified pin (the same 4x + # location_proximity_km the slow path accepts). The LLM tier had no + # distance check at all, which is how one incident drew a 37 km path. + incident_max_spread_km: float = 2.0 location_proximity_km: float = 0.5 # radius for location-proximity matching geocode_max_km: float = 40.0 # reject geocode results farther than this from the node incident_auto_resolve_minutes: int = 90 # auto-resolve after N minutes with no new calls (major / unknown severity) @@ -147,9 +172,7 @@ class Settings(BaseSettings): # at 9.7 min. Every correct thin attach in that dump was <= 3.4 min idle # and every wrong 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. Also the escape hatch in routers/upload.py's LLM-orphan gate - # (_recent_incident_on_same_talkgroup, server-26#115) — check both call - # sites before retuning this. + # above this. tg_dispatch_thin_idle_minutes: int = 5 # ── Hard caps: an incident past either of these stops accepting calls ────── diff --git a/drb-c2-core/app/internal/ai_health.py b/drb-c2-core/app/internal/ai_health.py index 7279a5d..e37bc32 100644 --- a/drb-c2-core/app/internal/ai_health.py +++ b/drb-c2-core/app/internal/ai_health.py @@ -32,7 +32,7 @@ from typing import Optional from app.internal.logger import logger from app.config import settings -TIERS = ("transcription", "correlation_cheap", "correlation_smart", "extraction") +TIERS = ("transcription", "correlation_cheap", "extraction") # Consecutive failures a TRANSIENT condition must reach before it alerts. # Permanent conditions skip this entirely and alert on failure #1. diff --git a/drb-c2-core/app/internal/chatter_classifier.py b/drb-c2-core/app/internal/chatter_classifier.py index 81e244d..448f238 100644 --- a/drb-c2-core/app/internal/chatter_classifier.py +++ b/drb-c2-core/app/internal/chatter_classifier.py @@ -18,8 +18,8 @@ SHADOW MODE ONLY. As of this module's introduction, nothing skips scene extraction based on this verdict. `intelligence.extract_scenes` calls `classify_chatter` purely to record the verdict on the call doc (`chatter_classifier_verdict` / `chatter_classifier_reason`) so it becomes -observable in the next `/admin` correlation-debug dump, exactly like -`corr_gate_veto` (server-26#115 / PR #126). See the TODO at that call site for +observable in the next `/admin` correlation-debug dump (as `corr_gate_veto` +was, before server-26#299 removed that gate). See the TODO at that call site for what has to be true before this flips live. Precision over recall, deliberately. A false positive here — flagging a REAL diff --git a/drb-c2-core/app/internal/incident_correlator.py b/drb-c2-core/app/internal/incident_correlator.py index 2233517..17795e9 100644 --- a/drb-c2-core/app/internal/incident_correlator.py +++ b/drb-c2-core/app/internal/incident_correlator.py @@ -43,6 +43,13 @@ Matching priority (in order): Calls with no incident_type skip new-incident creation but still run paths 1–3, so unclassified calls (short transport end, "en route", etc.) can link to an existing incident via talkgroup match. + +Decision flow (server-26#299): the rules engine decides every confident case on +its own. When it flags its decision as ambiguous (`_ambiguity`: opening an +incident while live candidates exist, orphaning an unfit same-talkgroup +follow-up, or linking on unverified shared units), routers/upload.py makes ONE +cheap LLM call (llm_correlator.decide) and commits its answer instead. There is +no second LLM tier and no consensus vote. """ import math import re @@ -65,26 +72,65 @@ _SEVERITY_RANK = {"routine": 0, "minor": 1, "moderate": 2, "major": 3} def _max_severity(current: Optional[str], new: Optional[str]) -> str: """ - Highest of two severities on the four-level ladder — the merge rule used - every time a call attaches to an existing incident. - - Severity is monotonic by design: it only ever rises, never falls, as more - calls link. An incident briefly assessed "major" genuinely was major at - that moment; a later call that sounds calmer ("units clear", dispatcher - moving on) is evidence the SITUATION is winding down, not that the earlier - read was wrong. That's what `status`/`resolved_at` are for — resolution - retires an incident, it doesn't retroactively erase how serious it was. - Every severity-driven surface (worst-first incident rail, "Major only" - filter, map colour) exists to make sure a major event is never missed; - downgrading severity mid-incident would silently defeat that on the exact - incidents it exists to protect. A malformed/unrecognized value from either - side ranks as "routine" so it can never suppress a real escalation. + Highest of two severities on the four-level ladder. A malformed or + unrecognized value from either side ranks as "routine" so it can never + suppress a real escalation. """ current = current if current in _SEVERITY_RANK else "routine" new = new if new in _SEVERITY_RANK else "routine" return current if _SEVERITY_RANK[current] >= _SEVERITY_RANK[new] else new +# server-26#190 — explicit stand-down / unfounded dispositions. Deliberately a +# short list of phrases that only ever mean "there is less here than was +# reported"; a calmer-SOUNDING follow-up ("Engine 5 on scene", "copy") is not +# one, because extraction judges each transmission on its own words and a +# working fire's routine status traffic would otherwise talk it down to +# routine. Left out: "disregard"/"cancel" ("disregard my last" is a radio +# correction), "all clear" ("primary search all clear" on a working fire) and +# "code 4" (scene secure, not "nothing happened"). +_STAND_DOWN_RE = re.compile( + r"\b(?:unfounded|false alarm|no (?:fire|flames|smoke showing)|nothing (?:showing|found|evident)" + r"|gone on arrival|g\.?o\.?a\.?|stand(?:ing)? down|stood down" + r"|fire(?:'s| is)? out)\b", + re.IGNORECASE, +) + + +def _is_stand_down(transcript: Optional[str]) -> bool: + return bool(transcript and _STAND_DOWN_RE.search(transcript)) + + +def _severity_after_link( + inc: dict, call_severity: Optional[str], transcript: Optional[str], +) -> dict: + """ + The incident's severity fields after one more call links (server-26#190). + + `severity` is the CURRENT read and drives the live surfaces (worst-first + rail, map colour, severity-scaled quiet timer). It rises immediately on + any worse call, and falls only on an explicit stand-down disposition + ("they lit a fireplace... only smoke showing, no flames") — then to that + call's own severity. It used to only ever ratchet up, which left a false + Major on the wall for the full 90-minute Major quiet timer after units + reported nothing there. + + `severity_peak` keeps the old monotonic meaning (the worst this incident + was ever judged), so history and any "was this ever major" question still + have an answer that later chatter cannot erase. + """ + current = inc.get("severity") if inc.get("severity") in _SEVERITY_RANK else "routine" + peak = _max_severity(inc.get("severity_peak"), current) + new = call_severity if call_severity in _SEVERITY_RANK else "routine" + out = {"severity_peak": _max_severity(peak, new)} + if _SEVERITY_RANK[new] < _SEVERITY_RANK[current] and _is_stand_down(transcript): + out["severity"] = new + out["severity_deescalated_from"] = current + else: + out["severity"] = _max_severity(current, new) + return out + + # Maximum plausible ground speed for a moving incident (pursuit/transport). # ~3 miles/min ≈ 180 mph — well above real pursuit speeds, but generous enough # to tolerate GPS drift and call-timing jitter. Anything faster is a bad geocode @@ -278,16 +324,88 @@ def has_event_substance(ctx: dict) -> bool: a vehicle, a geocode, or a tag. This is the substance half of the incident-creation gate (see - `_run_decision`, "Severity, not type, decides..."), factored out so the - consensus LLM-orphan gate in routers/upload.py mirrors it exactly and can - never drop a call the creation gate would have opened. `call_units` and - `location` are deliberately excluded — radio protocol puts a unit ID and a - place name in almost every transmission, so counting them as substance - makes the check trivially true. + `resolve_new_incident_type`). `call_units` and `location` are deliberately + excluded — radio protocol puts a unit ID and a place name in almost every + transmission, so counting them as substance makes the check trivially true. """ return bool(ctx.get("call_vehicles") or ctx.get("coords") or ctx.get("tags")) +def resolve_new_incident_type(ctx: dict) -> Optional[str]: + """ + The type a NEW incident opened by this call gets, or None when the call is + not incident-worthy and must orphan instead. The one creation gate, shared + by the rules engine and the LLM tier (llm_correlator._parse_response), so + an LLM "new" can neither open what the rules gate refuses nor be demoted + to orphan for lacking a type the rules gate would have defaulted. + + Severity, not type, decides whether a call is incident-worthy. Requiring a + concrete incident_type meant a channel whose traffic never classifies — + transit/rail administration, records lookups, prisoner transports — could + never open a SECOND incident: every later call funnelled into whichever + incident was created first (2026-08-16: TG 9048, 28 calls / 49 min) + alongside 30+ permanent orphans. So anything the extractor judged a real + event (minor+), or that carries substance (`has_event_substance`), opens + an incident under the neutral "other" type. `call_units` and `location` + are NOT substance — the first version of this gate counted them, turned + "11-Victor, 72 at Holland Station" into its own incident, and left 37 of + 50 incidents open, one call each. + """ + resolved = ctx.get("incident_type") or _infer_type_from_tags(ctx.get("tags") or []) + if resolved: + return resolved + if ctx.get("call_severity") in ("minor", "moderate", "major") or has_event_substance(ctx): + return "other" + return None + + +def is_clearance_only(ctx: dict) -> bool: + """ + "45-9, I'm clear." carries one fact: which unit is done. Only the rules + engine's unit match can say which incident that is; an LLM link here would + apply the clear to whatever incident it picked, so such a call is never + sent to the LLM tier. + """ + units = ctx.get("call_units") or [] + cleared = ctx.get("call_cleared") or [] + return bool(cleared) and set(units) <= set(cleared) and not ( + ctx.get("tags") or ctx.get("location") or ctx.get("call_vehicles") or ctx.get("incident_type") + ) + + +_LLM_TRUE = frozenset({"true", "yes", "1"}) +_LLM_FALSE = frozenset({"false", "no", "0", "", "none", "null"}) + + +def parse_llm_bool(value) -> bool: + """ + A yes/no field from an LLM, failing CLOSED (server-26#88). + + `bool(value)` reads the JSON *string* "false" as True, so a model that + quoted its booleans would resolve whatever incident the scene linked into. + Only a real bool, or an exact true-ish token, counts as True; anything + else — "false", "maybe", a number, a dict — is False. For `resolved` and + `reassignment` a missed True costs a stale open incident or one missed + unit release; a false True closes a live incident or blanks a call's units. + + Public because the extraction boundary (intelligence.py) is where it + belongs; routers/upload.py applies it too, so a value that reaches the + scene loop un-coerced is still validated there. + """ + if isinstance(value, bool): + return value + if isinstance(value, str): + token = value.strip().lower() + if token in _LLM_TRUE: + return True + if token not in _LLM_FALSE: + logger.warning(f"parse_llm_bool: unrecognised value {value!r} read as False") + return False + if value is not None: + logger.warning(f"parse_llm_bool: non-boolean {type(value).__name__} {value!r} read as False") + return False + + def _infer_type_from_tags(tags: list[str]) -> Optional[str]: """Return an incident type inferred from tags, or None if ambiguous.""" for tag in tags: @@ -483,7 +601,7 @@ def _resolve_incident_title( of strictly higher severity. Rationale in the commit message; in short, an incident's identity is the event that opened it, and the one situation where the header must change is the one where things got worse. This - mirrors `_max_severity`: monotonic, never walked back by later chatter. + mirrors `severity_peak`: monotonic, never walked back by later chatter. Two non-renames are still allowed, because neither replaces an event name: • filling in a placeholder title on an incident that opened on a call @@ -656,6 +774,43 @@ def _incident_at_capacity(inc: dict, now: datetime) -> Optional[str]: return None +def _weak_link_capped(inc: dict, now: datetime) -> bool: + """server-26#189 — past this span an incident takes only strong evidence.""" + return _incident_span_minutes(inc, now) >= settings.incident_weak_link_max_span_minutes + + +def llm_candidates(ctx: dict) -> list[dict]: + """ + The incidents the LLM tier may see and link into: the rules engine's + `recent` pool minus anything at the hard caps or past the weak-link span + cap (server-26#189). Every LLM link is weak evidence — a model's reading + of two transcripts — so an incident that has been open long enough to + have become a catch-all is not offered to it at all. + """ + now = ctx["now"] + return [ + inc for inc in ctx.get("recent") or [] + if not _incident_at_capacity(inc, now) and not _weak_link_capped(inc, now) + ] + + +def llm_link_allowed(inc: dict, ctx: dict) -> bool: + """ + Distance guard on an LLM-chosen link (server-26#189). When both the call + and a non-pursuit incident are geocoded and farther apart than + `incident_max_spread_km`, the model's link is refused: every rules path + already has a location check of its own, and the LLM tier had none. + """ + coords = ctx.get("coords") + pin = _verified_pin(inc) + if not coords or not pin: + return True + if set(inc.get("tags") or []) & _PURSUIT_TAGS: + return True + dist = _haversine_km(coords["lat"], coords["lng"], pin["lat"], pin["lng"]) + return dist <= settings.incident_max_spread_km + + 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] = [] @@ -979,7 +1134,7 @@ def _run_decision(ctx: dict) -> dict: # 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. + # (tests) rather than through _build_context. _capped_ids = { inc.get("incident_id") for inc in (list(ctx["all_active"]) + list(ctx["recent"])) @@ -1013,6 +1168,10 @@ def _run_decision(ctx: dict) -> dict: matched_incident: Optional[dict] = None corr_debug: dict = {} + # Why the fast path refused a same-talkgroup candidate, when it did. A + # "no_signal" refusal is the one the rules cannot settle: a substantive + # call on a channel with a live incident and no evidence either way. + tg_reject: Optional[str] = None # ── 1. Fast path: talkgroup match with recency gate ────────────────────── # @@ -1168,13 +1327,14 @@ def _run_decision(ctx: dict) -> dict: "corr_incident_idle_min": round(_incident_idle_minutes(candidate, now), 1), "corr_fit_signal": fit_signal, } - if fit_signal == "unit_overlap" and call_units: + if fit_signal.startswith("unit_") and call_units: corr_debug["corr_matched_units"] = _matching_units(call_units, candidate.get("units")) logger.info( f"Correlator fast-path: call {call_id} → {candidate['incident_id']} " f"(signal={fit_signal})" ) else: + tg_reject = fit_signal logger.info( f"Correlator fast-path skipped: call {call_id} — {fit_signal} " f"from {candidate['incident_id']}; will attempt new incident" @@ -1207,13 +1367,14 @@ def _run_decision(ctx: dict) -> dict: "corr_candidates": len(tg_recent), "corr_fit_signal": fit_signal, } - if fit_signal == "unit_overlap" and call_units: + if fit_signal.startswith("unit_") and call_units: corr_debug["corr_matched_units"] = _matching_units(call_units, candidate.get("units")) logger.info( f"Correlator fast-path (disambig {len(tg_recent)} candidates): " f"call {call_id} → {candidate['incident_id']} (signal={fit_signal})" ) else: + tg_reject = fit_signal logger.info( f"Correlator fast-path disambig: no candidate fits call {call_id} " f"across {len(tg_recent)} incidents ({fit_signal}) — will attempt new incident" @@ -1249,28 +1410,27 @@ def _run_decision(ctx: dict) -> dict: and inc.get("updated_at", "") > best_unit_inc.get("updated_at", "") for inc in all_active ) - # Location conflict check: same unit at a clearly different location - # means the officer has moved to a new scene (e.g. dispatch sent - # them to a second call without an explicit reassignment signal). - location_conflict = False - if coords and not reassigned_away: - inc_coords_uc = best_unit_inc.get("location_coords") - if inc_coords_uc: - uc_dist = _haversine_km( - coords["lat"], coords["lng"], - inc_coords_uc["lat"], inc_coords_uc["lng"], - ) - if uc_dist > settings.location_proximity_km: - location_conflict = True - logger.info( - f"Correlator unit-continuity skipped: call {call_id} — " - f"unit match but location conflict ({uc_dist:.2f}km) with " - f"{best_unit_inc['incident_id']}" - ) - if not reassigned_away and not location_conflict: + # Same unit-overlap judgement as the fast path (_unit_fit): a + # location conflict means the officer has moved to a new scene, a + # sink past the weak-link span cap takes no unit-only links + # (server-26#189), and an old incident with no location or content + # agreement is linked but flagged unverified, i.e. ambiguous. + uc_fit, uc_signal = (False, "reassigned_away") + if not reassigned_away: + uc_fit, uc_signal = _unit_fit( + best_unit_inc, coords, call_embedding, now, + settings.location_proximity_km, + ) + if not uc_fit: + logger.info( + f"Correlator unit-continuity skipped: call {call_id} vs " + f"{best_unit_inc['incident_id']} — {uc_signal}" + ) + else: matched_incident = best_unit_inc corr_debug = { "corr_path": "unit-continuity", + "corr_fit_signal": uc_signal, "corr_incident_idle_min": round(_incident_idle_minutes(best_unit_inc, now), 1), # Unlike this file's other two paths (fast/single, fast/disambig), # a match here is unit-driven by construction (call_unit_set & @@ -1445,63 +1605,66 @@ def _run_decision(ctx: dict) -> dict: # ── Decision output ─────────────────────────────────────────────────────── if matched_incident: - return { - "action": "link", - "matched_incident": matched_incident, - "incident_type": incident_type, - "corr_debug": corr_debug, - } - - if not create_if_new: - return {"action": "orphan", "matched_incident": None, "incident_type": None, "corr_debug": corr_debug} - - # Attempt type inference from tags before giving up on creation - resolved_type = incident_type - if not resolved_type and tags: - resolved_type = _infer_type_from_tags(tags) - if resolved_type: + action, resolved_type = "link", incident_type + elif not create_if_new: + action, resolved_type = "orphan", None + else: + resolved_type = resolve_new_incident_type(ctx) + action = "new" if resolved_type else "orphan" + if resolved_type and resolved_type != incident_type: logger.info( - f"Correlator: inferred incident_type={resolved_type!r} from tags {tags} for call {call_id}" + f"Correlator: call {call_id} has no incident_type — opening a " + f"{resolved_type!r} incident (severity={call_severity}, tags={tags})" ) - # 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. - # - # `call_units` and `location` are NOT substance. Radio protocol puts a unit - # ID in essentially every transmission and a place name in most of them, so - # including them made has_substance true almost always and the severity check - # dead code — the first version of this gate turned "11-Victor, 72 at Holland - # Station" into its own incident and left 37 of 50 incidents open, one call - # each. A vehicle, a geocode, or a tag means the extractor found something - # beyond who was speaking and where they stood. - if not resolved_type: - has_substance = has_event_substance(ctx) - 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} - - return { - "action": "new", - "matched_incident": None, + decision = { + "action": action, + "matched_incident": matched_incident, "incident_type": resolved_type, "corr_debug": corr_debug, } + decision["ambiguous"] = _ambiguity(decision, tg_reject, ctx) + return decision + + +def _ambiguity(decision: dict, tg_reject: Optional[str], ctx: dict) -> Optional[str]: + """ + Why the rules decision is NOT confident enough to commit alone, or None + when it is (server-26#299). Only an ambiguous decision is sent to the LLM + tier, once; every other decision is the rules engine's alone. + + Ambiguous: + • "new_with_candidates" — the rules would OPEN an incident while live, + LLM-eligible incidents exist. Opening is the expensive mistake: this + system over-splits, and rules-only correlation opened ~1.8x the + incidents per hour of LLM-assisted (#170, 2026-10-04 control window) + — duplicates of a running job and typed radio housekeeping. + • "unfit_tg_candidate" — the rules would ORPHAN a substantive call that + has a live incident on its own talkgroup but no evidence for or + against it (fit test "no_signal"): an unfit follow-up. + • "unverified_unit_link" — the rules would LINK on shared units alone + into an incident that is no longer fresh, with no location or content + agreement (mechanism B; #103: the content veto has nothing to measure + when the scene has no embedding). Between the divergence and match + embedding bars (#86) a link lands here too. + + Never ambiguous: thin calls (nothing to reason about), clearance-only + calls (only the unit match can place a 10-8), link-only sweeps + (create_if_new=False — rules-only by design), or when the LLM would have + no candidates to choose from. + """ + if not ctx.get("create_if_new", True) or ctx.get("is_thin_call"): + return None + if is_clearance_only(ctx) or not llm_candidates(ctx): + return None + action = decision["action"] + if action == "new": + return "new_with_candidates" + if action == "orphan" and tg_reject == "no_signal": + return "unfit_tg_candidate" + if action == "link" and (decision["corr_debug"] or {}).get("corr_fit_signal") == "unit_overlap_unverified": + return "unverified_unit_link" + return None # ───────────────────────────────────────────────────────────────────────────── @@ -1514,7 +1677,7 @@ async def _apply_and_log(decision: dict, ctx: dict) -> Optional[str]: server-26#96: every scene of a multi-scene call reaches this function independently (upload.py's ``for scene_index, scene in enumerate(scenes):`` - loop → _correlate_with_consensus → preview_correlation/apply_correlation), + loop → _correlate → preview_correlation/apply_correlation), and every scene writes to the SAME call doc. Writing corr_debug only at the flat top level meant scene 2's write silently clobbered scene 1's corr_path/corr_consensus/etc — the call doc ended up describing a splice @@ -1553,18 +1716,14 @@ async def _apply_and_log(decision: dict, ctx: dict) -> Optional[str]: "incident_id": incident_id, "corr_debug": corr_debug, # server-26#139: this scene's OWN extracted incident_type/ - # severity, as read by _call_is_substanceless's ctx at - # decision time — not the call doc's flat top-level field, - # which is last-scene-wins (server-26#96) and was the reason - # #138's "type" veto couldn't be told apart from cross-scene - # contamination without re-guessing from a live dump. + # severity, as read from ctx at decision time — not the call + # doc's flat top-level field, which is last-scene-wins + # (server-26#96). # NOTE: unlike incident_type, call_severity is already # coerced to "routine" when extraction emitted nothing # (ctx build: `severity or "routine"`) — a scene reading # "routine" here doesn't distinguish "extraction said - # routine" from "extraction said nothing". Don't split a - # severity veto the way #138 splits the type veto without - # accounting for that. + # routine" from "extraction said nothing". "incident_type": ctx.get("incident_type"), "severity": ctx.get("call_severity"), } @@ -1613,6 +1772,7 @@ async def _apply_decision(decision: dict, ctx: dict) -> Optional[str]: talkgroup_name=talkgroup_name, incident_type=incident_type, cleared_units=call_cleared, refresh_activity=not thin_link, call_severity=call_severity, call_srcaddr=call_srcaddr, + call_transcript=ctx.get("scene_transcript"), ) return matched_incident["incident_id"] @@ -1798,6 +1958,65 @@ def _disambiguate( return best +def _unit_fit( + inc: dict, + call_coords: Optional[dict], + call_embedding: Optional[list], + now: Optional[datetime], + proximity_km: float, +) -> tuple[bool, str]: + """ + Judge a link whose evidence is a shared unit. Same officer usually means + same call — but on a shared dispatch channel a patrol unit works job after + job, so shared units alone grow chains (mechanism B, + CORRELATION_REVIEW_0820.md). In order: + + • call and incident both geocoded: far apart → (False, + "unit_loc_conflict"), the officer moved to a new scene; close → + (True, "unit_location_match"). + • embeddings agree at the match bar → (True, "unit_content_match"). + • incident past the weak-link span cap (#189) → (False, + "weak_link_span_cap"): a long-running incident stops taking + unit-only evidence. + • incident fresh (idle < unit_link_fresh_idle_minutes) and young (open + < unit_link_confident_max_span_minutes) → (True, "unit_overlap"). No + content veto here: an update on the same scene often shares little + vocabulary with the dispatch and does not restate the address. + • not fresh, embeddings below the DIVERGENCE bar (#86) → (False, + "content_divergence"). + • anything else → (True, "unit_overlap_unverified"): the rules keep the + link but cannot vouch for it, so the decision is ambiguous and goes + to the LLM tier. This is where a scene with no embedding lands + (#103) — it used to skip the veto and link as if verified, which + removed the only brake on the unit loop for ~93% of calls. + """ + inc_coords = inc.get("location_coords") + if call_coords and inc_coords: + dist_km = _haversine_km( + call_coords["lat"], call_coords["lng"], + inc_coords["lat"], inc_coords["lng"], + ) + return (False, "unit_loc_conflict") if dist_km > proximity_km else (True, "unit_location_match") + + sim: Optional[float] = None + inc_emb = inc.get("embedding") + if call_embedding and inc_emb: + sim = _cosine_similarity(call_embedding, inc_emb) + if sim >= settings.embedding_similarity_threshold: + return True, "unit_content_match" + + if now is None: + return True, "unit_overlap_unverified" + if _weak_link_capped(inc, now): + return False, "weak_link_span_cap" + fresh = _idle_gate_minutes(inc, now) < settings.unit_link_fresh_idle_minutes + if fresh and _incident_span_minutes(inc, now) < settings.unit_link_confident_max_span_minutes: + return True, "unit_overlap" + if not fresh and sim is not None and sim < settings.embedding_divergence_threshold: + return False, "content_divergence" + return True, "unit_overlap_unverified" + + def _call_fits_incident( inc: dict, call_units: list[str], @@ -1812,18 +2031,14 @@ def _call_fits_incident( Return (fits, signal) — fits is True when this call plausibly belongs to the incident; signal names the specific evidence that drove the decision. - fits=True signals: "unit_overlap" | "vehicle_overlap" | "location_proximity" + fits=True signals: see `_unit_fit` for the unit_* family | "vehicle_overlap" + | "location_proximity" fits=False signals: "unit_loc_conflict" | "content_divergence" - | "location_conflict" | "no_signal" + | "weak_link_span_cap" | "location_conflict" | "no_signal" Evaluation order: - 1. Unit overlap. Same officer = same call. Also runs a location-conflict - guard: geocoded on both sides and clearly different → the officer has - moved to a new scene, false positive. No geocode on the call (or on - the incident, with a call geocode) AND the incident is old (≥ 15 min) - → content divergence as a location proxy (embedding similarity < 0.82 - → different scene). Skipped for recent incidents (< 15 min) — an - update without re-stating the address is normal. + 1. Unit overlap — judged by `_unit_fit`, which also decides whether the + link is confident or only "unit_overlap_unverified" (ambiguous). 2. Vehicle overlap → True. 3. Location proximity. Both geocoded and close → True; far apart with no other positive signal → False. @@ -1850,44 +2065,9 @@ def _call_fits_incident( inc_units = _unit_keys(inc.get("units")) matched_units = _matching_units(call_units, inc.get("units")) if matched_units: - if call_coords: - # Hard location conflict: geocoded on both sides and clearly different. - inc_coords_u = inc.get("location_coords") - if inc_coords_u: - dist_km = _haversine_km( - call_coords["lat"], call_coords["lng"], - inc_coords_u["lat"], inc_coords_u["lng"], - ) - if dist_km > proximity_km: - logger.info(f" fits[{inc_id}]: unit_overlap({matched_units}) but location_conflict dist={dist_km:.2f}km → unit_loc_conflict") - return False, "unit_loc_conflict" - elif call_embedding and idle_min >= 15: - # Call has geocode but incident doesn't — fall back to content - # divergence as a location proxy. Without this, stale incidents - # that never geocoded absorb unrelated calls purely on unit - # overlap (e.g. a patrol officer working a second scene 70 min - # after the original call). - inc_emb_u = inc.get("embedding") - if inc_emb_u: - sim = _cosine_similarity(call_embedding, inc_emb_u) - if sim < 0.82: - logger.info(f" fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence (has_call_coords/no_inc_coords) sim={sim:.3f} → content_divergence") - return False, "content_divergence" - elif call_embedding and idle_min >= 15: - # No geocode available AND old incident: use content divergence as a - # location-proxy veto. After 15+ minutes an officer at a completely - # different scene will be discussing clearly different content. - # Skip this for recent incidents — an officer updating on the same - # scene without re-stating the address is normal and their update - # won't share much vocabulary with the original dispatch. - inc_emb_u = inc.get("embedding") - if inc_emb_u: - sim = _cosine_similarity(call_embedding, inc_emb_u) - if sim < 0.82: - logger.info(f" fits[{inc_id}]: unit_overlap({matched_units}) but content_divergence sim={sim:.3f} → content_divergence") - return False, "content_divergence" - logger.info(f" fits[{inc_id}]: unit_overlap matched={matched_units} → unit_overlap") - return True, "unit_overlap" + fit, signal = _unit_fit(inc, call_coords, call_embedding, now, proximity_km) + logger.info(f" fits[{inc_id}]: unit_overlap matched={matched_units} → {signal}") + return fit, signal # ── 2. Vehicle overlap ──────────────────────────────────────────────────── inc_vehicles = set(inc.get("vehicles") or []) @@ -2047,6 +2227,7 @@ async def _update_incident( refresh_activity: bool = True, call_severity: Optional[str] = None, call_srcaddr: Optional[str] = None, + call_transcript: Optional[str] = None, ) -> None: incident_id = inc["incident_id"] @@ -2109,7 +2290,9 @@ async def _update_incident( "units_cleared": units_cleared, "location_mentions": location_mentions, "summary_stale": True, - "severity": _max_severity(inc.get("severity"), call_severity), + # Current severity can step DOWN on an explicit stand-down; the peak + # never does (server-26#190, _severity_after_link). + **_severity_after_link(inc, call_severity, call_transcript), # Always all three, always together — writing one without the others is # what let the label and the pin drift apart (server-26#23). **location_fields, @@ -2123,7 +2306,18 @@ async def _update_incident( # 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"] = _floor_at_started_at(inc, now).isoformat() + # Never move updated_at BACKWARD: a back-dated `now` (sweep, or a late + # upload's clock pin, server-26#301) would make a live incident look + # idle, failing on-time calls' recency gates and timer-closing it early. + when = _floor_at_started_at(inc, now) + try: + prev = datetime.fromisoformat(str(inc.get("updated_at") or "").replace("Z", "+00:00")) + if prev.tzinfo is None: + prev = prev.replace(tzinfo=timezone.utc) + when = max(when, prev) + except ValueError: + pass + updates["updated_at"] = when.isoformat() if is_new_call: # a second scene of the same call is not a second call updates["substantive_call_count"] = ( inc.get("substantive_call_count") @@ -2233,6 +2427,7 @@ async def _create_incident( "vehicles": call_vehicles, "srcaddrs": [call_srcaddr] if call_srcaddr else [], "severity": call_severity, + "severity_peak": call_severity, "summary": None, "summary_stale": True, "summary_last_run": None, diff --git a/drb-c2-core/app/internal/llm_correlator.py b/drb-c2-core/app/internal/llm_correlator.py index 2b3e0be..ce5be1f 100644 --- a/drb-c2-core/app/internal/llm_correlator.py +++ b/drb-c2-core/app/internal/llm_correlator.py @@ -1,23 +1,23 @@ """ -LLM-based incident correlator using Gemini. +LLM incident correlator (Gemini, settings.corr_cheap_model). -Two functions are exposed: - decide(call_id, ctx) — cheap first-pass (corr_cheap_model) - tiebreak(rules_decision, llm_decision, ctx) — smart tiebreaker (corr_smart_model) +One function is exposed: + decide(call_id, ctx) — called by routers/upload.py ONLY when the rules + engine marked its decision ambiguous + (incident_correlator._ambiguity), at most once per + scene. server-26#299 removed the second "smart" + tiebreak tier and the cheap/smart consensus vote. -Both return a decision dict compatible with _run_decision() in incident_correlator: +It returns a decision dict compatible with incident_correlator._apply_decision: {"action": "link"|"new"|"orphan", "matched_incident": dict|None, "incident_type": str|None, "corr_debug": dict, - "reasoning": str} ← extra field for logging/tiebreak comparison + "reasoning": str} -decide() is skipped for thin calls (no content to reason about) and when -GEMINI_API_KEY is not set — in those cases returns None so the caller knows -to fall back to the rules decision. - -Error handling: any Gemini failure returns None from decide() and the -rules_decision from tiebreak() so the pipeline never stalls. +or None — no API key, nothing to choose from, an answer that fails +validation, or any Gemini failure — and the caller then commits the rules +decision unchanged, so the pipeline never stalls on the LLM. """ import asyncio from datetime import datetime, timezone @@ -155,8 +155,9 @@ call that reports a new event, below.): same one — especially once that incident has been open a long time (see open:/calls:). - Reserve "new" for a call that clearly describes a DIFFERENT event from every - recent incident — a different place, different units, and a different subject, - not merely a different transmission about the same job. + recent incident — a different place or subject, not merely a different + transmission about the same job. Shared units alone do not make it the same + event (see EXCEPT above). - "orphan" a call that is not an incident at all: radio checks, roll call, a unit marking on/off duty or 10-8/10-98, mileage/log entries, a bare acknowledgement. Do not open a "new" incident for these. @@ -166,47 +167,15 @@ call that reports a new event, below.): """ -def _build_decide_prompt(ctx: dict) -> str: - now = ctx["now"] - recent = ctx["recent"] - inc_block = ( - "\n".join(_inc_summary(inc, now) for inc in _prompt_incidents(recent)) - if recent else "(none)" - ) +def _build_decide_prompt(ctx: dict, candidates: list[dict]) -> str: + now = ctx["now"] + inc_block = "\n".join(_inc_summary(inc, now) for inc in _prompt_incidents(candidates)) return ( "You are an incident correlator for a public safety radio monitoring system.\n\n" "A new radio call has arrived. Decide whether it belongs to an existing active incident, " - "represents a new incident, or should be orphaned (not enough information).\n\n" + "represents a new incident, or should be orphaned (not an incident at all).\n\n" f"NEW CALL:\n{_call_block(ctx)}\n\n" - f"ACTIVE INCIDENTS ({len(recent)} recent):\n{inc_block}\n" - f"{_RULES}\n" - f"Respond with JSON only (no markdown):\n{_SCHEMA}" - ) - - -def _build_tiebreak_prompt(rules_decision: dict, llm_decision: dict, ctx: dict) -> str: - now = ctx["now"] - recent = ctx["recent"] - inc_block = ( - "\n".join(_inc_summary(inc, now) for inc in _prompt_incidents(recent)) - if recent else "(none)" - ) - - def _fmt(d: dict, name: str) -> str: - action = d.get("action", "?") - inc = d.get("matched_incident") - inc_id = inc["incident_id"] if inc else (d.get("incident_id") or "null") - reason = d.get("reasoning") or (d.get("corr_debug") or {}).get("corr_fit_signal") or "—" - return f" {name}: action={action}, incident_id={inc_id}, reasoning={reason!r}" - - return ( - "You are a senior incident correlator for a public safety radio monitoring system.\n\n" - "Two correlation engines disagree. You must make the final decision.\n\n" - f"NEW CALL:\n{_call_block(ctx)}\n\n" - f"ACTIVE INCIDENTS ({len(recent)} recent):\n{inc_block}\n\n" - "DISAGREEMENT:\n" - f"{_fmt(rules_decision, 'Rules engine')}\n" - f"{_fmt(llm_decision, 'LLM correlator')}\n" + f"ACTIVE INCIDENTS ({len(candidates)} recent):\n{inc_block}\n" f"{_RULES}\n" f"Respond with JSON only (no markdown):\n{_SCHEMA}" ) @@ -225,44 +194,48 @@ def _sync_gemini(model_name: str, prompt: str) -> dict: # Decision parsing # ───────────────────────────────────────────────────────────────────────────── -def _parse_response(raw: dict, ctx: dict) -> dict: +def _parse_response(raw: dict, ctx: dict, candidates: list[dict]) -> Optional[dict]: """ - Convert raw Gemini JSON output to a decision dict compatible with _run_decision(). - Resolves incident_id → full incident doc from ctx["all_active"]. - Handles type inference for "new" actions the same way as the rules engine. - """ - from app.internal.incident_correlator import _infer_type_from_tags # same-package import + Convert raw Gemini JSON to a decision dict, or None when the answer is + unusable and the rules decision should stand. - action = raw.get("action", "orphan") + A "link" must name one of the candidates the model was shown — not any + active incident: the shown list is the one with the #189 weak-link cap + applied — and pass the distance guard (incident_correlator. + llm_link_allowed). A "new" takes its type from the same creation gate the + rules use, so it is refused (orphan) exactly when the rules would refuse + it. An unknown action is unusable, not an orphan: the old "default to + orphan" silently dropped calls on a malformed reply. + """ + from app.internal.incident_correlator import llm_link_allowed, resolve_new_incident_type + + action = raw.get("action") reasoning = raw.get("reasoning", "") - if action not in ("link", "new", "orphan"): - action = "orphan" + logger.warning(f"LLM correlator: unusable action {action!r} — keeping the rules decision") + return None matched_incident: Optional[dict] = None - + incident_type: Optional[str] = None if action == "link": inc_id = raw.get("incident_id") - if inc_id: - matched_incident = next( - (i for i in ctx["all_active"] if i.get("incident_id") == inc_id), - None, - ) + matched_incident = next((i for i in candidates if i.get("incident_id") == inc_id), None) if not matched_incident: logger.warning( - f"LLM correlator: incident_id={inc_id!r} not in active incidents — orphaning" + f"LLM correlator: incident_id={inc_id!r} is not a shown candidate — keeping the rules decision" ) - action = "orphan" - - incident_type: Optional[str] = None - if action in ("link", "new"): + return None + if not llm_link_allowed(matched_incident, ctx): + logger.info( + f"LLM correlator: link to {inc_id} refused — call is geocoded beyond " + f"incident_max_spread_km of its pin (server-26#189)" + ) + return None incident_type = ctx["incident_type"] + elif action == "new": + incident_type = resolve_new_incident_type(ctx) if not incident_type: - incident_type = _infer_type_from_tags(ctx["tags"]) - if action == "new" and not incident_type: - # Can't create an incident without a type — demote to orphan action = "orphan" - matched_incident = None return { "action": action, @@ -279,54 +252,34 @@ def _parse_response(raw: dict, ctx: dict) -> dict: async def decide(call_id: str, ctx: dict) -> Optional[dict]: """ - Run the cheap LLM correlator (corr_cheap_model) on the call. - - Returns a decision dict or None if: - - GEMINI_API_KEY is not configured - - the call is thin (content-free — no value from LLM) - - there are no recent active incidents to reason about - - Gemini fails - - Callers should treat None as "fall back to rules decision". + One cheap-model correlation decision for an ambiguous scene. Returns None + (keep the rules decision) when GEMINI_API_KEY is unset, there is no + candidate to show, the answer fails validation, or Gemini fails. """ + from app.internal.incident_correlator import llm_candidates + if not settings.gemini_api_key: return None - - if ctx["is_thin_call"]: - return None # thin calls have no transcript/units/coords to reason about - - if _is_clearance_only(ctx): - # "45-9, I'm clear." carries one fact: which unit is done. Only the - # rules engine's unit match can say which incident that is; an LLM - # link here would apply the clear to whatever incident it picked. + candidates = llm_candidates(ctx) + if not candidates: return None - if not ctx["recent"]: - return None # no incidents to correlate against — rules handles new-only - try: - prompt = _build_decide_prompt(ctx) + prompt = _build_decide_prompt(ctx, candidates) raw = await asyncio.to_thread(_sync_gemini, settings.corr_cheap_model, prompt) - decision = _parse_response(raw, ctx) + except Exception as e: + await _log_llm_failure("LLM correlator", "correlation_cheap", call_id, settings.corr_cheap_model, e) + return None + await ai_health.report_healthy("correlation_cheap") + decision = _parse_response(raw if isinstance(raw, dict) else {}, ctx, candidates) + if decision: _id = (decision["matched_incident"] or {}).get("incident_id", "null") logger.info( f"LLM correlator ({settings.corr_cheap_model}): call {call_id} → " f"action={decision['action']} incident={_id} " f"reasoning={decision['reasoning']!r}" ) - await ai_health.report_healthy("correlation_cheap") - return decision - except Exception as e: - await _log_llm_failure("LLM correlator", "correlation_cheap", call_id, settings.corr_cheap_model, e) - return None - - -def _is_clearance_only(ctx: dict) -> bool: - units = ctx.get("call_units") or [] - cleared = ctx.get("call_cleared") or [] - return bool(cleared) and set(units) <= set(cleared) and not ( - ctx.get("tags") or ctx.get("location") or ctx.get("call_vehicles") or ctx.get("incident_type") - ) + return decision _dead_models: set[str] = set() @@ -350,7 +303,7 @@ async def _log_llm_failure(where: str, tier: str, call_id: str, model: str, exc: if kind == "dead_model": await _log_tier_down(where, tier, model, "model is unavailable", - "Update CORR_CHEAP_MODEL/CORR_SMART_MODEL in config.py", text) + "Update CORR_CHEAP_MODEL in config.py", text) return # A depleted balance reads as 429, the same status as an ordinary rate limit, @@ -379,37 +332,3 @@ async def _log_tier_down(where: str, tier: str, model: str, problem: str, fix: s f"{where}: {problem} ({model!r}) -- the LLM correlation tier is DISABLED " f"and every call is falling back to rules-only. {fix}. API said: {text}" ) - - -async def tiebreak(rules_decision: dict, llm_decision: dict, ctx: dict) -> dict: - """ - Run the smart tiebreaker (corr_smart_model) when rules and LLM disagree. - Falls back to rules_decision on any error. - """ - call_id = ctx["call_id"] - try: - prompt = _build_tiebreak_prompt(rules_decision, llm_decision, ctx) - raw = await asyncio.to_thread(_sync_gemini, settings.corr_smart_model, prompt) - decision = _parse_response(raw, ctx) - _id = (decision["matched_incident"] or {}).get("incident_id", "null") - logger.info( - f"LLM tiebreak ({settings.corr_smart_model}): call {call_id} → " - f"action={decision['action']} incident={_id} " - f"reasoning={decision['reasoning']!r}" - ) - await ai_health.report_healthy("correlation_smart") - return decision - except Exception as e: - await _log_llm_failure("LLM tiebreak", "correlation_smart", call_id, settings.corr_smart_model, e) - return rules_decision - - -def decisions_agree(rules: dict, llm: dict) -> bool: - """True if both decisions agree on action and (when action=="link") on the target incident.""" - if rules["action"] != llm["action"]: - return False - if rules["action"] == "link": - r_id = (rules.get("matched_incident") or {}).get("incident_id") - l_id = (llm.get("matched_incident") or {}).get("incident_id") - return r_id == l_id - return True diff --git a/drb-c2-core/app/internal/recorrelation_sweep.py b/drb-c2-core/app/internal/recorrelation_sweep.py index a4d9fed..c01d686 100644 --- a/drb-c2-core/app/internal/recorrelation_sweep.py +++ b/drb-c2-core/app/internal/recorrelation_sweep.py @@ -72,7 +72,7 @@ def pipeline_worst_case_seconds(n_scenes: int = _BOUND_SCENES) -> float: correction = settings.gemini_timeout_seconds extraction = openai_attempts * settings.openai_timeout_seconds embeddings = n_scenes * openai_attempts * settings.openai_timeout_seconds - correlation = 2 * settings.gemini_timeout_seconds # decide + tiebreak, scene 0 + correlation = settings.gemini_timeout_seconds # at most one LLM call, scene 0 (#299) return stt + correction + extraction + embeddings + correlation + _NON_AI_SLACK_SECONDS @@ -83,21 +83,25 @@ def _derive_min_minutes() -> int: # 21 with the defaults (stt 120s, openai 60s x2 attempts, gemini 60s, 3 scenes). MIN_MINUTES_SINCE_PIPELINE_START = _derive_min_minutes() -# Standard link-only retry budget before a call is tombstoned corr_path="unlinked". -MAX_SWEEP_ATTEMPTS = 3 -# server-26#115 — a call the consensus LLM-orphan gate parked (llm=orphan vs -# rules=new, no substance) gets a longer budget. The gate fires before any -# incident for the job may exist, so the substantive call that would justify -# linking can land well after the standard ~6 min. Still link-only: a genuinely -# thin call must not mint an incident, and the rules creation gate would re-orphan -# it anyway. -GATED_ORPHAN_SWEEP_ATTEMPTS = 10 +def _last_chance(call: dict, now: datetime) -> bool: + """ + server-26#27 — the retry budget is wall-clock, not an attempt count. -def _max_sweep_attempts(call: dict) -> int: - if call.get("corr_consensus") == "llm_orphan_gate": - return GATED_ORPHAN_SWEEP_ATTEMPTS - return MAX_SWEEP_ATTEMPTS + The old budget was 3 attempts, which with the #131 hold-back expired ~19 + minutes after the pipeline started, while the incident the orphan belongs + to may be opened later than that (`f321b047` was tombstoned three minutes + after its own incident was created). An orphan is now retried on every + pass for as long as the scan query can see it, and tombstoned + corr_path="unlinked" only on its LAST visible pass: the one after which + its ended_at falls out of `recorrelation_scan_minutes`. A call with no + parseable ended_at gets no further retries. + """ + ended = _parse_dt(call.get("ended_at")) + if ended is None: + return True + remaining = timedelta(minutes=settings.recorrelation_scan_minutes) - (now - ended) + return remaining <= timedelta(minutes=settings.summary_interval_minutes) async def recorrelation_loop() -> None: @@ -149,9 +153,8 @@ async def _run_sweep_pass() -> None: ("status", "==", "ended"), ("ended_at", ">=", cutoff), ]) - # corr_path="unlinked" is written after the attempt budget is exhausted. - # Allows a few retries so a welfare-check call can link to an escalation - # incident that is created a few minutes later, without sweeping 30× forever. + # corr_path="unlinked" is the tombstone written on an orphan's last + # visible pass (_last_chance); the scan window itself is the budget. orphans = [ c for c in recent_ended if not c.get("incident_ids") and not c.get("incident_id") @@ -163,7 +166,6 @@ async def _run_sweep_pass() -> None: # 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(c) and not _pipeline_likely_still_running(c, now) ] @@ -173,14 +175,14 @@ async def _run_sweep_pass() -> None: logger.info(f"Re-correlation sweep: {len(orphans)} orphaned call(s) to check") linked = 0 for call in orphans: - if await _recorrelate_orphan(call): + if await _recorrelate_orphan(call, now): linked += 1 if linked: logger.info(f"Re-correlation sweep: linked {linked}/{len(orphans)} orphaned call(s)") -async def _recorrelate_orphan(call: dict) -> bool: +async def _recorrelate_orphan(call: dict, now: datetime) -> bool: """ Attempt to link a single orphaned call to an existing incident. Returns True if a match was found and the call was linked. @@ -223,12 +225,10 @@ async def _recorrelate_orphan(call: dict) -> bool: ) return True - # Increment the attempt counter. Once the budget is reached the orphan filter - # above will stop picking this call up, and we write corr_path="unlinked" as - # a permanent tombstone. - attempts = call.get("corr_sweep_count", 0) + 1 - update: dict = {"corr_sweep_count": attempts} - if attempts >= _max_sweep_attempts(call): + # The attempt count is kept for the admin debug view only; the budget is + # the scan window (_last_chance), after which the call is tombstoned. + update: dict = {"corr_sweep_count": call.get("corr_sweep_count", 0) + 1} + if _last_chance(call, now): update["corr_path"] = "unlinked" await fstore.doc_set("calls", call_id, update) return False diff --git a/drb-c2-core/app/internal/replay.py b/drb-c2-core/app/internal/replay.py index ee7b3a8..36c4c4f 100644 --- a/drb-c2-core/app/internal/replay.py +++ b/drb-c2-core/app/internal/replay.py @@ -60,7 +60,8 @@ USD_WHISPER_PER_MIN = 0.006 USD_PER_EXTRACTION = 0.0005 # gpt-4o-mini scene extraction + embedding USD_PER_CORRECTION = 0.0003 # Gemini flash transcript correction USD_PER_GEOCODE = 0.005 # Google geocode, roughly one per located scene -USD_PER_LLM_CORRELATE = 0.0005 # Gemini flash consensus decision +USD_PER_LLM_CORRELATE = 0.0005 # one Gemini flash correlation decision (ambiguous scenes only) +_LLM_CONSENSUS = ("llm", "rules_guard", "agreed", "tiebreak", "llm_orphan_gate") # Fields the pipeline writes onto a call doc. Stripped when a call is copied # into the sandbox so the replay recomputes them instead of inheriting the @@ -636,7 +637,11 @@ def compute_metrics(incidents: list[dict], calls: list[dict]) -> dict: corr_path[r.get("corr_path") or "none"] += 1 consensus[r.get("corr_consensus") or "none"] += 1 linked = sum(1 for c in calls if c.get("incident_ids")) - llm = sum(n for k, n in consensus.items() if k not in ("none", "rules_only")) + # Scene decisions that made an LLM call (server-26#299: at most one each). + # "rules_fallback" is excluded — the call failed or was unusable, so it + # was mostly not billed. "agreed"/"tiebreak" are pre-#299 runs' values + # (a tiebreak was two calls, so old runs under-count by their tiebreaks). + llm = sum(n for k, n in consensus.items() if k in _LLM_CONSENSUS) return { "calls": len(calls), "calls_linked": linked, diff --git a/drb-c2-core/app/routers/admin.py b/drb-c2-core/app/routers/admin.py index af7eac2..37ba5c3 100644 --- a/drb-c2-core/app/routers/admin.py +++ b/drb-c2-core/app/routers/admin.py @@ -109,9 +109,9 @@ async def debug_correlation( "scene_index": scene_index, "transcript": scene.get("transcript"), "incident_id": scene.get("incident_id"), - # server-26#139: this scene's OWN incident_type/severity, as seen - # by _call_is_substanceless at decision time — not the call doc's - # flat top-level field, which is last-scene-wins (server-26#96). + # server-26#139: this scene's OWN incident_type/severity at decision + # time — not the call doc's flat top-level field, which is + # last-scene-wins (server-26#96). "incident_type": scene.get("incident_type"), "severity": scene.get("severity"), "corr_path": corr_debug.get("corr_path"), @@ -126,7 +126,7 @@ async def debug_correlation( "corr_llm_reasoning": corr_debug.get("corr_llm_reasoning"), "corr_llm_action": corr_debug.get("corr_llm_action"), "corr_rules_action": corr_debug.get("corr_rules_action"), - "corr_gate_veto": corr_debug.get("corr_gate_veto"), + "corr_band": corr_debug.get("corr_band"), } def _call_summary(call: dict) -> dict: @@ -173,20 +173,18 @@ async def debug_correlation( "corr_matched_units": call.get("corr_matched_units"), "corr_sweep_count": call.get("corr_sweep_count"), "skip_reason": call.get("skip_reason"), - # LLM consensus tier fields — written by upload.py's - # _correlate_with_consensus / llm_correlator.py, but previously - # dropped here, making it impossible to tell from this endpoint - # whether the LLM correlation tier is actually running (server-26#24). + # LLM tier fields — written by upload.py's _correlate / + # llm_correlator.py, but previously dropped here, making it + # impossible to tell from this endpoint whether the LLM + # correlation tier is actually running (server-26#24). "corr_consensus": call.get("corr_consensus"), "corr_llm_reasoning": call.get("corr_llm_reasoning"), "corr_llm_action": call.get("corr_llm_action"), "corr_rules_action": call.get("corr_rules_action"), - # server-26#115 — why an llm=orphan/rules=new disagreement escalated - # to tiebreak instead of being gated (see upload.py's - # _call_is_substanceless). Present only on that disagreement shape; - # written here specifically so a live measurement window can read - # the reason instead of reconstructing it by hand from the dump. - "corr_gate_veto": call.get("corr_gate_veto"), + # server-26#299 — why the rules marked this decision ambiguous and + # sent it to the LLM (incident_correlator._ambiguity). Absent on a + # confident rules decision. + "corr_band": call.get("corr_band"), # server-26#127 — shadow-mode upstream chatter classifier verdict. # Written by intelligence.extract_scenes on every transcript that # reaches real scene extraction (not on garbage/too-short skips). @@ -298,7 +296,8 @@ async def debug_correlation( orphans_by_tg[tg_key]["count"] += 1 if not o.get("incident_type") and not o.get("tags"): orphans_by_tg[tg_key]["no_type_count"] += 1 - if (o.get("corr_sweep_count") or 0) >= 3: + # Tombstoned by the sweep's wall-clock budget (server-26#27). + if o.get("corr_path") == "unlinked": orphans_by_tg[tg_key]["sweep_exhausted_count"] += 1 # ── Summary ─────────────────────────────────────────────────────────────── @@ -374,9 +373,8 @@ async def debug_correlation( "corr_fit_signal": _tally(e.get("corr_fit_signal") for e in scene_entries), "corr_consensus": _tally(e.get("corr_consensus") for e in scene_entries), "corr_llm_action": _tally(e.get("corr_llm_action") for e in scene_entries), - # server-26#115 — this IS the number the escape-hatch fix exists to - # produce: why each llm=orphan/rules=new call escaped the gate. - "corr_gate_veto": _tally(e.get("corr_gate_veto") for e in scene_entries), + # server-26#299 — how often, and why, the LLM tier was consulted. + "corr_band": _tally(e.get("corr_band") for e in scene_entries), # server-26#127 — shadow-mode chatter classifier. The target # population is non-events, which land as orphans or single-call # incidents, NOT as a slice of every linked call -- tally `orphans` diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index 43b72b5..965a15a 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/drb-c2-core/app/routers/upload.py @@ -112,99 +112,56 @@ async def upload_call_audio( return {"url": gcs_uri} -# server-26#115 — the consensus LLM-orphan gate only fires when the call is -# genuinely substanceless. The earlier version tested `rules_decision["corr_debug"]` -# for a "positive signal", but corr_debug is EMPTY at preview time for -# action=="new" (corr_path:"new" is written at APPLY time), so that test was -# always False and the gate dropped real events — a major "extinguishing fire", -# geocoded calls, pursuit updates. The substance test now runs against `ctx`, -# which is fully populated at preview time. +# New-event signals: an LLM "orphan" may not drop a call carrying one when the +# rules would open an incident for it (server-26#299). A missed real event is +# worse than one extra incident, and the sweep is link-only, so a dropped event +# never comes back. These are the old #115 gate's vetoes, which existed because +# a looser gate dropped typed and geocoded fire calls. +_LLM_ORPHAN_VETO_SEVERITIES = ("moderate", "major") -def _recent_incident_on_same_talkgroup(ctx: dict) -> bool: +def _llm_orphan_vetoed(ctx: dict) -> bool: + from app.internal.incident_correlator import has_event_substance + return bool( + ctx.get("reassignment") or ctx.get("incident_type") + or ctx.get("call_severity") in _LLM_ORPHAN_VETO_SEVERITIES + or has_event_substance(ctx) + ) + + +def _merge_llm_decision(rules: dict, llm: dict, ctx: dict) -> tuple[dict, str]: """ - True when a recent incident is running on this call's own system + - talkgroup, within `settings.tg_dispatch_thin_idle_minutes` (5 min) — - applied uniformly regardless of the talkgroup's name (server-26#134). - Covers "unit dispatched, thin ack 10-30s later": the ack has no - substance of its own but plainly belongs to the job just opened. - - Reads ctx["recent"] (the rules engine's own candidate list — no extra - Firestore read). That list is status=="active" incidents only, so an - already-resolved or capacity-capped same-talkgroup incident won't be - seen here even if chronologically recent (server-26#115, unresolved — - would need a dedicated non-status-filtered query). - - Whether this limitation explains the 2/24 unexplained gate misses in the - window #3 measurement is UNANSWERED, not confirmed either way — a prior - pass here claimed a "confirmed explanation" for both that turned out to - be self-contradictory. Read `corr_gate_veto` (written to corr_debug on - every escalation of this exact disagreement shape — see the caller) in - the next measurement window instead of guessing from the raw dump again. - # TODO(server-26#115): add a talkgroup-scoped incident lookup (any - # status, no capacity filter) if a future measurement window pins a real - # gate miss on a resolved/capped same-talkgroup incident. + The decision to commit for an ambiguous scene, and the corr_consensus + value that says who made it: "llm", or "rules_guard" when the LLM's + orphan was refused for a call with a strong new-event signal. """ - from app.internal.incident_correlator import _idle_gate_minutes - - tg_id = ctx.get("talkgroup_id") - system_id = ctx.get("system_id") - if tg_id is None or not system_id: - return False - tg_str = str(tg_id) - now = ctx.get("now") or clock.now() - idle_limit = settings.tg_dispatch_thin_idle_minutes - for inc in ctx.get("recent") or []: - if system_id not in (inc.get("system_ids") or []): - continue - if tg_str not in (inc.get("talkgroup_ids") or []): - continue - if _idle_gate_minutes(inc, now) <= idle_limit: - return True - return False + if llm["action"] == "orphan" and rules["action"] == "new" and _llm_orphan_vetoed(ctx): + return rules, "rules_guard" + if ( + llm["action"] == rules["action"] == "link" + and llm["matched_incident"]["incident_id"] == rules["matched_incident"]["incident_id"] + ): + # The LLM confirmed the rules' own link: keep the rules' path/fit + # signal so corr_path histograms stay comparable across runs. + return rules, "llm" + final = { + "action": llm["action"], + "matched_incident": llm["matched_incident"], + "incident_type": llm["incident_type"], + "corr_debug": dict(llm.get("corr_debug") or {}), + } + if llm["action"] == "link": + # The rules' corr_path/fit signal described a different judgement + # (possibly a different incident); keep it under corr_rules_* only. + final["corr_debug"]["corr_path"] = "llm" + elif llm["action"] == "orphan" and rules["action"] == "link": + # Tombstone: the rules-only sweep would otherwise re-commit the very + # unverified unit link the LLM just refused. + final["corr_debug"]["corr_path"] = "llm_orphan" + return final, "llm" -def _call_is_substanceless(ctx: dict) -> tuple[bool, Optional[str]]: - """ - True when the call carries nothing that marks it as a real event: - • no resolved incident_type and not a reassignment, AND - • severity is not moderate/major, AND - • no vehicle, geocode or tag (incident_correlator.has_event_substance — - the same predicate the incident-creation gate uses), AND - • no recent incident already running on the same talkgroup. - Only then may the LLM-orphan gate drop the call without a tiebreak. - - Returns (substanceless, veto_reason). veto_reason names whichever - condition kept the tiebreak alive ("type" | "reassignment" | "severity" | - "substance" | "recent_tg"), or None when the call is substanceless. The - caller writes this into corr_debug on the escalation path so a live - measurement window can see *why* each llm=orphan/rules=new call escaped - the gate instead of inferring it after the fact from the raw dump — - exactly the guesswork that produced a wrong "confirmed explanation" for - 2 window-#3 misses on the first pass of this fix. - """ - from app.internal import incident_correlator - - # The incident-creation gate skips the has_event_substance check entirely - # when a type resolved (incident_correlator._run_decision ~:1397), so a - # typed call — fire/medical/etc. — opens an incident on substance we do not - # re-check here. reassignment=True is dispatch pulling a unit onto a NEW - # job (units are blanked at :296 for exactly that reason): the strongest - # new-incident signal in the pipeline. Either one means "keep the tiebreak". - if ctx.get("incident_type"): - return False, "type" - if ctx.get("reassignment"): - return False, "reassignment" - if (ctx.get("call_severity") or "routine") in ("moderate", "major"): - return False, "severity" - if incident_correlator.has_event_substance(ctx): - return False, "substance" - if _recent_incident_on_same_talkgroup(ctx): - return False, "recent_tg" - return True, None - - -async def _correlate_with_consensus( +async def _correlate( call_id: str, node_id: str, system_id: Optional[str], @@ -222,19 +179,22 @@ async def _correlate_with_consensus( severity: Optional[str] = None, transcript: Optional[str] = None, scene_index: int = 0, -) -> Optional[str]: +) -> tuple[Optional[str], dict]: """ - Consensus correlator: runs the rules engine and the cheap LLM in sequence. - If they agree the rules decision is committed directly. - If they disagree a smarter tiebreaker LLM makes the final call. + Correlate one scene (server-26#299): the rules engine decides; only when + it marks its decision ambiguous is ONE cheap LLM call made, and a usable + LLM answer is committed instead. Returns (incident_id, committed decision). - Falls back to rules-only when GEMINI_API_KEY is absent, the call is - content-free (thin), or any LLM call fails. + corr_consensus on the call doc says who decided: + "rules_only" — confident rules decision, no LLM call + "llm" — ambiguous, the LLM's answer was committed + "rules_fallback" — ambiguous, LLM unavailable/failed/unusable → rules + "rules_guard" — ambiguous, LLM said orphan on a call with a strong + new-event signal → rules "new" kept + and corr_band names why it was ambiguous. - ``scene_index`` (server-26#96) — which scene of the call this is, from the - caller's ``enumerate(scenes)`` loop. Threaded through so the call doc's - per-scene ``scenes`` map records this scene's own corr_debug/transcript - instead of colliding with every other scene's write on the flat fields. + ``scene_index`` (server-26#96) — which scene of the call this is, so the + call doc's per-scene ``scenes`` map records this scene's own decision. """ from app.internal import incident_correlator, llm_correlator @@ -247,68 +207,119 @@ async def _correlate_with_consensus( embedding=embedding, severity=severity, transcript=transcript, scene_index=scene_index, ) - ctx = preview["ctx"] - rules_decision = preview["decision"] + ctx = preview["ctx"] + rules = preview["decision"] + band = rules.get("ambiguous") - llm_decision = await llm_correlator.decide(call_id, ctx) + if not band: + rules["corr_debug"]["corr_consensus"] = "rules_only" + return await incident_correlator.apply_correlation(preview), rules - if llm_decision is None: - # LLM unavailable, skipped (thin call), or errored — rules wins. - rules_decision["corr_debug"]["corr_consensus"] = "rules_only" - return await incident_correlator.apply_correlation(preview) - - if llm_correlator.decisions_agree(rules_decision, llm_decision): - rules_decision["corr_debug"]["corr_consensus"] = "agreed" - rules_decision["corr_debug"]["corr_llm_reasoning"] = llm_decision.get("reasoning", "") - return await incident_correlator.apply_correlation(preview) - - # server-26#115 — LLM-orphan gate. - # When the cheap LLM says `orphan`, the rules engine says `new`, and the call - # is genuinely substanceless (routine severity, no vehicle/geocode/tag, and - # no incident already running on this talkgroup), resolve to `orphan` and DO - # NOT pay for the smart tiebreaker. A bare rules `new` there means only - # "nothing to link to" — trivially true for radio housekeeping (check-ins, - # roll call, 10-8/10-98) — and the tiebreaker rubber-stamped it ~21/21 of the - # time on exactly this disagreement (CORRELATION_REVIEW_0907b.md). Any real - # signal (severity, coords, tags, a live same-talkgroup incident) still - # escalates, so an event the LLM misreads as orphan is not lost. - is_orphan_vs_new = llm_decision["action"] == "orphan" and rules_decision["action"] == "new" - substanceless, gate_veto_reason = _call_is_substanceless(ctx) if is_orphan_vs_new else (False, None) - if is_orphan_vs_new and substanceless: - logger.info( - f"Consensus gate for call {call_id}: llm=orphan vs rules=new and call " - f"is substanceless — resolving orphan, skipping tiebreak" - ) - gated = { - "action": "orphan", - "matched_incident": None, - "incident_type": None, - "corr_debug": dict(rules_decision.get("corr_debug") or {}), - } - gated["corr_debug"].update({ - "corr_consensus": "llm_orphan_gate", - "corr_rules_action": rules_decision["action"], - "corr_llm_action": llm_decision["action"], - "corr_llm_reasoning": llm_decision.get("reasoning", ""), - }) - return await incident_correlator.apply_correlation({"decision": gated, "ctx": ctx}) - - # Disagree — escalate to the smarter tiebreaker. + llm = await llm_correlator.decide(call_id, ctx) + if llm is None: + final, consensus = rules, "rules_fallback" + else: + final, consensus = _merge_llm_decision(rules, llm, ctx) + final["corr_debug"].update({ + "corr_consensus": consensus, + "corr_band": band, + "corr_rules_action": rules["action"], + }) + if llm is not None: + final["corr_debug"]["corr_llm_action"] = llm["action"] + final["corr_debug"]["corr_llm_reasoning"] = llm.get("reasoning", "") logger.info( - f"Consensus disagreement for call {call_id}: " - f"rules={rules_decision['action']} vs llm={llm_decision['action']} — tiebreak" + f"Correlation for call {call_id} scene {scene_index}: ambiguous ({band}) — " + f"rules={rules['action']} llm={llm['action'] if llm else 'none'} → {consensus}" ) - final = await llm_correlator.tiebreak(rules_decision, llm_decision, ctx) - final["corr_debug"]["corr_consensus"] = "tiebreak" - final["corr_debug"]["corr_rules_action"] = rules_decision["action"] - final["corr_debug"]["corr_llm_action"] = llm_decision["action"] - if is_orphan_vs_new: - # server-26#115 — record *why* the llm=orphan/rules=new gate stood - # down instead of leaving a future measurement window to guess it - # from the raw dump (which produced a wrong "confirmed explanation" - # for 2/24 misses the first time around). - final["corr_debug"]["corr_gate_veto"] = gate_veto_reason - return await incident_correlator.apply_correlation({"decision": final, "ctx": ctx}) + return await incident_correlator.apply_correlation({"decision": final, "ctx": ctx}), final + + +def _may_resolve(decision: dict) -> bool: + """ + server-26#88: honour a scene's `resolved` flag only on a link the + correlator was confident about, or on the incident this very scene opened. + A scene that attached by thin recency, or whose placement was ambiguous + (an LLM pick, or a rules fallback in the ambiguous band), has no business + closing an incident it may not belong to — that closed live incidents + while the event was still on the air. + """ + if decision.get("action") == "new": + return True + debug = decision.get("corr_debug") or {} + return ( + decision.get("action") == "link" + and debug.get("corr_consensus") == "rules_only" + and debug.get("corr_path") != "fast/thin" + ) + + +async def _correlate_scenes( + call_id: str, + node_id: str, + system_id: Optional[str], + talkgroup_id: Optional[int], + talkgroup_name: Optional[str], + scenes: list[dict], +) -> tuple[list[str], list[str]]: + """ + Correlate each scene of one call independently — a single recording can + carry several events on a busy channel — and apply each scene's LLM + closure. Returns (incident_ids, merged tags). Shared by the live pipeline, + replay (via _extract_and_correlate) and the transcript-PATCH path. + """ + from app.internal import incident_correlator + + incident_ids: list[str] = [] + all_tags: list[str] = [] + for scene_index, scene in enumerate(scenes): + all_tags.extend(scene["tags"]) + # When dispatch is pulling a unit to a NEW call (reassignment), suppress + # unit overlap so the new scene doesn't chain into the unit's previous + # incident. parse_llm_bool (#88): a quoted "false" is not True. + is_reassignment = incident_correlator.parse_llm_bool(scene.get("reassignment")) + corr_units = [] if is_reassignment else scene.get("units") + incident_id, decision = await _correlate( + call_id=call_id, + node_id=node_id, + system_id=system_id, + talkgroup_id=talkgroup_id, + talkgroup_name=talkgroup_name, + tags=scene["tags"], + incident_type=scene["incident_type"], + location=scene["location"], + location_coords=scene["location_coords"], + units=corr_units, + vehicles=scene.get("vehicles"), + cleared_units=scene.get("cleared_units"), + reassignment=is_reassignment, + embedding=scene.get("embedding"), + severity=scene.get("severity"), + transcript=scene.get("transcript"), + scene_index=scene_index, + ) + if incident_id and incident_id not in incident_ids: + incident_ids.append(incident_id) + if incident_id and incident_correlator.parse_llm_bool(scene.get("resolved")): + if not _may_resolve(decision): + logger.info( + f"Ignoring closure from call {call_id} scene {scene_index}: link into " + f"{incident_id} was not confident (server-26#88)" + ) + continue + await fstore.doc_set("incidents", incident_id, { + "status": "resolved", + "resolved_at": clock.now().isoformat(), + "resolved_via": "llm_closure", + # One transmission read as "it's over" ("transport complete") + # closed the whole 09-22 bridge MVA at 14:44 and its next 56 + # calls opened a second incident (server-26#170). Inferred from + # a single call, so provisional, like a timer close. + "reopenable": True, + }) + await incident_correlator.maybe_resolve_parent(incident_id) + logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)") + return incident_ids, all_tags async def _resolve_flags(system_id: Optional[str]): @@ -335,7 +346,7 @@ async def _run_extraction_pipeline( preserve_transcript_correction: bool = False, ) -> None: """Run steps 2-4 of the intelligence pipeline using an existing transcript.""" - from app.internal import intelligence, incident_correlator, alerter + from app.internal import intelligence, alerter flags, _flag = await _resolve_flags(system_id) @@ -353,45 +364,9 @@ async def _run_extraction_pipeline( ) # Step 3: Correlate each scene to an incident independently. - # server-26#96: scene_index is threaded through so each scene's - # corr_debug/transcript lands in its own entry of the call doc's - # `scenes` map instead of clobbering every other scene's write. - for scene_index, scene in enumerate(scenes): - all_tags.extend(scene["tags"]) - # When dispatch is pulling a unit to a NEW call (reassignment), suppress unit - # overlap so the new scene doesn't chain into the unit's previous incident. - is_reassignment = bool(scene.get("reassignment")) - corr_units = [] if is_reassignment else scene.get("units") - incident_id = await _correlate_with_consensus( - call_id=call_id, - node_id=node_id, - system_id=system_id, - talkgroup_id=talkgroup_id, - talkgroup_name=talkgroup_name, - tags=scene["tags"], - incident_type=scene["incident_type"], - location=scene["location"], - location_coords=scene["location_coords"], - units=corr_units, - vehicles=scene.get("vehicles"), - cleared_units=scene.get("cleared_units"), - reassignment=is_reassignment, - embedding=scene.get("embedding"), - severity=scene.get("severity"), - transcript=scene.get("transcript"), - scene_index=scene_index, - ) - if incident_id and incident_id not in incident_ids: - incident_ids.append(incident_id) - if scene["resolved"] and incident_id: - await fstore.doc_set("incidents", incident_id, { - "status": "resolved", - "resolved_at": clock.now().isoformat(), - "resolved_via": "llm_closure", - "reopenable": True, # provisional, see _extract_and_correlate - }) - await incident_correlator.maybe_resolve_parent(incident_id) - logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)") + incident_ids, all_tags = await _correlate_scenes( + call_id, node_id, system_id, talkgroup_id, talkgroup_name, scenes, + ) else: scope = "globally" if not flags["correlation_enabled"] else f"system {system_id}" logger.info(f"Correlation disabled ({scope}) — skipping scene extraction and correlation for call {call_id} (reprocess)") @@ -430,7 +405,7 @@ async def _extract_and_correlate( copy of it that can drift. Caller owns the correlation feature-flag check and alerting. """ - from app.internal import intelligence, incident_correlator + from app.internal import intelligence # Step 2: Scene detection + intelligence extraction if scenes is None: @@ -443,50 +418,9 @@ async def _extract_and_correlate( ) # Step 3: Correlate each scene independently. - # A single recording can produce multiple incidents on a busy channel. - incident_ids: list[str] = [] - all_tags: list[str] = [] - # server-26#96: scene_index is threaded through so each scene's - # corr_debug/transcript lands in its own entry of the call doc's - # `scenes` map instead of clobbering every other scene's write. - for scene_index, scene in enumerate(scenes): - all_tags.extend(scene["tags"]) - is_reassignment = bool(scene.get("reassignment")) - corr_units = [] if is_reassignment else scene.get("units") - incident_id = await _correlate_with_consensus( - call_id=call_id, - node_id=node_id, - system_id=system_id, - talkgroup_id=talkgroup_id, - talkgroup_name=talkgroup_name, - tags=scene["tags"], - incident_type=scene["incident_type"], - location=scene["location"], - location_coords=scene["location_coords"], - units=corr_units, - vehicles=scene.get("vehicles"), - cleared_units=scene.get("cleared_units"), - reassignment=is_reassignment, - embedding=scene.get("embedding"), - severity=scene.get("severity"), - transcript=scene.get("transcript"), - scene_index=scene_index, - ) - if incident_id and incident_id not in incident_ids: - incident_ids.append(incident_id) - if scene["resolved"] and incident_id: - await fstore.doc_set("incidents", incident_id, { - "status": "resolved", - "resolved_at": clock.now().isoformat(), - "resolved_via": "llm_closure", - # One transmission read as "it's over" ("transport complete") - # closed the whole 09-22 bridge MVA at 14:44 and its next 56 - # calls opened a second incident (server-26#170). Inferred from - # a single call, so provisional, like a timer close. - "reopenable": True, - }) - await incident_correlator.maybe_resolve_parent(incident_id) - logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)") + incident_ids, all_tags = await _correlate_scenes( + call_id, node_id, system_id, talkgroup_id, talkgroup_name, scenes, + ) # Correlator also runs for calls with no scenes (unclassified) to attempt # talkgroup-based linking even when no transcript could be produced. @@ -502,7 +436,7 @@ async def _extract_and_correlate( _call_doc = await fstore.doc_get("calls", call_id) skip_reason = (_call_doc or {}).get("skip_reason") if not skip_reason or skip_reason == "transcript_too_short": - incident_id = await _correlate_with_consensus( + incident_id, _ = await _correlate( call_id=call_id, node_id=node_id, system_id=system_id, diff --git a/drb-c2-core/tests/test_ai_flag_gating.py b/drb-c2-core/tests/test_ai_flag_gating.py index b7524c0..49b0690 100644 --- a/drb-c2-core/tests/test_ai_flag_gating.py +++ b/drb-c2-core/tests/test_ai_flag_gating.py @@ -100,7 +100,7 @@ async def _run_ingest(global_correlation, system_ai_flags): ) with g, sysdoc, \ patch.object(upload, "fstore") as fs, \ - patch.object(upload, "_correlate_with_consensus", AsyncMock(return_value=None)) as corr, \ + patch.object(upload, "_correlate", AsyncMock(return_value=(None, {}))) as corr, \ patch("app.internal.transcription.transcribe_call", AsyncMock(return_value=("units respond to main street", []))), \ patch("app.internal.intelligence.extract_scenes", AsyncMock(return_value=[])) as scenes, \ diff --git a/drb-c2-core/tests/test_ai_health.py b/drb-c2-core/tests/test_ai_health.py index d75e419..34ba7f7 100644 --- a/drb-c2-core/tests/test_ai_health.py +++ b/drb-c2-core/tests/test_ai_health.py @@ -136,15 +136,15 @@ async def test_transient_failure_alerts_once_threshold_crossed(): async def test_recovery_alerts_once_after_an_alerted_episode(): with patch.object(ai_health, "_post_webhook", new=AsyncMock()) as webhook: await ai_health.report_degraded( - "correlation_smart", "gemini", "gemini-1.5-pro", + "correlation_cheap", "gemini", "gemini-3.6-flash", "the Gemini account is out of credit", "top up billing", permanent=True, ) webhook.reset_mock() - await ai_health.report_healthy("correlation_smart") + await ai_health.report_healthy("correlation_cheap") webhook.assert_awaited_once() - state = ai_health.snapshot()["correlation_smart"] + state = ai_health.snapshot()["correlation_cheap"] assert state["degraded"] is False assert state["alerted"] is False assert state["consecutive_failures"] == 0 diff --git a/drb-c2-core/tests/test_ai_timeouts.py b/drb-c2-core/tests/test_ai_timeouts.py index ffb4b56..140ddac 100644 --- a/drb-c2-core/tests/test_ai_timeouts.py +++ b/drb-c2-core/tests/test_ai_timeouts.py @@ -153,8 +153,10 @@ def test_timeout_messages_classify_as_transient(): def _corr_ctx(): + from datetime import datetime, timezone return {"call_id": "c1", "is_thin_call": False, "recent": [{"incident_id": "i1"}], - "all_active": [], "call_units": ["12"], "call_cleared": [], "tags": ["mva"]} + "all_active": [], "call_units": ["12"], "call_cleared": [], "tags": ["mva"], + "now": datetime.now(timezone.utc)} @pytest.mark.asyncio @@ -163,27 +165,14 @@ async def test_gemini_timeout_in_decide_falls_back_to_rules(): raise httpx.ReadTimeout("The read operation timed out") with patch.object(settings, "gemini_api_key", "test-key"), \ - patch.object(llm_correlator, "_build_decide_prompt", lambda ctx: "p"), \ + patch.object(llm_correlator, "_build_decide_prompt", lambda ctx, cands: "p"), \ patch.object(llm_correlator, "_sync_gemini", hang): - assert await llm_correlator.decide("c1", _corr_ctx()) is None # None == rules_only + assert await llm_correlator.decide("c1", _corr_ctx()) is None # None == keep the rules decision state = ai_health.snapshot()["correlation_cheap"] assert state["consecutive_failures"] == 1 assert state["permanent"] is False -@pytest.mark.asyncio -async def test_gemini_timeout_in_tiebreak_keeps_the_rules_decision(): - def hang(model, prompt): - raise httpx.ReadTimeout("The read operation timed out") - - rules = {"action": "new", "matched_incident": None, "corr_debug": {}} - with patch.object(llm_correlator, "_build_tiebreak_prompt", lambda r, l, c: "p"), \ - patch.object(llm_correlator, "_sync_gemini", hang): - out = await llm_correlator.tiebreak(rules, {"action": "orphan"}, _corr_ctx()) - assert out is rules - assert ai_health.snapshot()["correlation_smart"]["permanent"] is False - - @pytest.mark.asyncio async def test_gemini_timeout_in_transcript_correction_keeps_raw_text(): def hang(model, prompt): @@ -233,8 +222,8 @@ def test_worst_case_bound_follows_the_settings(): patch.object(settings, "gemini_timeout_seconds", 20.0), \ patch.object(settings, "stt_retry_on_degenerate", True): bound = recorrelation_sweep.pipeline_worst_case_seconds(n_scenes=2) - # stt 2x2x100 + correction 20 + extraction 2x10 + embeds 2x2x10 + decide/tiebreak 2x20 + slack - assert bound == 400 + 20 + 20 + 40 + 40 + recorrelation_sweep._NON_AI_SLACK_SECONDS + # stt 2x2x100 + correction 20 + extraction 2x10 + embeds 2x2x10 + one correlation call 20 + slack + assert bound == 400 + 20 + 20 + 40 + 20 + recorrelation_sweep._NON_AI_SLACK_SECONDS def test_hold_back_covers_the_worst_case_and_never_drops_below_old_floor(): diff --git a/drb-c2-core/tests/test_clearance_tracking.py b/drb-c2-core/tests/test_clearance_tracking.py index 318d07e..38cff93 100644 --- a/drb-c2-core/tests/test_clearance_tracking.py +++ b/drb-c2-core/tests/test_clearance_tracking.py @@ -53,12 +53,11 @@ def test_a_unit_never_on_the_incident_cannot_close_it(): 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": []}) + assert ic.is_clearance_only(ctx) + assert not ic.is_clearance_only({**ctx, "tags": ["mva"]}) + assert not ic.is_clearance_only({**ctx, "call_cleared": []}) def test_only_numbered_units_hold_an_incident_open(): diff --git a/drb-c2-core/tests/test_consensus_gate.py b/drb-c2-core/tests/test_consensus_gate.py deleted file mode 100644 index 1eeab8e..0000000 --- a/drb-c2-core/tests/test_consensus_gate.py +++ /dev/null @@ -1,430 +0,0 @@ -""" -server-26#115 — two consensus-quality fixes. - -Fix 1 (routers/upload.py): when the cheap LLM says `orphan`, the rules engine -says `new`, and the call is genuinely SUBSTANCELESS (routine severity, no -vehicle/geocode/tag, and no incident already running on the same talkgroup), -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). - -The substance test runs against `ctx` (fully populated at preview time), NOT -against `rules_decision["corr_debug"]` — that dict is EMPTY at preview time for -action=="new" (corr_path:"new" is written at APPLY time), so the first version of -this gate fired on real events (a `major` "extinguishing fire", geocoded calls, -pursuit updates). - -Fix 2 (incident_correlator.py): the `location` correlation path linked on a bare -sub-`location_proximity_km` (0.5 km) distance alone, taking whichever incident -came first in an unsorted `recent`. 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, and picks the NEAREST qualifying -candidate. A unit-overlap location link is tagged `location_unit_overlap` so it -does not merge into the fast path's bucket in the admin fit-signal histogram. -""" -from datetime import datetime, timedelta, timezone -from unittest.mock import AsyncMock, patch - -import pytest - -from app.routers import upload -from app.internal.incident_correlator import _run_decision, has_event_substance - -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, ctx=None): - base_ctx = {"call_id": "call-1"} - if ctx: - base_ctx.update(ctx) - 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": base_ctx, - } - - -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_substanceless_no_recent_same_tg_incident_gates_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("severity", ["moderate", "major"]) -async def test_moderate_or_major_severity_is_not_gated(severity): - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx={"call_severity": severity}), _llm("orphan"), - ) - m_tiebreak.assert_called_once() - - -async def test_routine_severity_alone_still_gates(): - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx={"call_severity": "routine"}), _llm("orphan"), - ) - m_tiebreak.assert_not_called() - assert m_apply.call_args[0][0]["decision"]["action"] == "orphan" - - -async def test_call_with_coords_is_not_gated(): - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx={"coords": {"lat": 41.15, "lng": -73.86}}), - _llm("orphan"), - ) - m_tiebreak.assert_called_once() - - -async def test_call_with_tags_is_not_gated(): - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx={"tags": ["structure-fire"]}), _llm("orphan"), - ) - m_tiebreak.assert_called_once() - - -async def test_call_with_vehicles_is_not_gated(): - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx={"call_vehicles": ["red sedan"]}), _llm("orphan"), - ) - m_tiebreak.assert_called_once() - - -async def test_call_with_resolved_incident_type_is_not_gated(): - # The creation gate skips has_event_substance when a type resolved, so a - # typed call (fire/medical/…) opens an incident on substance the gate does - # not re-check — it must keep the tiebreak, not be dropped. - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx={"incident_type": "fire"}), _llm("orphan"), - ) - m_tiebreak.assert_called_once() - - -async def test_reassignment_call_is_not_gated(): - # reassignment=True is dispatch pulling a unit onto a NEW job (units are - # blanked for exactly that reason) — the strongest new-incident signal. - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx={"reassignment": True}), _llm("orphan"), - ) - m_tiebreak.assert_called_once() - - -async def test_recent_incident_on_same_talkgroup_is_not_gated(): - ctx = { - "system_id": "sys-1", - "talkgroup_id": 9048, - "talkgroup_name": "Dispatch", - "now": NOW, - "recent": [{ - "incident_id": "inc-live", - "system_ids": ["sys-1"], - "talkgroup_ids": ["9048"], - "updated_at": (NOW - timedelta(minutes=1)).isoformat(), - }], - } - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx=ctx), _llm("orphan"), - ) - m_tiebreak.assert_called_once() - - -# server-26#115 window #3 (CORRELATION_REVIEW_0912.md): the escape hatch used -# to treat ANY same-talkgroup incident inside the 2h correlation_window_hours -# as "recent", which on a busy dispatch channel (3-13 incidents/2h) was -# satisfied almost unconditionally — the gate fired 0/24 times against its own -# target shape. It now only counts an incident as recent within -# settings.tg_dispatch_thin_idle_minutes (5 min), applied uniformly regardless -# of the talkgroup's name (owner correction, 2026-09-13 — see -# test_channel_name_does_not_affect_the_window below for why the dichotomy -# this originally had with incident_correlator's fast/thin idle selection was -# removed here). - -async def test_recent_same_tg_incident_inside_new_short_window_still_escapes_gate(): - ctx = { - "system_id": "sys-1", - "talkgroup_id": 9048, - "talkgroup_name": "Dispatch", - "now": NOW, - "recent": [{ - "incident_id": "inc-live", - "system_ids": ["sys-1"], - "talkgroup_ids": ["9048"], - # 3 min ago — inside tg_dispatch_thin_idle_minutes (5). - "updated_at": (NOW - timedelta(minutes=3)).isoformat(), - }], - } - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx=ctx), _llm("orphan"), - ) - m_tiebreak.assert_called_once() - - -async def test_recent_same_tg_incident_older_than_short_window_now_gates(): - # Regression test for the fix: 8 minutes is past the 5-minute bound but - # still inside the OLD 2-hour correlation_window_hours lookback. Before - # the fix this escaped the gate on any channel; after the fix it gates. - ctx = { - "system_id": "sys-1", - "talkgroup_id": 9048, - "talkgroup_name": "Dispatch", - "now": NOW, - "recent": [{ - "incident_id": "inc-stale", - "system_ids": ["sys-1"], - "talkgroup_ids": ["9048"], - "updated_at": (NOW - timedelta(minutes=8)).isoformat(), - }], - } - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx=ctx), _llm("orphan"), - ) - m_tiebreak.assert_not_called() - assert m_apply.call_args[0][0]["decision"]["action"] == "orphan" - - -async def test_channel_name_does_not_affect_the_window(): - # Owner correction, 2026-09-13 (direct scanning experience): a talkgroup - # named "tac"/"tactical" only sees materially different traffic during a - # real incident, and that's rare -- the bulk of traffic on any monitored - # channel, including high-risk stops and pursuits, runs on the main - # channel regardless of what it's named. An earlier version of this used - # a longer 15-minute window on anything not literally named "dispatch"/ - # "patched"/"primary" (mirroring incident_correlator's fast/thin idle - # selection); that meant a busy single-channel department not literally - # named "dispatch" silently got the more permissive window and could - # reproduce #115's original bug. Same 8-minute age as the dispatch-named - # test above, but on a channel named "Tac 3" -- must gate identically, - # not escape into a longer window just because of the name. - ctx = { - "system_id": "sys-1", - "talkgroup_id": 383, - "talkgroup_name": "Tac 3", - "now": NOW, - "recent": [{ - "incident_id": "inc-tac", - "system_ids": ["sys-1"], - "talkgroup_ids": ["383"], - "updated_at": (NOW - timedelta(minutes=8)).isoformat(), - }], - } - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx=ctx), _llm("orphan"), - ) - m_tiebreak.assert_not_called() - assert m_apply.call_args[0][0]["decision"]["action"] == "orphan" - - -async def test_gate_veto_reason_is_recorded_on_the_escalation_path(): - # server-26#115: a live measurement window must be able to see *why* an - # llm=orphan/rules=new call escaped the gate without guessing from the raw - # dump (which produced a wrong "confirmed explanation" for 2 window-#3 - # misses the first time). corr_gate_veto names the surviving condition. - ctx = {"call_severity": "major"} - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx=ctx), _llm("orphan"), - ) - m_tiebreak.assert_called_once() - final = m_apply.call_args[0][0]["decision"] - assert final["corr_debug"]["corr_gate_veto"] == "severity" - - -async def test_gate_veto_reason_is_absent_when_the_disagreement_is_not_orphan_vs_new(): - # corr_gate_veto is only meaningful for the llm=orphan/rules=new shape the - # gate targets — it must not appear (or be misleadingly None-vs-absent) on - # an unrelated disagreement shape. - m_apply, m_tiebreak = await _run_consensus( - _preview("link", {}), _llm("orphan"), - ) - m_tiebreak.assert_called_once() - final = m_apply.call_args[0][0]["decision"] - assert "corr_gate_veto" not in final["corr_debug"] - - -async def test_recent_incident_on_a_different_talkgroup_still_gates(): - ctx = { - "system_id": "sys-1", - "talkgroup_id": 9048, - "recent": [{ - "incident_id": "inc-other", - "system_ids": ["sys-1"], - "talkgroup_ids": ["1200"], - }], - } - m_apply, m_tiebreak = await _run_consensus( - _preview("new", {}, ctx=ctx), _llm("orphan"), - ) - m_tiebreak.assert_not_called() - assert m_apply.call_args[0][0]["decision"]["action"] == "orphan" - - -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() - - -async def test_llm_orphan_vs_rules_link_still_escalates(): - # Not the gate condition (gate needs rules=="new"); must fall through. - m_apply, m_tiebreak = await _run_consensus(_preview("link", {}), _llm("orphan")) - m_tiebreak.assert_called_once() - - -def test_has_event_substance_predicate(): - assert has_event_substance({"coords": {"lat": 1, "lng": 2}}) - assert has_event_substance({"tags": ["fire"]}) - assert has_event_substance({"call_vehicles": ["sedan"]}) - assert not has_event_substance({}) - assert not has_event_substance({"coords": None, "tags": [], "call_vehicles": []}) - # units and location are NOT substance — nearly every transmission has them. - assert not has_event_substance({"call_units": ["7-Adam"], "location": "Main St"}) - - -# ───────────────────────────────────────────────────────────────────────────── -# 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} -# ~0.28 km north — inside the 0.5 radius, outside the 0.2 tight bar; used as a -# second candidate that must lose the nearest-wins sort to NEAR_INC_COORDS. -MID_INC_COORDS = {"lat": 41.152500, "lng": -73.860000} - - -def _inc(incident_id, coords, units): - return { - "incident_id": incident_id, - "system_ids": ["sys-1"], - "talkgroup_ids": ["100"], # different TGID → fast path is a no-op - "location_coords": coords, - "units": units, - "tags": [], - "type": "police", - "updated_at": (NOW - timedelta(minutes=6)).isoformat(), - "started_at": (NOW - timedelta(minutes=20)).isoformat(), - "status": "active", - "call_ids": ["c0"], - } - - -def _loc_ctx(*, incidents, call_units): - return { - "call_id": "call-loc", - "all_active": list(incidents), - "recent": list(incidents), - "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_in_radius_but_no_unit_overlap_no_tight_proximity_does_not_link(caplog): - ctx = _loc_ctx( - incidents=[_inc("inc-loc", FAR_INC_COORDS, ["7-Adam"])], - call_units=["3-Boy"], - ) - with caplog.at_level("INFO", logger="drb-c2-core"): - decision = _run_decision(ctx) - # Reaches, and is rejected by, the new guard (not an earlier path). - assert "location-path skipped" in caplog.text - assert decision["action"] != "link" - assert (decision.get("corr_debug") or {}).get("corr_path") != "location" - - -def test_location_path_links_on_unit_overlap_with_distinct_fit_signal(): - ctx = _loc_ctx( - incidents=[_inc("inc-loc", FAR_INC_COORDS, ["5-Adam"])], - call_units=["5-Adam"], - ) - decision = _run_decision(ctx) - assert decision["action"] == "link" - assert decision["corr_debug"]["corr_path"] == "location" - # NOT "unit_overlap" — that value belongs to the fast path's histogram bucket. - assert decision["corr_debug"]["corr_fit_signal"] == "location_unit_overlap" - - -def test_location_path_links_on_tight_proximity_without_unit_overlap(): - ctx = _loc_ctx( - incidents=[_inc("inc-loc", NEAR_INC_COORDS, ["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" - - -def test_location_path_picks_nearest_in_radius_candidate(): - # `recent` order puts the farther tight-proximity incident first; the guard - # must still select the nearest one. - ctx = _loc_ctx( - incidents=[ - _inc("inc-mid", MID_INC_COORDS, ["3-Boy"]), # ~0.28 km, tight-fail - _inc("inc-near", NEAR_INC_COORDS, ["3-Boy"]), # ~0.13 km, tight-pass - ], - call_units=["3-Boy"], - ) - decision = _run_decision(ctx) - assert decision["action"] == "link" - assert decision["matched_incident"]["incident_id"] == "inc-near" - assert decision["corr_debug"]["corr_path"] == "location" diff --git a/drb-c2-core/tests/test_correlation_decision_flow.py b/drb-c2-core/tests/test_correlation_decision_flow.py new file mode 100644 index 0000000..99b99f6 --- /dev/null +++ b/drb-c2-core/tests/test_correlation_decision_flow.py @@ -0,0 +1,367 @@ +""" +server-26#299 — rules decide confident outcomes; ONE cheap LLM call only in the +ambiguous band; no cheap/smart consensus tiebreak. + +Also pins the section fixes that ride on the new shape: + #86 divergence veto bar is a setting, and the band between it and the match + bar is ambiguous rather than a silent merge + #103 a scene with no embedding is "unverified", not verified + #88 LLM yes/no fields fail closed; a closure only counts on a confident link + #189 weak links refused past the span cap; LLM links distance-guarded + #190 severity steps down on an explicit stand-down, the peak never does + #27 sweep budget is wall-clock +""" +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, patch + +import pytest + +from app.config import settings +from app.internal import incident_correlator as ic +from app.internal import llm_correlator, recorrelation_sweep +from app.routers import upload + +NOW = datetime(2026, 10, 27, 15, 0, 0, tzinfo=timezone.utc) + + +def _inc(incident_id="inc-1", idle=1.0, span=None, **kw) -> dict: + updated = NOW - timedelta(minutes=idle) + started = NOW - timedelta(minutes=span if span is not None else idle) + inc = { + "incident_id": incident_id, "system_ids": ["sys-1"], "talkgroup_ids": ["383"], + "updated_at": updated.isoformat(), "started_at": started.isoformat(), + "status": "active", "call_ids": ["seed"], "type": "police", + } + inc.update(kw) + return inc + + +def _ctx(incidents=(), **kw) -> dict: + incidents = list(incidents) + base = { + "call_id": "call-1", "all_active": incidents, "recent": incidents, + "call_doc": {}, "call_embedding": None, "call_units": [], "call_vehicles": [], + "call_cleared": [], "call_severity": "routine", "coords": None, + "is_thin_call": False, "now": NOW, "system_id": "sys-1", "talkgroup_id": 383, + "talkgroup_name": "Ch 1", "tags": [], "incident_type": None, "location": None, + "location_coords": None, "reassignment": False, "create_if_new": True, + } + base.update(kw) + return base + + +def _vec(sim: float) -> tuple[list, list]: + """Two unit vectors with cosine similarity `sim`.""" + import math + return [1.0, 0.0], [sim, math.sqrt(max(0.0, 1 - sim * sim))] + + +# ── The band ──────────────────────────────────────────────────────────────── + +def test_fresh_unit_link_is_confident(): + d = ic._run_decision(_ctx([_inc(idle=2, units=["6-Adam"])], call_units=["6-Adam"])) + assert d["action"] == "link" + assert d["corr_debug"]["corr_fit_signal"] == "unit_overlap" + assert d["ambiguous"] is None + + +def test_stale_unit_link_without_embedding_is_ambiguous_not_verified(): + # #103: no embedding used to skip the veto and link as if verified. + d = ic._run_decision(_ctx([_inc(idle=20, units=["6-Adam"])], call_units=["6-Adam"])) + assert d["action"] == "link" + assert d["corr_debug"]["corr_fit_signal"] == "unit_overlap_unverified" + assert d["ambiguous"] == "unverified_unit_link" + + +def test_young_but_long_running_unit_link_is_ambiguous(): + # #189: busy channel keeps idle low; age is what reveals a sink forming. + inc = _inc(idle=1, span=settings.unit_link_confident_max_span_minutes + 5, units=["6-Adam"]) + d = ic._run_decision(_ctx([inc], call_units=["6-Adam"])) + assert d["ambiguous"] == "unverified_unit_link" + + +@pytest.mark.parametrize("sim,expected", [ + (0.80, ("orphan", "content_divergence")), # below the divergence bar → veto + (0.88, ("link", "unit_overlap_unverified")), # between the bars → ambiguous (#86) + (0.95, ("link", "unit_content_match")), # at the match bar → confident +]) +def test_divergence_band_between_the_two_thresholds(sim, expected): + call_emb, inc_emb = _vec(sim) + inc = _inc(idle=20, units=["6-Adam"], embedding=inc_emb) + fit, signal = ic._call_fits_incident( + inc, ["6-Adam"], [], None, settings.location_proximity_km, + call_embedding=call_emb, now=NOW, + ) + assert (("link" if fit else "orphan"), signal) == expected + + +def test_divergence_bar_is_the_setting_not_a_literal(): + call_emb, inc_emb = _vec(0.88) + inc = _inc(idle=20, units=["6-Adam"], embedding=inc_emb) + with patch.object(settings, "embedding_divergence_threshold", 0.90): + fit, signal = ic._call_fits_incident( + inc, ["6-Adam"], [], None, settings.location_proximity_km, + call_embedding=call_emb, now=NOW, + ) + assert (fit, signal) == (False, "content_divergence") + + +def test_new_with_live_candidates_is_ambiguous_without_none(): + d = ic._run_decision(_ctx([_inc(talkgroup_ids=["999"])], incident_type="police")) + assert (d["action"], d["ambiguous"]) == ("new", "new_with_candidates") + d = ic._run_decision(_ctx([], incident_type="police")) + assert (d["action"], d["ambiguous"]) == ("new", None) + + +def test_orphan_with_unfit_same_tg_candidate_is_ambiguous(): + # Substantive (a location) but untyped, live incident on its talkgroup, no evidence. + d = ic._run_decision(_ctx([_inc(idle=2)], location="Mohegan Park Ave")) + assert (d["action"], d["ambiguous"]) == ("orphan", "unfit_tg_candidate") + + +def test_thin_sweep_and_clearance_calls_are_never_ambiguous(): + inc = _inc(idle=0.2) + assert ic._run_decision(_ctx([inc], is_thin_call=True))["ambiguous"] is None + assert ic._run_decision(_ctx([inc], incident_type="police", create_if_new=False))["ambiguous"] is None + stale = _inc(idle=20, units=["45-9"]) + d = ic._run_decision(_ctx([stale], call_units=["45-9"], call_cleared=["45-9"])) + assert d["action"] == "link" and d["ambiguous"] is None + + +# ── #189: caps on absorption ──────────────────────────────────────────────── + +def test_weak_link_span_cap_refuses_unit_only_links(): + inc = _inc(idle=1, span=settings.incident_weak_link_max_span_minutes + 1, units=["6-Adam"]) + d = ic._run_decision(_ctx([inc], call_units=["6-Adam"], incident_type="accident")) + assert d["action"] == "new" + assert ic.llm_candidates(_ctx([inc])) == [] + + +def test_weak_link_span_cap_also_binds_unit_continuity(): + inc = _inc(idle=1, span=settings.incident_weak_link_max_span_minutes + 1, + units=["6-Adam"], talkgroup_ids=["999"]) + d = ic._run_decision(_ctx([inc], call_units=["6-Adam"])) + assert d["action"] == "orphan" + + +def test_llm_link_distance_guard(): + pin = {"lat": 41.0, "lng": -73.8} + inc = _inc(location="Main St", location_coords=pin, location_coords_source="Main St") + near = _ctx(coords={"lat": 41.005, "lng": -73.8}) + far = _ctx(coords={"lat": 41.2, "lng": -73.8}) # ~22 km + assert ic.llm_link_allowed(inc, near) + assert not ic.llm_link_allowed(inc, far) + assert ic.llm_link_allowed({**inc, "tags": ["vehicle-pursuit"]}, far) + + +# ── llm_correlator: one call, validated ───────────────────────────────────── + +def test_parse_rejects_links_to_unshown_incidents_and_bad_actions(): + shown = [_inc("inc-1")] + ctx = _ctx(shown) + assert llm_correlator._parse_response({"action": "link", "incident_id": "nope"}, ctx, shown) is None + assert llm_correlator._parse_response({"action": "maybe"}, ctx, shown) is None + ok = llm_correlator._parse_response({"action": "link", "incident_id": "inc-1"}, ctx, shown) + assert ok["action"] == "link" and ok["matched_incident"]["incident_id"] == "inc-1" + + +def test_parse_new_uses_the_rules_creation_gate(): + ctx = _ctx(call_severity="minor") + assert llm_correlator._parse_response({"action": "new"}, ctx, [])["incident_type"] == "other" + out = llm_correlator._parse_response({"action": "new"}, _ctx(), []) + assert out["action"] == "orphan" + + +async def _correlate_with(rules_decision, llm_return, ctx=None): + ctx = ctx or _ctx([_inc()]) + preview = {"decision": rules_decision, "ctx": ctx} + with patch("app.internal.incident_correlator.preview_correlation", AsyncMock(return_value=preview)), \ + patch("app.internal.incident_correlator.apply_correlation", AsyncMock(return_value="inc-x")) as apply, \ + patch("app.internal.llm_correlator.decide", AsyncMock(return_value=llm_return)) as decide: + _, final = await upload._correlate( + call_id="call-1", node_id="n", system_id="sys-1", talkgroup_id=383, + talkgroup_name="Ch 1", tags=[], incident_type=None, location=None, + location_coords=None, + ) + return final, decide, apply + + +def _rules(action, ambiguous=None, inc=None): + return {"action": action, "matched_incident": inc, "incident_type": "police" if action == "new" else None, + "corr_debug": {}, "ambiguous": ambiguous} + + +def _llm(action, inc=None): + return {"action": action, "matched_incident": inc, "incident_type": "police" if action == "new" else None, + "corr_debug": {"corr_llm_reasoning": "r"}, "reasoning": "r"} + + +@pytest.mark.asyncio +async def test_confident_rules_decision_never_calls_the_llm(): + final, decide, _ = await _correlate_with(_rules("new"), _llm("orphan")) + decide.assert_not_awaited() + assert final["corr_debug"]["corr_consensus"] == "rules_only" + + +@pytest.mark.asyncio +async def test_ambiguous_decision_makes_exactly_one_llm_call_and_commits_it(): + inc = _inc("inc-2") + final, decide, apply = await _correlate_with(_rules("new", "new_with_candidates"), _llm("link", inc)) + decide.assert_awaited_once() + committed = apply.await_args.args[0]["decision"] + assert committed["action"] == "link" and committed["matched_incident"]["incident_id"] == "inc-2" + dbg = final["corr_debug"] + assert dbg["corr_consensus"] == "llm" and dbg["corr_path"] == "llm" + assert dbg["corr_band"] == "new_with_candidates" + assert (dbg["corr_rules_action"], dbg["corr_llm_action"]) == ("new", "link") + + +@pytest.mark.asyncio +async def test_llm_unavailable_keeps_the_rules_decision(): + rules = _rules("new", "new_with_candidates") + final, decide, _ = await _correlate_with(rules, None) + assert final is rules and final["corr_debug"]["corr_consensus"] == "rules_fallback" + + +@pytest.mark.asyncio +async def test_llm_orphan_cannot_drop_a_major_or_reassignment_call(): + final, _, _ = await _correlate_with( + _rules("new", "new_with_candidates"), _llm("orphan"), _ctx([_inc()], call_severity="major")) + assert final["action"] == "new" and final["corr_debug"]["corr_consensus"] == "rules_guard" + final, _, _ = await _correlate_with( + _rules("new", "new_with_candidates"), _llm("orphan"), _ctx([_inc()])) + assert final["action"] == "orphan" and final["corr_debug"]["corr_consensus"] == "llm" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("extra", [{"incident_type": "fire"}, {"tags": ["structure-fire"]}, + {"coords": {"lat": 41.0, "lng": -73.8}}, {"reassignment": True}]) +async def test_llm_orphan_cannot_drop_a_typed_or_substantive_call(extra): + final, _, _ = await _correlate_with( + _rules("new", "new_with_candidates"), _llm("orphan"), _ctx([_inc()], **extra)) + assert final["action"] == "new" and final["corr_debug"]["corr_consensus"] == "rules_guard" + + +@pytest.mark.asyncio +async def test_llm_orphan_over_a_rules_link_is_tombstoned_for_the_sweep(): + inc = _inc() + rules = _rules("link", "unverified_unit_link", inc) + final, _, _ = await _correlate_with(rules, _llm("orphan")) + assert final["action"] == "orphan" and final["corr_debug"]["corr_path"] == "llm_orphan" + + +@pytest.mark.asyncio +async def test_back_dated_link_never_moves_updated_at_backward(): + inc = _inc(idle=1, span=10) + with patch.object(ic, "fstore") as fs: + fs.doc_set = AsyncMock() + await ic._update_incident(inc, "late-call", 383, "sys-1", [], None, None, ["6-Adam"], [], None, + NOW - timedelta(hours=3)) + assert fs.doc_set.await_args.args[2]["updated_at"] == inc["updated_at"] + + +@pytest.mark.asyncio +async def test_llm_confirming_the_rules_link_keeps_the_rules_path(): + inc = _inc() + rules = _rules("link", "unverified_unit_link", inc) + rules["corr_debug"] = {"corr_path": "fast/single", "corr_fit_signal": "unit_overlap_unverified"} + final, _, _ = await _correlate_with(rules, _llm("link", inc)) + assert final["corr_debug"]["corr_path"] == "fast/single" + assert final["corr_debug"]["corr_consensus"] == "llm" + + +def test_no_tiebreak_tier_remains(): + assert not hasattr(llm_correlator, "tiebreak") + assert not hasattr(settings, "corr_smart_model") + assert not hasattr(upload, "_correlate_with_consensus") + + +# ── #88 ───────────────────────────────────────────────────────────────────── + +@pytest.mark.parametrize("value,expected", [ + (True, True), (False, False), ("true", True), ("True", True), ("yes", True), + ("false", False), ("False", False), ("no", False), ("", False), (None, False), + ("maybe", False), (1, False), ({"x": 1}, False), +]) +def test_parse_llm_bool_fails_closed(value, expected): + assert ic.parse_llm_bool(value) is expected + + +def test_closure_only_honoured_on_a_confident_link_or_own_new_incident(): + assert upload._may_resolve({"action": "new", "corr_debug": {"corr_consensus": "llm"}}) + assert upload._may_resolve({"action": "link", "corr_debug": {"corr_consensus": "rules_only", + "corr_path": "fast/single"}}) + assert not upload._may_resolve({"action": "link", "corr_debug": {"corr_consensus": "rules_only", + "corr_path": "fast/thin"}}) + assert not upload._may_resolve({"action": "link", "corr_debug": {"corr_consensus": "llm", + "corr_path": "llm"}}) + assert not upload._may_resolve({"action": "orphan", "corr_debug": {}}) + + +@pytest.mark.asyncio +async def test_quoted_false_does_not_resolve_or_blank_units(): + scene = {"tags": [], "incident_type": "police", "location": None, "location_coords": None, + "units": ["6-Adam"], "resolved": "false", "reassignment": "false"} + corr = AsyncMock(return_value=("inc-1", {"action": "new", "corr_debug": {}})) + with patch.object(upload, "_correlate", corr), patch.object(upload, "fstore") as fs: + fs.doc_set = AsyncMock() + await upload._correlate_scenes("call-1", "n", "sys-1", 383, "Ch 1", [scene]) + assert corr.await_args.kwargs["units"] == ["6-Adam"] + assert corr.await_args.kwargs["reassignment"] is False + fs.doc_set.assert_not_awaited() + + +# ── #190 ──────────────────────────────────────────────────────────────────── + +def test_severity_escalates_immediately_and_keeps_its_peak(): + out = ic._severity_after_link({"severity": "minor"}, "major", "shots fired") + assert out["severity"] == "major" and out["severity_peak"] == "major" + + +def test_calmer_followup_without_stand_down_does_not_deescalate(): + out = ic._severity_after_link({"severity": "major"}, "routine", "Engine 5 on scene, 10-4") + assert out["severity"] == "major" and "severity_deescalated_from" not in out + + +def test_explicit_stand_down_deescalates_but_peak_stays(): + inc = {"severity": "major"} + out = ic._severity_after_link(inc, "minor", "Only smoke showing, no flames.") + assert out == {"severity": "minor", "severity_peak": "major", "severity_deescalated_from": "major"} + later = ic._severity_after_link({**inc, **out}, "routine", "copy") + assert later["severity"] == "minor" and later["severity_peak"] == "major" + + +@pytest.mark.parametrize("text", [ + "call was unfounded", "false alarm, resetting the panel", "GOA, nothing found", + "Code 4, you can stand down", "fire is out", +]) +def test_stand_down_phrases(text): + assert ic._is_stand_down(text) + + +@pytest.mark.parametrize("text", ["disregard my last", "code 40", "flames showing", + "primary search all clear", "Code 4", None]) +def test_not_stand_down_phrases(text): + assert not ic._is_stand_down(text) + + +# ── #27 ───────────────────────────────────────────────────────────────────── + +def test_sweep_tombstones_only_on_the_last_visible_pass(): + scan = settings.recorrelation_scan_minutes + young = {"ended_at": (NOW - timedelta(minutes=20)).isoformat()} + last = {"ended_at": (NOW - timedelta(minutes=scan - 1)).isoformat()} + assert not recorrelation_sweep._last_chance(young, NOW) + assert recorrelation_sweep._last_chance(last, NOW) + assert recorrelation_sweep._last_chance({}, NOW) + + +@pytest.mark.asyncio +async def test_sweep_keeps_retrying_past_the_old_three_attempt_budget(): + call = {"call_id": "c", "started_at": (NOW - timedelta(minutes=30)).isoformat(), + "ended_at": (NOW - timedelta(minutes=30)).isoformat(), "corr_sweep_count": 5} + with patch("app.internal.incident_correlator.correlate_call", AsyncMock(return_value=None)), \ + patch.object(recorrelation_sweep, "fstore") as fs: + fs.doc_set = AsyncMock() + assert await recorrelation_sweep._recorrelate_orphan(call, NOW) is False + assert fs.doc_set.await_args.args[2] == {"corr_sweep_count": 6} diff --git a/drb-c2-core/tests/test_correlator_location_path.py b/drb-c2-core/tests/test_correlator_location_path.py new file mode 100644 index 0000000..8e96fc2 --- /dev/null +++ b/drb-c2-core/tests/test_correlator_location_path.py @@ -0,0 +1,141 @@ +""" +server-26#115 — location-path corroboration, plus the shared substance predicate. + +(The consensus LLM-orphan gate that used to be tested here was removed with the +cheap/smart tiebreak in server-26#299; its replacement is tested in +test_correlation_decision_flow.py.) + +The `location` correlation path linked on a bare sub-`location_proximity_km` +(0.5 km) distance alone, taking whichever incident came first in an unsorted +`recent`. 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, and picks the NEAREST qualifying candidate. A unit-overlap location +link is tagged `location_unit_overlap` so it does not merge into the fast path's +bucket in the admin fit-signal histogram. +""" +from datetime import datetime, timedelta, timezone + +import pytest + +from app.internal.incident_correlator import _run_decision, has_event_substance + +NOW = datetime(2026, 9, 7, 21, 30, 0, tzinfo=timezone.utc) + + +def test_has_event_substance_predicate(): + assert has_event_substance({"coords": {"lat": 1, "lng": 2}}) + assert has_event_substance({"tags": ["fire"]}) + assert has_event_substance({"call_vehicles": ["sedan"]}) + assert not has_event_substance({}) + assert not has_event_substance({"coords": None, "tags": [], "call_vehicles": []}) + # units and location are NOT substance — nearly every transmission has them. + assert not has_event_substance({"call_units": ["7-Adam"], "location": "Main St"}) + + +# ───────────────────────────────────────────────────────────────────────────── +# 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} +# ~0.28 km north — inside the 0.5 radius, outside the 0.2 tight bar; used as a +# second candidate that must lose the nearest-wins sort to NEAR_INC_COORDS. +MID_INC_COORDS = {"lat": 41.152500, "lng": -73.860000} + + +def _inc(incident_id, coords, units): + return { + "incident_id": incident_id, + "system_ids": ["sys-1"], + "talkgroup_ids": ["100"], # different TGID → fast path is a no-op + "location_coords": coords, + "units": units, + "tags": [], + "type": "police", + "updated_at": (NOW - timedelta(minutes=6)).isoformat(), + "started_at": (NOW - timedelta(minutes=20)).isoformat(), + "status": "active", + "call_ids": ["c0"], + } + + +def _loc_ctx(*, incidents, call_units): + return { + "call_id": "call-loc", + "all_active": list(incidents), + "recent": list(incidents), + "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_in_radius_but_no_unit_overlap_no_tight_proximity_does_not_link(caplog): + ctx = _loc_ctx( + incidents=[_inc("inc-loc", FAR_INC_COORDS, ["7-Adam"])], + call_units=["3-Boy"], + ) + with caplog.at_level("INFO", logger="drb-c2-core"): + decision = _run_decision(ctx) + # Reaches, and is rejected by, the new guard (not an earlier path). + assert "location-path skipped" in caplog.text + assert decision["action"] != "link" + assert (decision.get("corr_debug") or {}).get("corr_path") != "location" + + +def test_location_path_links_on_unit_overlap_with_distinct_fit_signal(): + ctx = _loc_ctx( + incidents=[_inc("inc-loc", FAR_INC_COORDS, ["5-Adam"])], + call_units=["5-Adam"], + ) + decision = _run_decision(ctx) + assert decision["action"] == "link" + assert decision["corr_debug"]["corr_path"] == "location" + # NOT "unit_overlap" — that value belongs to the fast path's histogram bucket. + assert decision["corr_debug"]["corr_fit_signal"] == "location_unit_overlap" + + +def test_location_path_links_on_tight_proximity_without_unit_overlap(): + ctx = _loc_ctx( + incidents=[_inc("inc-loc", NEAR_INC_COORDS, ["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" + + +def test_location_path_picks_nearest_in_radius_candidate(): + # `recent` order puts the farther tight-proximity incident first; the guard + # must still select the nearest one. + ctx = _loc_ctx( + incidents=[ + _inc("inc-mid", MID_INC_COORDS, ["3-Boy"]), # ~0.28 km, tight-fail + _inc("inc-near", NEAR_INC_COORDS, ["3-Boy"]), # ~0.13 km, tight-pass + ], + call_units=["3-Boy"], + ) + decision = _run_decision(ctx) + assert decision["action"] == "link" + assert decision["matched_incident"]["incident_id"] == "inc-near" + assert decision["corr_debug"]["corr_path"] == "location" diff --git a/drb-c2-core/tests/test_correlator_merge_caps.py b/drb-c2-core/tests/test_correlator_merge_caps.py index 2767ab9..cae8501 100644 --- a/drb-c2-core/tests/test_correlator_merge_caps.py +++ b/drb-c2-core/tests/test_correlator_merge_caps.py @@ -332,13 +332,18 @@ def test_duration_cap_blocks_the_thin_path_too(): assert _run_decision(_ctx(all_active=[inc], recent=[inc]))["action"] == "orphan" -def test_incident_just_under_the_duration_cap_still_accepts_calls(): +def test_incident_just_under_the_duration_cap_still_accepts_strong_evidence(): + # Past the weak-link span cap (server-26#189) shared units alone no longer + # link, but strong evidence (a shared vehicle) still does until the hard cap. inc = _long_running(settings.incident_max_duration_minutes - 10) inc["units"] = ["6-Adam"] + inc["vehicles"] = ["red sedan"] decision = _run_decision(_ctx( - all_active=[inc], recent=[inc], is_thin_call=False, call_units=["6-Adam"], + all_active=[inc], recent=[inc], is_thin_call=False, + call_units=["7-Boy"], call_vehicles=["red sedan"], )) assert decision["action"] == "link" + assert decision["corr_debug"]["corr_fit_signal"] == "vehicle_overlap" def test_call_count_cap_forces_a_new_incident(): diff --git a/drb-c2-core/tests/test_upload_short_transcript_correlation.py b/drb-c2-core/tests/test_upload_short_transcript_correlation.py index a70a3a3..059e394 100644 --- a/drb-c2-core/tests/test_upload_short_transcript_correlation.py +++ b/drb-c2-core/tests/test_upload_short_transcript_correlation.py @@ -29,7 +29,7 @@ async def _run_ingest(skip_reason): patch("app.internal.firestore.doc_get_cached", AsyncMock(return_value={"system_id": "sys-1", "ai_flags": {}})), \ patch.object(upload, "fstore") as fs, \ - patch.object(upload, "_correlate_with_consensus", AsyncMock(return_value=None)) as corr, \ + patch.object(upload, "_correlate", AsyncMock(return_value=(None, {}))) as corr, \ patch("app.internal.transcription.transcribe_call", AsyncMock(return_value=("10-8", []))), \ patch("app.internal.intelligence.extract_scenes", AsyncMock(return_value=[])), \