Logan CusanoandClaude Sonnet 5
fae84a45c3
Every scene of a multi-scene call correlates independently in upload.py's scene loop, but every scene's corr_debug was written flat onto the same shared call doc — scene 2's write silently clobbered scene 1's corr_path/corr_consensus/etc (#96), and summarizer.py read the whole call's raw transcript per linked call, mixing text from scenes the incident had nothing to do with, while ignoring transcript_corrected entirely (#114). Fix: thread a scene_index from both `for scene in scenes:` loops in upload.py down through _correlate_with_consensus -> incident_correlator.preview_correlation/correlate_call -> _build_context -> ctx["scene_index"]. incident_correlator._apply_and_log now writes, in the same Firestore call: - the existing flat corr_* fields, unchanged (last-scene-wins, the safe backward-compatible default for any reader that doesn't know about `scenes` yet) - a new nested `scenes.<scene_index>` entry with {transcript, incident_id, corr_debug}, via doc_set(..., merge=True). Firestore's DocumentReference.set(data, merge=True) recursively merges nested map fields by key (documented SDK behaviour, not assumed) — a write to scenes.1 merges alongside an existing scenes.0 instead of replacing the whole `scenes` map. scene_index defaults to 0 for every caller with no scene concept (the recorrelation sweep, the no-scenes-extracted orphan-check path), so a plain single-scene call still gets a one-entry `scenes` map equivalent to reading its flat fields today. admin.py's _call_summary exposes the new `scenes` list per call (each entry carrying the same corr_* field names as the flat fields, so the two shapes are interchangeable to the tally); the summary tally now iterates each call's scenes-if-present, else its own flat fields, so a 2-scene call with two different corr_path values counts as two data points instead of one blend. New `scene_decision_count` sits next to `linked_call_count` to make that distinction visible. summarizer.py's _scene_text_for_incident reads a linked call's `scenes` map to find the scene(s) whose corr_debug recorded a link into the specific incident being summarized, joining more than one if several scenes landed in the same incident. Falls back to transcript_corrected-or-transcript for a call doc with no `scenes` field (predates this change) — the one-liner half of #114, worth doing regardless since it stops raw-transcript summaries even for old-schema docs. Does not touch #80/#95/#102's existing ctx-threading fixes (embedding/severity/coords/LLM-prompt-transcript) — correct as-is, out of scope here. Tests: 14 new (test_per_scene_call_doc.py, test_summarizer_scene_transcript.py, additions to test_admin_debug_correlation.py) covering the merge shape, last-scene-wins flat-field backward compat, the admin tally's per-scene vs per-call counting (including old-schema fallback), and the summarizer's scene-specific text selection (including old-schema fallback). Full sandboxed suite: 364 -> 378 passed, all green. Fixes server-26#96, server-26#114 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix
568 lines
26 KiB
Python
568 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 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.
|
|
|
|
The window mirrors whatever the fast/thin path would use for this same
|
|
channel — `tg_dispatch_thin_idle_minutes` (5 min) on a dispatch backbone,
|
|
`tg_thin_idle_minutes` (15 min) on a tactical/working channel, selected via
|
|
the same `_is_dispatch_channel` test incident_correlator.py uses at its own
|
|
fast/thin idle-window selection (~:1005-1007). Using the dispatch constant
|
|
unconditionally would be wrong off dispatch — a retune of one for fast/thin
|
|
reasons would then silently widen or narrow this gate too, on channels
|
|
window #3 never measured.
|
|
|
|
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. A proper
|
|
fix needs a dedicated Firestore query that is not status/capacity
|
|
filtered — a new read, out of scope for this pass.
|
|
|
|
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, _is_dispatch_channel
|
|
|
|
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
|
|
if _is_dispatch_channel(ctx.get("talkgroup_name"))
|
|
else settings.tg_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
|
|
|
|
# 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,
|
|
)
|