Logan CusanoandClaude Sonnet 5
8dd636af8f
server-26#131: same call_id ends up in TWO incidents' call_ids, byte-identical extracted data, ~2% of linked calls across 3 live dumps. Root cause: the sweep's orphan filter (incident_id/incident_ids/corr_path all absent) can't tell 'never processed' apart from 'real-time pipeline is still mid-flight' -- a call whose STT/scene-extraction/correlation chain (routers/upload.py _run_intelligence_pipeline) hasn't finished yet has none of those fields set, so the sweep picks it up and correlates it independently, sometimes onto a different incident than the real-time path lands on. Confirmed in review: _update_incident/_create_incident append call_id to an incident's call_ids unconditionally, with no cross-incident dedup guard -- preventing the second correlation attempt is the only lever available at this layer. Fix: _run_intelligence_pipeline marks intelligence_started_at on the call doc before any slow step; the sweep holds back any call whose marker is under 15 minutes old (raised from an initial 5 -- see below), regardless of how orphaned it otherwise looks. No marker at all (pre-#131 call doc, or the marker write itself failed) is not held back -- absence isn't evidence of an in-flight pipeline, and that's #131's own pre-existing population. Also covers the /calls/{id}/reprocess path, which calls the same _run_intelligence_pipeline. 15 min, not 5: neither the OpenAI Whisper client nor the Gemini call in llm_correlator.py sets a request timeout (filed server-26#153), so 5 min was a guess against an unbounded tail -- drb-correlation-review flagged this. Raising it is free on the recovery side: a call that finished processing (linked or genuinely orphaned) always has corr_path set and is already excluded by the sweep's other filter, so this constant only ever delays calls that are still actually running. DEFERRED.md row 52 (outside this repo, Version 5C root) updated to flag its ~6 min timing figure as stale. recorrelation_sweep.py had zero test coverage before this. New file covers the guard function's boundary (age < threshold vs exactly-at vs old vs missing vs unparseable) and one integration-shaped test proving a racing call never reaches correlate_call while a genuinely-orphaned call still does. Sandboxed pytest: 381 -> 387. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix
556 lines
26 KiB
Python
556 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.
|
|
# 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,
|
|
)
|