Logan CusanoandClaude Sonnet 5
400b74b519
_recent_incident_on_same_talkgroup previously treated ANY same-talkgroup incident within the 2-hour correlation_window_hours lookback as 'recent', which disabled the whole LLM-orphan consensus gate on busy dispatch channels: window #3 (CORRELATION_REVIEW_0912.md) measured 0/24 gate fires against the exact target shape (rules=new, llm=orphan, tiebreak=new), with 22/24 explained by a same-talkgroup incident existing somewhere in the prior 2h — nearly guaranteed on channels producing 3-13 incidents/2h. Now the escape hatch only counts an incident as recent within settings.tg_dispatch_thin_idle_minutes (5 min), reusing the same recency bound the fast/thin path already uses for the 'dispatch, thin ack 10-30s later' case this hatch exists for, instead of inventing a new constant. Investigated the 2 unexplained misses (no same-tg incident found even by a naive full-collection timestamp scan): confirmed ctx["recent"] is built from status=="active" incidents with over-capacity incidents dropped (_build_context / _drop_capped), not a full collection scan — an incident that has auto-resolved or hit incident_max_calls/incident_max_duration within the window is invisible to this check even though it is chronologically recent. This does not explain the 2 misses (a same-tg incident was absent by both checks there, so some other _call_is_substanceless condition must be responsible), but it is a real gap in the check as written. Documented in the docstring with a TODO(server-26#115); fixing it needs a new, non-active-filtered Firestore query, out of scope for this pass. Tests: added a regression test proving an incident inside the old 2h window but outside the new 5-minute window now correctly gates (fails on main, passes here), plus a test proving a truly recent (<5min) same-tg incident still escapes the gate as intended. Sandboxed pytest: 327 -> 329 passed (2 new tests), all green. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix
530 lines
24 KiB
Python
530 lines
24 KiB
Python
import secrets
|
|
from typing import Optional
|
|
from datetime import datetime, timezone
|
|
from fastapi import APIRouter, BackgroundTasks, UploadFile, File, Form, HTTPException, Security
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from app.internal.storage import upload_audio
|
|
from app.internal import dedup
|
|
from app.internal import firestore as fstore
|
|
from app.internal.logger import logger
|
|
from app.config import settings
|
|
|
|
router = APIRouter(tags=["upload"])
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
@router.post("/upload")
|
|
async def upload_call_audio(
|
|
background_tasks: BackgroundTasks,
|
|
file: UploadFile = File(...),
|
|
call_id: str = Form(...),
|
|
node_id: str = Form(...),
|
|
talkgroup_id: Optional[int] = Form(None),
|
|
talkgroup_name: Optional[str] = Form(None),
|
|
system_id: Optional[str] = Form(None),
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Security(_bearer),
|
|
):
|
|
"""
|
|
Receive an audio recording from an edge node.
|
|
Upload to GCS, update the call document in Firestore with the audio URL,
|
|
then kick off the intelligence pipeline as a background task.
|
|
"""
|
|
# Verify the per-node API key
|
|
if not credentials:
|
|
raise HTTPException(401, "Missing authorization")
|
|
key_doc = await fstore.doc_get("node_keys", node_id)
|
|
if not key_doc:
|
|
logger.warning(f"Upload 401: no key_doc in Firestore for node_id={node_id!r}")
|
|
raise HTTPException(401, "Invalid node API key")
|
|
# compare_digest, not !=, so the comparison cost does not depend on how many
|
|
# leading characters matched. enrollment.py and dynsec.py were explicit about
|
|
# this for the same class of credential; this route was the odd one out.
|
|
stored_key = key_doc.get("api_key") or ""
|
|
if not secrets.compare_digest(stored_key, credentials.credentials):
|
|
logger.warning(
|
|
f"Upload 401: key mismatch for node_id={node_id!r} "
|
|
f"(received prefix: {credentials.credentials[:8]}...)"
|
|
)
|
|
raise HTTPException(401, "Invalid node API key")
|
|
|
|
data = await file.read()
|
|
if not data:
|
|
raise HTTPException(400, "Empty file.")
|
|
if len(data) > settings.upload_max_bytes:
|
|
raise HTTPException(413, f"File too large (max {settings.upload_max_bytes // (1024*1024)} MB).")
|
|
|
|
gcs_uri = await upload_audio(data, file.filename or "", call_id=call_id)
|
|
|
|
if gcs_uri:
|
|
try:
|
|
# Canonical object location only. The playback link is minted per
|
|
# read in storage.playback_url() — nothing durable is stored here.
|
|
# org_id is stamped defensively here too (not just in
|
|
# mqtt_handler.py's call_start/call_end): key_doc above proves this
|
|
# node_id is real and authenticated, so resolving org_id from the
|
|
# node doc here covers a call whose Firestore doc was somehow
|
|
# never written by call_start (the upload is otherwise the
|
|
# authoritative record of which node this audio came from).
|
|
node = await fstore.doc_get_cached("nodes", node_id)
|
|
updates = {"audio_gcs_uri": gcs_uri}
|
|
if node and node.get("org_id"):
|
|
updates["org_id"] = node["org_id"]
|
|
await fstore.doc_set("calls", call_id, updates)
|
|
except Exception as e:
|
|
logger.warning(f"Could not update call {call_id} with audio_gcs_uri: {e}")
|
|
|
|
# Another node in range recorded the same transmission. Keep the audio
|
|
# (it may be the cleaner capture) but don't transcribe or correlate it
|
|
# a second time — see app/internal/dedup.py.
|
|
call_doc = await fstore.doc_get("calls", call_id)
|
|
duplicate_of = await dedup.find_duplicate_of(call_doc) if call_doc else None
|
|
if duplicate_of:
|
|
await fstore.doc_set("calls", call_id, {"duplicate_of": duplicate_of})
|
|
logger.info(
|
|
f"Call {call_id} from {node_id} duplicates {duplicate_of} "
|
|
f"— audio kept, AI pipeline skipped."
|
|
)
|
|
return {"url": gcs_uri, "duplicate_of": duplicate_of}
|
|
|
|
background_tasks.add_task(
|
|
_run_intelligence_pipeline,
|
|
call_id=call_id,
|
|
node_id=node_id,
|
|
system_id=system_id,
|
|
talkgroup_id=talkgroup_id,
|
|
talkgroup_name=talkgroup_name,
|
|
gcs_uri=gcs_uri,
|
|
)
|
|
|
|
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.
|
|
|
|
|
|
def _recent_incident_on_same_talkgroup(ctx: dict) -> bool:
|
|
"""
|
|
True when one of the already-loaded recent incidents is running on this
|
|
call's own system + talkgroup AND was active within the last
|
|
`settings.tg_dispatch_thin_idle_minutes` minutes. Covers the "unit
|
|
dispatched on the dispatch channel, thin acknowledgement 10-30s later"
|
|
case: the ack carries no substance of its own but plainly belongs to the
|
|
job just opened.
|
|
|
|
That window intentionally reuses `tg_dispatch_thin_idle_minutes` (5 min)
|
|
rather than inventing a new constant — it's the same recency bound the
|
|
fast/thin path already uses for this exact "dispatch, thin ack" scenario
|
|
(see its tuning note above in config.py), so both places agree on what
|
|
"just happened on this channel" means.
|
|
|
|
This used to be a plain "does any recent incident exist on this
|
|
talkgroup" check against a 2-hour window (`correlation_window_hours`).
|
|
Measured live in production (server-26#115, CORRELATION_REVIEW_0912.md,
|
|
window #3): on a busy dispatch channel producing 3-13 incidents per 2h,
|
|
that condition is satisfied almost unconditionally, so the surrounding
|
|
LLM-orphan gate never fired on exactly the channels it exists to
|
|
protect (0/24 target-shaped calls gated in a 4h window). The docstring's
|
|
own intent was always "10-30 seconds", not "hours" — a few minutes is
|
|
the right shape.
|
|
|
|
Reads ctx["recent"] — the same window-filtered candidate list the rules
|
|
engine already loaded — so this adds no Firestore read.
|
|
|
|
Known limitation (server-26#115): ctx["recent"] is derived from
|
|
`all_active` in `_build_context` — incidents with `status=="active"`
|
|
for the call's org, with over-capacity incidents already dropped by
|
|
`_drop_capped` — not a full scan of the `incidents` collection. A
|
|
same-talkgroup incident that has already auto-resolved (no longer
|
|
"active") or hit `incident_max_calls`/`incident_max_duration_minutes`
|
|
will NOT appear here even though it is chronologically recent. This is
|
|
the confirmed explanation for 2/24 gate misses in the window #3
|
|
measurement where a naive full-collection timestamp scan found no
|
|
same-talkgroup candidate either once status/capacity are accounted for
|
|
— i.e. the check here was already correct for those two calls; some
|
|
other `_call_is_substanceless` condition (severity/substance/type) must
|
|
have been true instead. A proper fix for the truncation case (an
|
|
incident that WAS same-talkgroup-recent by clock time but is invisible
|
|
here because it resolved or capped) needs a dedicated Firestore query
|
|
that is not status/capacity filtered — a new read, out of scope for
|
|
this pass.
|
|
# TODO(server-26#115): add a talkgroup-scoped incident lookup (any
|
|
# status, no capacity filter) if this escape hatch ever needs to see
|
|
# resolved/capped incidents rather than just the active candidate pool.
|
|
"""
|
|
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 datetime.now(timezone.utc)
|
|
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) <= settings.tg_dispatch_thin_idle_minutes:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _call_is_substanceless(ctx: dict) -> bool:
|
|
"""
|
|
True when the call carries nothing that marks it as a real event:
|
|
• 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.
|
|
"""
|
|
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") or ctx.get("reassignment"):
|
|
return False
|
|
if (ctx.get("call_severity") or "routine") in ("moderate", "major"):
|
|
return False
|
|
if incident_correlator.has_event_substance(ctx):
|
|
return False
|
|
if _recent_incident_on_same_talkgroup(ctx):
|
|
return False
|
|
return True
|
|
|
|
|
|
async def _correlate_with_consensus(
|
|
call_id: str,
|
|
node_id: str,
|
|
system_id: Optional[str],
|
|
talkgroup_id: Optional[int],
|
|
talkgroup_name: Optional[str],
|
|
tags: list[str],
|
|
incident_type: Optional[str],
|
|
location: Optional[str],
|
|
location_coords: Optional[dict],
|
|
units: Optional[list] = None,
|
|
vehicles: Optional[list] = None,
|
|
cleared_units: Optional[list] = None,
|
|
reassignment: bool = False,
|
|
embedding: Optional[list] = None,
|
|
severity: Optional[str] = None,
|
|
transcript: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
"""
|
|
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.
|
|
|
|
Falls back to rules-only when GEMINI_API_KEY is absent, the call is
|
|
content-free (thin), or any LLM call fails.
|
|
"""
|
|
from app.internal import incident_correlator, llm_correlator
|
|
|
|
preview = await incident_correlator.preview_correlation(
|
|
call_id=call_id, node_id=node_id, system_id=system_id,
|
|
talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name,
|
|
tags=tags, incident_type=incident_type, location=location,
|
|
location_coords=location_coords, units=units, vehicles=vehicles,
|
|
cleared_units=cleared_units, reassignment=reassignment,
|
|
embedding=embedding, severity=severity, transcript=transcript,
|
|
)
|
|
ctx = preview["ctx"]
|
|
rules_decision = preview["decision"]
|
|
|
|
llm_decision = await llm_correlator.decide(call_id, ctx)
|
|
|
|
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.
|
|
if (
|
|
llm_decision["action"] == "orphan"
|
|
and rules_decision["action"] == "new"
|
|
and _call_is_substanceless(ctx)
|
|
):
|
|
logger.info(
|
|
f"Consensus gate for call {call_id}: llm=orphan vs rules=new and call "
|
|
f"is substanceless — resolving orphan, skipping tiebreak"
|
|
)
|
|
gated = {
|
|
"action": "orphan",
|
|
"matched_incident": None,
|
|
"incident_type": None,
|
|
"corr_debug": dict(rules_decision.get("corr_debug") or {}),
|
|
}
|
|
gated["corr_debug"].update({
|
|
"corr_consensus": "llm_orphan_gate",
|
|
"corr_rules_action": rules_decision["action"],
|
|
"corr_llm_action": llm_decision["action"],
|
|
"corr_llm_reasoning": llm_decision.get("reasoning", ""),
|
|
})
|
|
return await incident_correlator.apply_correlation({"decision": gated, "ctx": ctx})
|
|
|
|
# Disagree — escalate to the smarter tiebreaker.
|
|
logger.info(
|
|
f"Consensus disagreement for call {call_id}: "
|
|
f"rules={rules_decision['action']} vs llm={llm_decision['action']} — tiebreak"
|
|
)
|
|
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"]
|
|
return await incident_correlator.apply_correlation({"decision": final, "ctx": ctx})
|
|
|
|
|
|
async def _resolve_flags(system_id: Optional[str]):
|
|
"""
|
|
Resolve AI feature flags for a given system.
|
|
|
|
Thin alias for `feature_flags.resolve_flags` — the resolver lives there
|
|
because transcription and the calls router need the same answer, and three
|
|
copies of it is how server-26#75 happened in the first place.
|
|
"""
|
|
from app.internal.feature_flags import resolve_flags
|
|
|
|
return await resolve_flags(system_id)
|
|
|
|
|
|
async def _run_extraction_pipeline(
|
|
call_id: str,
|
|
node_id: str,
|
|
system_id: Optional[str],
|
|
talkgroup_id: Optional[int],
|
|
talkgroup_name: Optional[str],
|
|
transcript: str,
|
|
segments: Optional[list] = None,
|
|
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
|
|
|
|
flags, _flag = await _resolve_flags(system_id)
|
|
|
|
incident_ids: list[str] = []
|
|
all_tags: list[str] = []
|
|
|
|
if _flag("correlation_enabled"):
|
|
# Step 2: Scene detection + intelligence extraction.
|
|
# Returns one scene per distinct incident detected in the recording.
|
|
scenes = await intelligence.extract_scenes(
|
|
call_id, transcript, talkgroup_name,
|
|
talkgroup_id=talkgroup_id, system_id=system_id, segments=segments,
|
|
node_id=node_id,
|
|
preserve_transcript_correction=preserve_transcript_correction,
|
|
)
|
|
|
|
# Step 3: Correlate each scene to an incident independently.
|
|
for scene in 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"),
|
|
)
|
|
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": datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
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)")
|
|
|
|
if incident_ids:
|
|
await fstore.doc_set("calls", call_id, {"incident_ids": incident_ids})
|
|
|
|
# Step 4: Alert dispatch — run once with merged tags from all scenes.
|
|
await alerter.check_and_dispatch(
|
|
call_id=call_id,
|
|
node_id=node_id,
|
|
talkgroup_id=talkgroup_id,
|
|
talkgroup_name=talkgroup_name,
|
|
tags=list(dict.fromkeys(all_tags)),
|
|
transcript=transcript,
|
|
)
|
|
|
|
|
|
async def _run_intelligence_pipeline(
|
|
call_id: str,
|
|
node_id: str,
|
|
system_id: Optional[str],
|
|
talkgroup_id: Optional[int],
|
|
talkgroup_name: Optional[str],
|
|
gcs_uri: Optional[str],
|
|
) -> None:
|
|
"""
|
|
Post-upload intelligence pipeline (runs as a background task):
|
|
1. Transcribe audio via Google STT
|
|
2. Detect scenes + extract intelligence (one result per incident in recording)
|
|
3. Correlate each scene with existing incidents (or create new ones)
|
|
4. Check alert rules and dispatch notifications
|
|
"""
|
|
from app.internal import transcription, intelligence, incident_correlator, alerter, talkgroups
|
|
|
|
# The node only sends talkgroup_name when OP25 had it in the loaded tags
|
|
# file, so it arrives empty for exactly the talkgroups C2 can name from the
|
|
# system config. Resolve it once, here, at the single funnel both /upload
|
|
# and /calls/{id}/reprocess pass through — everything downstream (the
|
|
# dispatch-channel test, scene extraction, and the incident title) then
|
|
# gets a real name instead of "TGID 9048". server-26#34.
|
|
_call_doc = await fstore.doc_get("calls", call_id)
|
|
talkgroup_name = await talkgroups.resolve(
|
|
system_id, talkgroup_id, hint=talkgroup_name, call_doc=_call_doc,
|
|
)
|
|
# Backfill the call document too, so the archive and the orphan panel stop
|
|
# showing a bare TGID for a channel we can now name.
|
|
if talkgroup_name and _call_doc is not None and not _call_doc.get("talkgroup_name"):
|
|
try:
|
|
await fstore.doc_set("calls", call_id, {"talkgroup_name": talkgroup_name})
|
|
except Exception as e:
|
|
logger.warning(f"Could not backfill talkgroup_name on call {call_id}: {e}")
|
|
|
|
flags, _flag = await _resolve_flags(system_id)
|
|
|
|
transcript: Optional[str] = None
|
|
segments: list[dict] = []
|
|
|
|
# Step 1: Transcription
|
|
if gcs_uri:
|
|
if _flag("stt_enabled"):
|
|
transcript, segments = await transcription.transcribe_call(
|
|
call_id, gcs_uri, talkgroup_name,
|
|
system_id=system_id, talkgroup_id=talkgroup_id,
|
|
)
|
|
else:
|
|
scope = "globally" if not flags["stt_enabled"] else f"system {system_id}"
|
|
logger.info(f"STT disabled ({scope}) — skipping transcription for call {call_id}")
|
|
|
|
# Step 2: Scene detection + intelligence extraction
|
|
scenes: list[dict] = []
|
|
if _flag("correlation_enabled"):
|
|
if transcript:
|
|
scenes = await intelligence.extract_scenes(
|
|
call_id, transcript, talkgroup_name,
|
|
talkgroup_id=talkgroup_id, system_id=system_id, segments=segments,
|
|
node_id=node_id,
|
|
)
|
|
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}")
|
|
|
|
# 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] = []
|
|
if _flag("correlation_enabled"):
|
|
for scene in 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"),
|
|
)
|
|
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": datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
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.
|
|
# Skip when extraction flagged the call — garbage or too-short transcripts
|
|
# carry no signal and would only attach spuriously via the thin path.
|
|
if not scenes:
|
|
_call_doc = await fstore.doc_get("calls", call_id)
|
|
if not (_call_doc or {}).get("skip_reason"):
|
|
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=[],
|
|
incident_type=None,
|
|
location=None,
|
|
location_coords=None,
|
|
)
|
|
if incident_id:
|
|
incident_ids.append(incident_id)
|
|
|
|
if incident_ids:
|
|
await fstore.doc_set("calls", call_id, {"incident_ids": incident_ids})
|
|
|
|
# Step 4: Alert dispatch (always runs — talkgroup ID rules don't need a transcript)
|
|
await alerter.check_and_dispatch(
|
|
call_id=call_id,
|
|
node_id=node_id,
|
|
talkgroup_id=talkgroup_id,
|
|
talkgroup_name=talkgroup_name,
|
|
tags=list(dict.fromkeys(all_tags)),
|
|
transcript=transcript,
|
|
)
|