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, )