Logan CusanoandClaude Sonnet 5
1ffff25cd2
intelligence.py skips GPT extraction for transcripts <=5 words ("10-8", "show
me clear", a unit check-in) -- real cost/hallucination guard, kept as-is. But
upload.py's no-scenes correlation fallback (the path that thin-links a call
by talkgroup even with zero extracted content) excluded ANY skip_reason,
including transcript_too_short -- so this exact population, brief but real
follow-up and clearance traffic, never even attempted to attach to anything.
Found live: "Live, Ossining." sitting an orphan 6 seconds before a real
incident's founding call, on the same talkgroup.
Now only garbage_transcript (Whisper hallucination, no real content) stays
excluded; transcript_too_short reaches the same thin/fast-path fallback
already trusted for no-transcript calls, gated the same way -- same-talkgroup,
recently-active incident required before anything attaches. No GPT re-invoked,
no new cost.
Verified: 410 pass, 0 fail.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
563 lines
26 KiB
Python
563 lines
26 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 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.
|
|
"""
|
|
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)
|
|
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
|
|
|
|
|
|
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(
|
|
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,
|
|
scene_index: int = 0,
|
|
) -> 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.
|
|
|
|
``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.
|
|
"""
|
|
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,
|
|
scene_index=scene_index,
|
|
)
|
|
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.
|
|
is_orphan_vs_new = llm_decision["action"] == "orphan" and rules_decision["action"] == "new"
|
|
substanceless, gate_veto_reason = _call_is_substanceless(ctx) if is_orphan_vs_new else (False, None)
|
|
if is_orphan_vs_new and substanceless:
|
|
logger.info(
|
|
f"Consensus gate for call {call_id}: llm=orphan vs rules=new and call "
|
|
f"is substanceless — resolving orphan, skipping tiebreak"
|
|
)
|
|
gated = {
|
|
"action": "orphan",
|
|
"matched_incident": None,
|
|
"incident_type": None,
|
|
"corr_debug": dict(rules_decision.get("corr_debug") or {}),
|
|
}
|
|
gated["corr_debug"].update({
|
|
"corr_consensus": "llm_orphan_gate",
|
|
"corr_rules_action": rules_decision["action"],
|
|
"corr_llm_action": llm_decision["action"],
|
|
"corr_llm_reasoning": llm_decision.get("reasoning", ""),
|
|
})
|
|
return await incident_correlator.apply_correlation({"decision": gated, "ctx": ctx})
|
|
|
|
# Disagree — escalate to the smarter tiebreaker.
|
|
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"]
|
|
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})
|
|
|
|
|
|
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.
|
|
# server-26#96: scene_index is threaded through so each scene's
|
|
# corr_debug/transcript lands in its own entry of the call doc's
|
|
# `scenes` map instead of clobbering every other scene's write.
|
|
for scene_index, scene in enumerate(scenes):
|
|
all_tags.extend(scene["tags"])
|
|
# When dispatch is pulling a unit to a NEW call (reassignment), suppress unit
|
|
# overlap so the new scene doesn't chain into the unit's previous incident.
|
|
is_reassignment = bool(scene.get("reassignment"))
|
|
corr_units = [] if is_reassignment else scene.get("units")
|
|
incident_id = await _correlate_with_consensus(
|
|
call_id=call_id,
|
|
node_id=node_id,
|
|
system_id=system_id,
|
|
talkgroup_id=talkgroup_id,
|
|
talkgroup_name=talkgroup_name,
|
|
tags=scene["tags"],
|
|
incident_type=scene["incident_type"],
|
|
location=scene["location"],
|
|
location_coords=scene["location_coords"],
|
|
units=corr_units,
|
|
vehicles=scene.get("vehicles"),
|
|
cleared_units=scene.get("cleared_units"),
|
|
reassignment=is_reassignment,
|
|
embedding=scene.get("embedding"),
|
|
severity=scene.get("severity"),
|
|
transcript=scene.get("transcript"),
|
|
scene_index=scene_index,
|
|
)
|
|
if incident_id and incident_id not in incident_ids:
|
|
incident_ids.append(incident_id)
|
|
if scene["resolved"] and incident_id:
|
|
await fstore.doc_set("incidents", incident_id, {
|
|
"status": "resolved",
|
|
"resolved_at": 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
|
|
|
|
# server-26#131: mark that real-time processing has started for this call
|
|
# BEFORE any of the slow steps below (STT, scene extraction, correlation).
|
|
# The re-correlation sweep (internal/recorrelation_sweep.py) scans for
|
|
# calls that still look orphaned within a wide window (recorrelation_scan_
|
|
# minutes, default 60) — with no guard here, a call whose real-time
|
|
# pipeline is still mid-flight (still transcribing, still waiting on a
|
|
# Gemini call) has no incident_id/corr_path written yet, so the sweep's
|
|
# orphan filter can't tell "never processed" from "processing right now"
|
|
# and correlates it a second time, independently, sometimes landing on a
|
|
# different incident than the real-time path — the exact duplicate-link
|
|
# bug #131 found (same call in two incidents' call_ids, ~2% of linked
|
|
# calls). Best-effort: a write failure here must not abort the pipeline.
|
|
try:
|
|
await fstore.doc_set("calls", call_id, {
|
|
"intelligence_started_at": datetime.now(timezone.utc).isoformat()
|
|
})
|
|
except Exception as e:
|
|
logger.warning(f"Could not mark intelligence_started_at for call {call_id}: {e}")
|
|
|
|
# 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"):
|
|
# server-26#96: scene_index is threaded through so each scene's
|
|
# corr_debug/transcript lands in its own entry of the call doc's
|
|
# `scenes` map instead of clobbering every other scene's write.
|
|
for scene_index, scene in enumerate(scenes):
|
|
all_tags.extend(scene["tags"])
|
|
is_reassignment = bool(scene.get("reassignment"))
|
|
corr_units = [] if is_reassignment else scene.get("units")
|
|
incident_id = await _correlate_with_consensus(
|
|
call_id=call_id,
|
|
node_id=node_id,
|
|
system_id=system_id,
|
|
talkgroup_id=talkgroup_id,
|
|
talkgroup_name=talkgroup_name,
|
|
tags=scene["tags"],
|
|
incident_type=scene["incident_type"],
|
|
location=scene["location"],
|
|
location_coords=scene["location_coords"],
|
|
units=corr_units,
|
|
vehicles=scene.get("vehicles"),
|
|
cleared_units=scene.get("cleared_units"),
|
|
reassignment=is_reassignment,
|
|
embedding=scene.get("embedding"),
|
|
severity=scene.get("severity"),
|
|
transcript=scene.get("transcript"),
|
|
scene_index=scene_index,
|
|
)
|
|
if incident_id and incident_id not in incident_ids:
|
|
incident_ids.append(incident_id)
|
|
if scene["resolved"] and incident_id:
|
|
await fstore.doc_set("incidents", incident_id, {
|
|
"status": "resolved",
|
|
"resolved_at": 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.
|
|
# transcript_too_short (<=5 words: "10-8", "show me clear", a unit
|
|
# check-in) still carries a real transcript and talkgroup — exactly the
|
|
# brief follow-up/clearance traffic an incident needs, and the thin-path
|
|
# merge below already requires a same-talkgroup, recently-active
|
|
# incident before attaching anything, same guard already trusted for
|
|
# no-transcript calls. Previously excluded here, so these calls never
|
|
# attached to anything at all. garbage_transcript (Whisper
|
|
# hallucination) has no real content behind it and stays excluded.
|
|
if not scenes:
|
|
_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(
|
|
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,
|
|
)
|