import secrets from datetime import datetime, timedelta, timezone from typing import Optional 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 radios from app.internal import firestore as fstore from app.internal import clock 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), radios_json: Optional[str] = Form(None, alias="radios"), 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") # server-26#195: a valid key is not enough, the node must still be # approved. reject_node() deletes node_keys too, but this closes the # window where that delete failed or was never run (a node rejected before # reject revoked anything). Uncached read: revocation must bite on the # next upload, not after doc_get_cached's 5-minute TTL. node = await fstore.doc_get("nodes", node_id) if not node or node.get("approval_status") != "approved": logger.warning( f"Upload 403: node_id={node_id!r} has a valid key but approval_status=" f"{(node or {}).get('approval_status')!r}" ) raise HTTPException(403, "Node is not approved") 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).") # server-26#302: the edge node retries uploads (spooled for up to 72h), and # a retry arrives whenever the node lost the response to an upload C2 had # already stored. call_id is the idempotency key: audio already stored # means this upload was already processed, so # re-running Whisper/extraction/correlation would double the AI spend and # double-link the call (the server-26#131 shape). Not atomic: two retries # racing inside one request's lifetime can both pass this check. # Keyed on stored audio only: the pipeline marker alone would refuse a # retry forever (and hand the node a null URL to keep retrying) after a # pipeline crash; such a call is recovered by the sweep or a reprocess. existing = await fstore.doc_get("calls", call_id) or {} if existing.get("audio_gcs_uri"): logger.info( f"Upload for call {call_id} from {node_id} is a retry of one already " f"processed — returning the stored audio, pipeline not re-run" ) return {"url": existing.get("audio_gcs_uri"), "already_uploaded": True} 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 (fetched by the approval gate above) 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). updates = {"audio_gcs_uri": gcs_uri} if node and node.get("org_id"): updates["org_id"] = node["org_id"] # Every radio heard on the call (server-26#304). Written here, not # only on call_end, because the pipeline below starts before the # node publishes call_end. radio_fields = radios.call_fields(radios_json) updates.update(radio_fields) if radio_fields and existing.get("srcaddr") is None: # Same meaning call_end gives it: the latest talker. updates["srcaddr"] = radio_fields["radios"][-1]["radio"] 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} # New-event signals: an LLM "orphan" may not drop a call carrying one when the # rules would open an incident for it (server-26#299). A missed real event is # worse than one extra incident, and the sweep is link-only, so a dropped event # never comes back. These are the old #115 gate's vetoes, which existed because # a looser gate dropped typed and geocoded fire calls. _LLM_ORPHAN_VETO_SEVERITIES = ("moderate", "major") def _llm_orphan_vetoed(ctx: dict) -> bool: from app.internal.incident_correlator import has_event_substance return bool( ctx.get("reassignment") or ctx.get("incident_type") or ctx.get("call_severity") in _LLM_ORPHAN_VETO_SEVERITIES or has_event_substance(ctx) ) def _merge_llm_decision(rules: dict, llm: dict, ctx: dict) -> tuple[dict, str]: """ The decision to commit for an ambiguous scene, and the corr_consensus value that says who made it: "llm", or "rules_guard" when the LLM's orphan was refused for a call with a strong new-event signal. """ if llm["action"] == "orphan" and rules["action"] == "new" and _llm_orphan_vetoed(ctx): return rules, "rules_guard" if ( llm["action"] == rules["action"] == "link" and llm["matched_incident"]["incident_id"] == rules["matched_incident"]["incident_id"] ): # The LLM confirmed the rules' own link: keep the rules' path/fit # signal so corr_path histograms stay comparable across runs. return rules, "llm" final = { "action": llm["action"], "matched_incident": llm["matched_incident"], "incident_type": llm["incident_type"], "corr_debug": dict(llm.get("corr_debug") or {}), } if llm["action"] == "link": # The rules' corr_path/fit signal described a different judgement # (possibly a different incident); keep it under corr_rules_* only. final["corr_debug"]["corr_path"] = "llm" elif llm["action"] == "orphan" and rules["action"] == "link": # Tombstone: the rules-only sweep would otherwise re-commit the very # unverified unit link the LLM just refused. final["corr_debug"]["corr_path"] = "llm_orphan" return final, "llm" async def _correlate( 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, ) -> tuple[Optional[str], dict]: """ Correlate one scene (server-26#299): the rules engine decides; only when it marks its decision ambiguous is ONE cheap LLM call made, and a usable LLM answer is committed instead. Returns (incident_id, committed decision). corr_consensus on the call doc says who decided: "rules_only" — confident rules decision, no LLM call "llm" — ambiguous, the LLM's answer was committed "rules_fallback" — ambiguous, LLM unavailable/failed/unusable → rules "rules_guard" — ambiguous, LLM said orphan on a call with a strong new-event signal → rules "new" kept and corr_band names why it was ambiguous. ``scene_index`` (server-26#96) — which scene of the call this is, so the call doc's per-scene ``scenes`` map records this scene's own decision. """ 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 = preview["decision"] band = rules.get("ambiguous") if not band: rules["corr_debug"]["corr_consensus"] = "rules_only" return await incident_correlator.apply_correlation(preview), rules llm = await llm_correlator.decide(call_id, ctx) if llm is None: final, consensus = rules, "rules_fallback" else: final, consensus = _merge_llm_decision(rules, llm, ctx) final["corr_debug"].update({ "corr_consensus": consensus, "corr_band": band, "corr_rules_action": rules["action"], }) if llm is not None: final["corr_debug"]["corr_llm_action"] = llm["action"] final["corr_debug"]["corr_llm_reasoning"] = llm.get("reasoning", "") logger.info( f"Correlation for call {call_id} scene {scene_index}: ambiguous ({band}) — " f"rules={rules['action']} llm={llm['action'] if llm else 'none'} → {consensus}" ) return await incident_correlator.apply_correlation({"decision": final, "ctx": ctx}), final def _may_resolve(decision: dict) -> bool: """ server-26#88: honour a scene's `resolved` flag only on a link the correlator was confident about, or on the incident this very scene opened. A scene that attached by thin recency, or whose placement was ambiguous (an LLM pick, or a rules fallback in the ambiguous band), has no business closing an incident it may not belong to — that closed live incidents while the event was still on the air. """ if decision.get("action") == "new": return True debug = decision.get("corr_debug") or {} return ( decision.get("action") == "link" and debug.get("corr_consensus") == "rules_only" and debug.get("corr_path") != "fast/thin" ) async def _correlate_scenes( call_id: str, node_id: str, system_id: Optional[str], talkgroup_id: Optional[int], talkgroup_name: Optional[str], scenes: list[dict], ) -> tuple[list[str], list[str]]: """ Correlate each scene of one call independently — a single recording can carry several events on a busy channel — and apply each scene's LLM closure. Returns (incident_ids, merged tags). Shared by the live pipeline, replay (via _extract_and_correlate) and the transcript-PATCH path. """ from app.internal import incident_correlator incident_ids: list[str] = [] all_tags: list[str] = [] 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. parse_llm_bool (#88): a quoted "false" is not True. is_reassignment = incident_correlator.parse_llm_bool(scene.get("reassignment")) corr_units = [] if is_reassignment else scene.get("units") incident_id, decision = await _correlate( 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 incident_id and incident_correlator.parse_llm_bool(scene.get("resolved")): if not _may_resolve(decision): logger.info( f"Ignoring closure from call {call_id} scene {scene_index}: link into " f"{incident_id} was not confident (server-26#88)" ) continue await fstore.doc_set("incidents", incident_id, { "status": "resolved", "resolved_at": clock.now().isoformat(), "resolved_via": "llm_closure", # One transmission read as "it's over" ("transport complete") # closed the whole 09-22 bridge MVA at 14:44 and its next 56 # calls opened a second incident (server-26#170). Inferred from # a single call, so provisional, like a timer close. "reopenable": True, }) await incident_correlator.maybe_resolve_parent(incident_id) logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)") return incident_ids, all_tags 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, 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. incident_ids, all_tags = await _correlate_scenes( call_id, node_id, system_id, talkgroup_id, talkgroup_name, scenes, ) 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 _extract_and_correlate( call_id: str, node_id: str, system_id: Optional[str], talkgroup_id: Optional[int], talkgroup_name: Optional[str], transcript: Optional[str], segments: Optional[list[dict]] = None, scenes: Optional[list[dict]] = None, ) -> tuple[list[str], list[str], list[dict]]: """ Steps 2-3 of the intelligence pipeline for one call: scene extraction (skipped when `scenes` is passed in), then per-scene correlation, then the no-scene thin fallback. Returns (incident_ids, merged tags, scenes). Shared by the live pipeline below and by replay (app/internal/replay.py), so a replay run measures exactly the code that runs live rather than a copy of it that can drift. Caller owns the correlation feature-flag check and alerting. """ from app.internal import intelligence # Step 2: Scene detection + intelligence extraction if scenes is None: scenes = [] 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, ) # Step 3: Correlate each scene independently. incident_ids, all_tags = await _correlate_scenes( call_id, node_id, system_id, talkgroup_id, talkgroup_name, scenes, ) # 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") # non_event_chatter (server-26#298) is the same population as # transcript_too_short, just longer: housekeeping that skipped the # paid extraction but may still be a same-talkgroup follow-up. if not skip_reason or skip_reason in ("transcript_too_short", "non_event_chatter"): incident_id, _ = await _correlate( 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, # server-26#104: the call doc's own severity, as # recorrelation_sweep passes it. None resolved to "routine", # which turned a reprocessed not-thin call into a thin one # that attaches with no fit check. severity=(_call_doc or {}).get("severity"), ) if incident_id: incident_ids.append(incident_id) if incident_ids: await fstore.doc_set("calls", call_id, {"incident_ids": incident_ids}) return incident_ids, all_tags, scenes def _late_upload_anchor(call_doc: Optional[dict]) -> Optional[datetime]: """ The call's own started_at when its upload is late (server-26#301), else None. "Late" is `late_upload_anchor_minutes` behind the real clock: an on-time call keeps being judged against now exactly as before, so the live path's recency windows are unchanged; only a spooled/retried upload is back-dated. A started_at in the future or unparseable is ignored. """ raw = (call_doc or {}).get("started_at") if not raw: return None if isinstance(raw, datetime): # mqtt_handler stores a native timestamp started = raw else: try: started = datetime.fromisoformat(str(raw).replace("Z", "+00:00")) except ValueError: return None if started.tzinfo is None: started = started.replace(tzinfo=timezone.utc) lag = clock.now() - started if lag > timedelta(minutes=settings.late_upload_anchor_minutes): logger.info(f"Late upload: call started {lag} ago — correlating as of {started.isoformat()}") return started return None 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, 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": clock.now().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, node_id=node_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}") # Steps 2-3: scene extraction + correlation. incident_ids: list[str] = [] all_tags: list[str] = [] if _flag("correlation_enabled"): # server-26#301: a late upload (the node spools and retries for up to # 72h) is correlated as of when the call happened, not as of now — the # same clock pin replay uses, so every recency gate and every # started_at/updated_at/resolved_at it writes is on the call's own # timeline. Without it a call hours late linked into whatever incident # was live on its talkgroup now, or opened one at the wrong time. anchor = _late_upload_anchor(_call_doc) tok = clock.pin(anchor) if anchor else None try: incident_ids, all_tags, _ = await _extract_and_correlate( call_id=call_id, node_id=node_id, system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name, transcript=transcript, segments=segments, ) finally: if tok is not None: clock.unpin(tok) 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 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, )