logan/server-26
1
0
Fork 0
Code Issues 84 Pull Requests Actions Packages Projects Releases Wiki Activity

correlation: rules decide, one cheap LLM call only when ambiguous; drop the consensus tiebreak (server-26#299)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Logan Cusano
2026-10-27 20:26:38 -04:00
co-authored by Claude Opus 5.5
parent f4d78aba77
commit 3246fc97b0
18 changed files with 1207 additions and 1062 deletions
Download Patch File Download Diff File Expand all files Collapse all files
drb-c2-core/app/config.py
+39 -16
View File
@@ -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 ──────
drb-c2-core/app/internal/ai_health.py
+1 -1
View File
@@ -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.
drb-c2-core/app/internal/chatter_classifier.py
+2 -2
View File
@@ -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
drb-c2-core/app/internal/incident_correlator.py
+346 -151
View File
@@ -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"],
# 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 uc_dist > settings.location_proximity_km:
location_conflict = True
if not uc_fit:
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']}"
f"Correlator unit-continuity skipped: call {call_id} vs "
f"{best_unit_inc['incident_id']} — {uc_signal}"
)
if not reassigned_away and not location_conflict:
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",
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: call {call_id} has no incident_type — opening a "
f"{resolved_type!r} incident (severity={call_severity}, tags={tags})"
)
decision = {
"action": action,
"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:
logger.info(
f"Correlator: inferred incident_type={resolved_type!r} from tags {tags} for call {call_id}"
)
# Severity, not type, decides whether a call is incident-worthy.
#
# Requiring a concrete incident_type here meant a channel whose traffic never
# classifies — transit/rail administration, records lookups, prisoner
# transports — could never open a SECOND incident. Every later call on that
# talkgroup then funnelled into whichever incident happened to be created
# first, producing hour-long incidents made of unrelated transmissions
# (2026-08-16: TG 9048, 28 calls / 49 min) alongside 30+ permanent orphans.
#
# Anything the extractor judged a real event, or that carries any concrete
# content, now opens an incident under the neutral "other" type. Only
# content-free routine traffic is still left for the thin path to attach.
#
# `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,
"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,
drb-c2-core/app/internal/llm_correlator.py
+62 -143
View File
@@ -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:
def _build_decide_prompt(ctx: dict, candidates: list[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)"
)
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")
)
_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
drb-c2-core/app/internal/recorrelation_sweep.py
+26 -26
View File
@@ -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
drb-c2-core/app/internal/replay.py
+7 -2
View File
@@ -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,
drb-c2-core/app/routers/admin.py
+16 -18
View File
@@ -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`
drb-c2-core/app/routers/upload.py
+171 -237
View File
@@ -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
@@ -248,67 +208,118 @@ async def _correlate_with_consensus(
scene_index=scene_index,
)
ctx = preview["ctx"]
rules_decision = preview["decision"]
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", ""),
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"],
})
return await incident_correlator.apply_correlation({"decision": gated, "ctx": ctx})
# Disagree — escalate to the smarter tiebreaker.
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,
incident_ids, all_tags = await _correlate_scenes(
call_id, node_id, system_id, talkgroup_id, talkgroup_name, scenes,
)
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)")
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,
incident_ids, all_tags = await _correlate_scenes(
call_id, node_id, system_id, talkgroup_id, talkgroup_name, scenes,
)
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)")
# 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,
drb-c2-core/tests/test_ai_flag_gating.py
+1 -1
View File
@@ -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, \
drb-c2-core/tests/test_ai_health.py
+3 -3
View File
@@ -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
drb-c2-core/tests/test_ai_timeouts.py
+7 -18
View File
@@ -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():
drb-c2-core/tests/test_clearance_tracking.py
+3 -4
View File
@@ -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():
drb-c2-core/tests/test_consensus_gate.py
-430
View File
@@ -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"
drb-c2-core/tests/test_correlation_decision_flow.py
+367
View File
@@ -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}
drb-c2-core/tests/test_correlator_location_path.py
+141
View File
@@ -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"
drb-c2-core/tests/test_correlator_merge_caps.py
+7 -2
View File
@@ -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():
drb-c2-core/tests/test_upload_short_transcript_correlation.py
+1 -1
View File
@@ -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=[])), \