From aff3f16d32336d318ecf497c0d0f945546c02ceb Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sat, 26 Sep 2026 15:23:57 -0400 Subject: [PATCH] Admin Replay: re-run the pipeline over past calls in a sandbox (#170) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Correlation has only ever been measured through live AI windows: days of wall time per change, and the 09-20→22 window was invalidated outright by unfunded AI accounts (#169). Recordings are kept regardless of AI, so the traffic to measure against already exists. - internal/replay.py: runs a time range of real calls through the live pipeline code in original order, clock pinned per call, into replay_runs/{run_id}/calls|incidents. Modes: audio (re-transcribe), transcripts (re-extract), reuse (correlation only from a prior run's scenes). Simulates the idle-resolve and orphan-recorrelation sweeps on virtual time. No alerts, summaries, vocab, AI-health alerts or pending terms. One run at a time, <=5000 calls, <=7 days. - firestore.py: ContextVar sandbox redirect for calls/incidents. - clock.py: ContextVar-pinnable now(), used on the correlation path. - feature_flags.py: ContextVar flag override so replay runs with live AI off. - upload.py: scene loop extracted to _extract_and_correlate, shared by the live pipeline and replay so replay measures the code that runs live. - resolved_via on every incident resolve, so a real clear can be told from the idle timeout — live and in replay. - routers/replay.py + /admin Replay tab: estimate, start, compare runs, drill into incidents with audio. Reviewed by drb-correlation-review; its leak and fidelity findings are fixed and covered by tests. c2-core: 456 pass. Frontend typecheck not run (no Node on the authoring box). Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/app/internal/ai_health.py | 8 + drb-c2-core/app/internal/area_context.py | 2 + drb-c2-core/app/internal/clock.py | 32 + drb-c2-core/app/internal/feature_flags.py | 23 +- drb-c2-core/app/internal/firestore.py | 53 +- .../app/internal/incident_correlator.py | 8 +- .../app/internal/recorrelation_sweep.py | 3 +- drb-c2-core/app/internal/replay.py | 617 ++++++++++++++++++ drb-c2-core/app/internal/summarizer.py | 4 +- drb-c2-core/app/main.py | 3 +- drb-c2-core/app/routers/calls.py | 1 + drb-c2-core/app/routers/incidents.py | 1 + drb-c2-core/app/routers/replay.py | 181 +++++ drb-c2-core/app/routers/upload.py | 212 +++--- drb-c2-core/tests/test_replay.py | 371 +++++++++++ drb-frontend/app/admin/page.tsx | 7 +- drb-frontend/components/admin/ReplayTab.tsx | 439 +++++++++++++ drb-frontend/lib/c2api.ts | 22 + drb-frontend/lib/types.ts | 90 +++ 19 files changed, 1976 insertions(+), 101 deletions(-) create mode 100644 drb-c2-core/app/internal/clock.py create mode 100644 drb-c2-core/app/internal/replay.py create mode 100644 drb-c2-core/app/routers/replay.py create mode 100644 drb-c2-core/tests/test_replay.py create mode 100644 drb-frontend/components/admin/ReplayTab.tsx diff --git a/drb-c2-core/app/internal/ai_health.py b/drb-c2-core/app/internal/ai_health.py index f14af5a..22c07c2 100644 --- a/drb-c2-core/app/internal/ai_health.py +++ b/drb-c2-core/app/internal/ai_health.py @@ -108,6 +108,11 @@ async def report_degraded( TRANSIENT_ALERT_THRESHOLD consecutive failures have been reported for this tier, so an ordinary blip never pages anyone. """ + from app.internal import firestore as fstore + if fstore.in_sandbox(): + # A replay's rate limits are not a live outage, and must never page + # the AI-alert webhook or flip /health/ai (app/internal/replay.py). + return if tier not in _state: _state[tier] = _default_state() entry = _state[tier] @@ -145,6 +150,9 @@ async def report_healthy(tier: str) -> None: just after a failure -- it is what lets a degraded tier recover on its own instead of staying red forever after one transient blip. """ + from app.internal import firestore as fstore + if fstore.in_sandbox(): + return # nor may a replay's success "recover" a real live outage if tier not in _state: _state[tier] = _default_state() entry = _state[tier] diff --git a/drb-c2-core/app/internal/area_context.py b/drb-c2-core/app/internal/area_context.py index 67cc4bc..cc80f88 100644 --- a/drb-c2-core/app/internal/area_context.py +++ b/drb-c2-core/app/internal/area_context.py @@ -448,6 +448,8 @@ async def add_pending(system_id: str, talkgroup_id: Any, entries: list[dict]) -> """ from app.internal import firestore as fstore + if fstore.in_sandbox(): + return 0 # a replay proposes nothing to the live review queue if not system_id or talkgroup_id is None or not entries: return 0 system_doc = await fstore.doc_get("systems", system_id) diff --git a/drb-c2-core/app/internal/clock.py b/drb-c2-core/app/internal/clock.py new file mode 100644 index 0000000..62308bf --- /dev/null +++ b/drb-c2-core/app/internal/clock.py @@ -0,0 +1,32 @@ +""" +Pipeline clock. + +`now()` is `datetime.now(timezone.utc)` everywhere except inside a replay run +(app/internal/replay.py), which pins it to the replayed call's own time so the +correlator's recency windows, the idle-resolve sweep and every started_at / +updated_at / resolved_at it writes behave the way they did live. + +A ContextVar rather than a module global: a replay runs as a background task +alongside real uploads, and each asyncio task (and every asyncio.to_thread it +spawns) carries its own copy of the context, so a pinned clock can never leak +into a live call's pipeline. +""" +from contextvars import ContextVar +from datetime import datetime, timezone +from typing import Optional + +_pinned: ContextVar[Optional[datetime]] = ContextVar("drb_clock_pinned", default=None) + + +def now() -> datetime: + pinned = _pinned.get() + return pinned if pinned is not None else datetime.now(timezone.utc) + + +def pin(when: Optional[datetime]): + """Pin the clock for the current context. Returns a token for `unpin`.""" + return _pinned.set(when) + + +def unpin(token) -> None: + _pinned.reset(token) diff --git a/drb-c2-core/app/internal/feature_flags.py b/drb-c2-core/app/internal/feature_flags.py index c6ff9fa..7cfe8a5 100644 --- a/drb-c2-core/app/internal/feature_flags.py +++ b/drb-c2-core/app/internal/feature_flags.py @@ -6,7 +6,8 @@ in-memory TTL cache so flag reads don't add a Firestore round-trip to every call upload. """ import time -from typing import Any +from contextvars import ContextVar +from typing import Any, Optional from app.internal.logger import logger from app.internal import firestore as fstore @@ -36,6 +37,21 @@ _DEFAULTS: dict[str, bool] = { "transcript_correction_enabled": True, } +# A replay run (app/internal/replay.py) states exactly which AI steps it runs, +# independent of the live switches — the whole point is re-running the pipeline +# while live AI is OFF. ContextVar so the override never reaches a live upload. +_forced: ContextVar[Optional[dict[str, bool]]] = ContextVar("drb_forced_flags", default=None) + + +def force_flags(flags: Optional[dict[str, bool]]): + """Override resolve_flags() for the current context. Returns a reset token.""" + return _forced.set(flags) + + +def unforce_flags(token) -> None: + _forced.reset(token) + + _cache: dict[str, Any] = {} _cache_ts: float = 0.0 @@ -211,6 +227,11 @@ async def resolve_flags(system_id: str | None): """ from app.internal import firestore as _fstore + forced = _forced.get() + if forced is not None: + full = {k: bool(forced.get(k, False)) for k in _DEFAULTS} + return full, lambda name: full.get(name, False) + flags = await get_flags() system_ai_flags: dict = {} diff --git a/drb-c2-core/app/internal/firestore.py b/drb-c2-core/app/internal/firestore.py index c9415c2..f7843a7 100644 --- a/drb-c2-core/app/internal/firestore.py +++ b/drb-c2-core/app/internal/firestore.py @@ -1,5 +1,6 @@ import asyncio import time as _time +from contextvars import ContextVar from typing import Optional, Any import firebase_admin from firebase_admin import credentials, firestore as fs @@ -40,23 +41,61 @@ _init_firebase() db = fs.client(database_id=settings.firestore_database) +# --------------------------------------------------------------------------- +# Replay sandbox (app/internal/replay.py) +# --------------------------------------------------------------------------- +# While a replay run is executing, every read and write the pipeline makes to +# `calls` or `incidents` is redirected to that run's own subcollections under +# replay_runs/{run_id}/, so re-running the pipeline over past traffic can never +# touch a live call or incident. A subcollection keeps the same collection ID +# ("calls"/"incidents"), so the composite indexes prod queries depend on apply +# to it unchanged. Everything else (systems, nodes, config) is read from prod +# as-is. ContextVar for the same reason as app/internal/clock.py: the redirect +# follows the replay task and never a concurrent live upload. +SANDBOXED_COLLECTIONS = frozenset({"calls", "incidents"}) +_sandbox_root: ContextVar[Optional[str]] = ContextVar("drb_fstore_sandbox", default=None) + + +def enter_sandbox(root: Optional[str]): + """Redirect calls/incidents under `root` (e.g. "replay_runs/") for this context.""" + return _sandbox_root.set(root) + + +def exit_sandbox(token) -> None: + _sandbox_root.reset(token) + + +def in_sandbox() -> bool: + """True inside a replay run. Anything that writes live state OTHER than + calls/incidents (AI health alerts, pending-term queues) checks this and + stands down — the redirect below only covers the two sandboxed collections.""" + return _sandbox_root.get() is not None + + +def _path(collection: str) -> str: + root = _sandbox_root.get() + if root and collection in SANDBOXED_COLLECTIONS: + return f"{root}/{collection}" + return collection + + # --------------------------------------------------------------------------- # Thin async wrappers — firebase-admin is synchronous, run in thread executor # --------------------------------------------------------------------------- async def doc_set(collection: str, doc_id: str, data: dict, merge: bool = True) -> None: - ref = db.collection(collection).document(doc_id) + ref = db.collection(_path(collection)).document(doc_id) await asyncio.to_thread(ref.set, data, merge=merge) async def doc_get(collection: str, doc_id: str) -> Optional[dict]: - ref = db.collection(collection).document(doc_id) + ref = db.collection(_path(collection)).document(doc_id) snap = await asyncio.to_thread(ref.get) return snap.to_dict() if snap.exists else None async def doc_update(collection: str, doc_id: str, data: dict) -> None: - ref = db.collection(collection).document(doc_id) + ref = db.collection(_path(collection)).document(doc_id) await asyncio.to_thread(ref.update, data) @@ -66,7 +105,7 @@ async def collection_list(collection: str, **filters) -> list[dict]: Optional keyword filters: field=value pairs passed as equality where-clauses. """ def _query(): - ref = db.collection(collection) + ref = db.collection(_path(collection)) for field, value in filters.items(): ref = ref.where(filter=FieldFilter(field, "==", value)) return [doc.to_dict() for doc in ref.stream()] @@ -103,7 +142,7 @@ async def collection_where( unscoped equality-only lookups can keep using collection_list(). """ def _query(): - ref = db.collection(collection) + ref = db.collection(_path(collection)) for field, op, value in conditions: ref = ref.where(filter=FieldFilter(field, op, value)) for field, direction in (order_by or []): @@ -118,7 +157,7 @@ async def collection_where( async def doc_delete(collection: str, doc_id: str) -> None: - ref = db.collection(collection).document(doc_id) + ref = db.collection(_path(collection)).document(doc_id) await asyncio.to_thread(ref.delete) @@ -128,7 +167,7 @@ async def doc_get_cached(collection: str, doc_id: str, ttl: float = 300.0) -> Op Use for documents that change rarely (systems config, node assignments). Default TTL is 5 minutes — a write will be visible within that window. """ - key = f"{collection}/{doc_id}" + key = f"{_path(collection)}/{doc_id}" now = _time.monotonic() entry = _doc_cache.get(key) if entry and now < entry[0]: diff --git a/drb-c2-core/app/internal/incident_correlator.py b/drb-c2-core/app/internal/incident_correlator.py index fbeb4c4..8ed9b44 100644 --- a/drb-c2-core/app/internal/incident_correlator.py +++ b/drb-c2-core/app/internal/incident_correlator.py @@ -51,6 +51,7 @@ from datetime import datetime, timezone, timedelta from typing import Optional from app.internal.logger import logger from app.internal import firestore as fstore +from app.internal import clock from app.config import settings _PURSUIT_TAGS = frozenset({ @@ -812,7 +813,7 @@ async def _build_context( transcript: Optional[str] = None, scene_index: int = 0, ) -> dict: - now = reference_time or datetime.now(timezone.utc) + now = reference_time or clock.now() window = timedelta(hours=settings.correlation_window_hours) call_doc = await fstore.doc_get("calls", call_id) or {} @@ -1951,6 +1952,7 @@ async def _release_reassigned_units(ctx: dict, exclude_incident_id: Optional[str if auto_resolved: updates["status"] = "resolved" updates["resolved_at"] = now.isoformat() + updates["resolved_via"] = "reassignment" await fstore.doc_set("incidents", inc["incident_id"], updates) logger.info( f"Correlator: reassignment released unit(s) {matched} from incident " @@ -2072,6 +2074,7 @@ async def _update_incident( if units_cleared and not units_active: updates["status"] = "resolved" updates["resolved_at"] = now.isoformat() + updates["resolved_via"] = "units_cleared" await fstore.doc_set("incidents", incident_id, updates) logger.info( f"Correlator: signal-resolved incident {incident_id} " @@ -2274,7 +2277,8 @@ async def maybe_resolve_parent(incident_id: str) -> None: # All children resolved — close the master await fstore.doc_set("incidents", parent_id, { "status": "resolved", - "resolved_at": datetime.now(timezone.utc).isoformat(), + "resolved_at": clock.now().isoformat(), + "resolved_via": "children_resolved", }) logger.info( f"Auto-resolved master incident {parent_id} " diff --git a/drb-c2-core/app/internal/recorrelation_sweep.py b/drb-c2-core/app/internal/recorrelation_sweep.py index c23a590..cdbc5a1 100644 --- a/drb-c2-core/app/internal/recorrelation_sweep.py +++ b/drb-c2-core/app/internal/recorrelation_sweep.py @@ -90,7 +90,8 @@ def _pipeline_likely_still_running(call: dict, now: datetime) -> bool: async def _run_sweep_pass() -> None: - now = datetime.now(timezone.utc) + from app.internal import clock + now = clock.now() cutoff = now - timedelta(minutes=settings.recorrelation_scan_minutes) # Server-side range query: only calls that ended within the scan window. diff --git a/drb-c2-core/app/internal/replay.py b/drb-c2-core/app/internal/replay.py new file mode 100644 index 0000000..1deea19 --- /dev/null +++ b/drb-c2-core/app/internal/replay.py @@ -0,0 +1,617 @@ +""" +Replay — re-run the intelligence pipeline over past calls, in a sandbox. + +Live AI windows are the only way the correlator has ever been measured, and +each one costs days of real time and whatever the credits allow: a change +ships, AI goes on, traffic trickles in, someone pulls a dump. Recordings are +kept whether AI is on or not, so the traffic to measure against already +exists. A replay run takes a time range of real calls, feeds them through +the SAME pipeline code the live upload path runs (routers/upload.py +`_extract_and_correlate`) in their original order with the clock pinned to +each call's own end time, and writes everything to +replay_runs/{run_id}/calls|incidents instead of the live collections. The +same range can then be replayed after every change and the runs compared. + +Three modes, cheapest last: + audio re-transcribe the saved audio (Whisper + correction), then + extract and correlate. For ranges where AI was off. + transcripts reuse the transcript already on each call, re-run extraction + and correlation. + reuse reuse the scenes an earlier run extracted, re-run correlation + only. Extraction is an LLM call and never returns quite the + same thing twice, so this is the mode that isolates a + correlator change from extraction noise. + +What never happens in a replay: alerts, summaries, vocabulary learning, and +any write to a live call or incident. The sandbox is enforced by the +ContextVar redirect in app/internal/firestore.py, not by this module +remembering to use different collection names. + +One run at a time per process — a run spends real AI credits and its cost is +only estimated, so two concurrent runs would be two unbounded bills. +""" +import asyncio +import os +import statistics +import uuid +from collections import Counter +from datetime import datetime, timedelta, timezone +from typing import Optional + +from app.config import settings +from app.internal import clock +from app.internal import firestore as fstore +from app.internal.feature_flags import force_flags, unforce_flags +from app.internal.logger import logger + +RUNS = "replay_runs" +MODES = ("audio", "transcripts", "reuse") +MAX_CALLS = 5000 +MAX_RANGE_DAYS = 7 + +# Extraction/transcription run ahead of correlation with this much +# concurrency. They depend only on the call itself; correlation depends on +# every call before it and is kept strictly in order. +PREFETCH = 6 + +# Rough per-unit AI prices for the pre-run estimate and the running tally. +# Estimates, not a bill — nothing in DRB reads a real invoice (server-26#45). +USD_WHISPER_PER_MIN = 0.006 +USD_PER_EXTRACTION = 0.0005 # gpt-4o-mini scene extraction + embedding +USD_PER_CORRECTION = 0.0003 # Gemini flash transcript correction +USD_PER_GEOCODE = 0.005 # Google geocode, roughly one per located scene +USD_PER_LLM_CORRELATE = 0.0005 # Gemini flash consensus decision + +# Fields the pipeline writes onto a call doc. Stripped when a call is copied +# into the sandbox so the replay recomputes them instead of inheriting the +# live answer. Anything else on the doc (ids, times, talkgroup, srcaddr, audio +# location) is an input and is kept. +_DERIVED = { + "transcript", "transcript_corrected", "transcript_not_speech", + "segments", "segments_corrected", "scenes", "incident_id", "incident_ids", + "tags", "location", "location_coords", "location_mentions", "units", + "vehicles", "cleared_units", "severity", "incident_type", "type", + "embedding", "skip_reason", "intelligence_started_at", "reassignment", + "resolved", "has_updates", "audio_url", +} +_DERIVED_PREFIXES = ("corr_", "chatter_classifier_", "eval_") +_TRANSCRIPT_FIELDS = ( + "transcript", "transcript_corrected", "transcript_not_speech", + "segments", "segments_corrected", +) + +_active_run_id: Optional[str] = None +_active_task: Optional[asyncio.Task] = None +_cancel: set[str] = set() + + +class ReplayBusy(RuntimeError): + pass + + +def sandbox_root(run_id: str) -> str: + return f"{RUNS}/{run_id}" + + +def _scenes_coll(run_id: str) -> str: + # Extracted scenes (embeddings included) live beside the sandbox, not on + # its call docs, so reading a run's calls for metrics or the incident view + # doesn't haul every scene's embedding along a second time. + return f"{sandbox_root(run_id)}/scenes" + + +def active_run_id() -> Optional[str]: + if _active_task is not None and not _active_task.done(): + return _active_run_id + return None + + +# --------------------------------------------------------------------------- +# Call selection +# --------------------------------------------------------------------------- + +def _as_dt(value) -> Optional[datetime]: + if value is None: + return None + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + try: + dt = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + return None + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + + +async def select_calls( + org_id: str, + date_from: datetime, + date_to: datetime, + system_ids: Optional[list[str]] = None, + cap: int = MAX_CALLS, +) -> tuple[list[dict], bool]: + """ + Live calls in [date_from, date_to] for this org, oldest first. + + Pages newest-first because the one composite index on calls that carries + org_id is (org_id ASC, started_at DESC); ordering the other way would need + a new index for no gain. Returns (calls, truncated) — truncated means the + range holds more than `cap` calls and the caller must narrow it rather + than silently replaying only part of it. + """ + out: list[dict] = [] + cursor = None + page = 1000 + while True: + rows = await fstore.collection_where( + "calls", + [("org_id", "==", org_id), + ("started_at", ">=", date_from), + ("started_at", "<=", date_to)], + order_by=[("started_at", "DESCENDING")], + limit_to=page, + start_after={"started_at": cursor} if cursor is not None else None, + ) + for c in rows: + if c.get("duplicate_of"): + continue # another node's copy — live never processes these either + if system_ids and c.get("system_id") not in system_ids: + continue + out.append(c) + if len(out) > cap: + return sorted(out[:cap], key=_call_time), True + if len(rows) < page: + break + cursor = rows[-1].get("started_at") + return sorted(out, key=_call_time), False + + +def _call_time(call: dict) -> datetime: + return _as_dt(call.get("started_at")) or datetime.min.replace(tzinfo=timezone.utc) + + +def _pipeline_time(call: dict) -> datetime: + """When the live pipeline would have run for this call: at upload, i.e. call end.""" + return _as_dt(call.get("ended_at")) or _call_time(call) + + +def estimate(calls: list[dict], mode: str) -> dict: + n = len(calls) + with_transcript = sum(1 for c in calls if c.get("transcript_corrected") or c.get("transcript")) + audio_min = sum(float(c.get("duration_s") or 0) for c in calls) / 60 + with_audio = sum(1 for c in calls if c.get("audio_gcs_uri")) + # Roughly a third of calls carry a geocodable location (09-22 dump: 92/373). + per_call = USD_PER_EXTRACTION + USD_PER_LLM_CORRELATE + USD_PER_GEOCODE / 3 + if mode == "audio": + usd = audio_min * USD_WHISPER_PER_MIN + with_audio * (USD_PER_CORRECTION + per_call) + elif mode == "transcripts": + usd = with_transcript * per_call + else: + usd = n * USD_PER_LLM_CORRELATE + return { + "calls": n, + "calls_with_transcript": with_transcript, + "calls_with_audio": with_audio, + "audio_minutes": round(audio_min, 1), + "est_cost_usd": round(usd, 2), + } + + +# --------------------------------------------------------------------------- +# Run lifecycle +# --------------------------------------------------------------------------- + +async def start_run( + *, + org_id: str, + date_from: datetime, + date_to: datetime, + mode: str, + system_ids: Optional[list[str]], + source_run_id: Optional[str], + label: str, + actor: str, +) -> dict: + global _active_run_id, _active_task + if active_run_id(): + raise ReplayBusy(f"Replay {active_run_id()} is still running.") + if mode not in MODES: + raise ValueError(f"mode must be one of {MODES}") + if date_to <= date_from: + raise ValueError("date_to must be after date_from") + if date_to - date_from > timedelta(days=MAX_RANGE_DAYS): + raise ValueError(f"Range is capped at {MAX_RANGE_DAYS} days.") + if mode == "reuse": + src = await fstore.doc_get(RUNS, source_run_id or "") + if not src or src.get("org_id") != org_id: + raise ValueError("reuse mode needs a source_run_id from an earlier run in this org") + if src.get("status") != "done": + raise ValueError("The source run did not finish; its scenes are incomplete.") + + calls, truncated = await select_calls(org_id, date_from, date_to, system_ids) + if truncated: + raise ValueError(f"Range holds more than {MAX_CALLS} calls — narrow it.") + if not calls: + raise ValueError("No calls in that range.") + + run_id = uuid.uuid4().hex[:12] + now = datetime.now(timezone.utc).isoformat() + doc = { + "run_id": run_id, + "org_id": org_id, + "label": label or "", + "mode": mode, + "source_run_id": source_run_id if mode == "reuse" else None, + "date_from": date_from.isoformat(), + "date_to": date_to.isoformat(), + "system_ids": system_ids or [], + "git_sha": os.getenv("GIT_SHA", "unknown"), + "created_by": actor, + "created_at": now, + "status": "running", + "estimate": estimate(calls, mode), + "progress": {"total": len(calls), "done": 0, "errors": 0}, + "metrics": None, + "errors": [], + } + await fstore.doc_set(RUNS, run_id, doc, merge=False) + + _active_run_id = run_id + _active_task = asyncio.create_task(_run(run_id, org_id, calls, mode, source_run_id)) + return doc + + +def request_cancel(run_id: str) -> bool: + if active_run_id() != run_id: + return False + _cancel.add(run_id) + return True + + +async def get_run(run_id: str) -> Optional[dict]: + doc = await fstore.doc_get(RUNS, run_id) + return await _reconcile(doc) if doc else None + + +async def list_runs(org_id: str) -> list[dict]: + docs = await fstore.collection_list(RUNS, org_id=org_id) + docs = [await _reconcile(d) for d in docs] + return sorted(docs, key=lambda d: d.get("created_at") or "", reverse=True) + + +async def _reconcile(doc: dict) -> dict: + """A run left "running" by a process that restarted (a deploy) never finishes.""" + if doc.get("status") == "running" and doc.get("run_id") != active_run_id(): + doc["status"] = "interrupted" + await fstore.doc_set(RUNS, doc["run_id"], {"status": "interrupted"}) + return doc + + +async def delete_run(run_id: str) -> None: + if active_run_id() == run_id: + raise ReplayBusy("Cancel the run before deleting it.") + token = fstore.enter_sandbox(sandbox_root(run_id)) + try: + for coll, key in (("calls", "call_id"), ("incidents", "incident_id"), + (_scenes_coll(run_id), "call_id")): + for d in await fstore.collection_list(coll): + if d.get(key): + await fstore.doc_delete(coll, d[key]) + finally: + fstore.exit_sandbox(token) + await fstore.doc_delete(RUNS, run_id) + + +async def sandbox_contents(run_id: str) -> tuple[list[dict], list[dict]]: + token = fstore.enter_sandbox(sandbox_root(run_id)) + try: + incidents = await fstore.collection_list("incidents") + calls = await fstore.collection_list("calls") + finally: + fstore.exit_sandbox(token) + return incidents, calls + + +# --------------------------------------------------------------------------- +# The run itself +# --------------------------------------------------------------------------- + +def _flags_for(mode: str) -> dict[str, bool]: + return { + "stt_enabled": mode == "audio", + "transcript_correction_enabled": mode == "audio", + "correlation_enabled": True, + "summaries_enabled": False, + "vocabulary_learning_enabled": False, + } + + +def _stored_input(call: dict) -> tuple[Optional[str], list]: + """ + The transcript + segments live extraction was handed for this call. + + Not simply `transcript_corrected`: live extraction overwrites that field + with its primary scene's rewrite (intelligence.py), so on a call with + several scenes it now holds only scene 0's text. The corrector's own + output survives intact in `segments_corrected`, so rebuild from those + when they exist; otherwise correction never produced anything and live + extraction read the raw Whisper transcript. + """ + if call.get("transcript_not_speech"): + return None, [] # transcribe_call hands nothing downstream for noise + corrected = call.get("segments_corrected") or [] + if corrected: + text = " ".join(str(seg.get("text") or "").strip() for seg in corrected).strip() + return (text or call.get("transcript")), corrected + return call.get("transcript"), call.get("segments") or [] + + +def _extraction_fields(sb_call: dict) -> dict: + """What extraction wrote onto the call doc (tags, units, location, + embedding, skip_reason, ...), minus everything correlation wrote. A reuse + run restores these so the orphan sweep, which reads them straight off the + call doc, sees what it saw in the source run.""" + return { + k: v for k, v in sb_call.items() + if (k in _DERIVED or k.startswith("chatter_classifier_")) + and k not in _TRANSCRIPT_FIELDS + and k not in ("scenes", "incident_id", "incident_ids", "intelligence_started_at") + } + + +def _sandbox_seed(call: dict, mode: str) -> dict: + keep_transcript = mode in ("transcripts", "reuse") + seed = {} + for k, v in call.items(): + if k in _TRANSCRIPT_FIELDS: + if keep_transcript: + seed[k] = v + continue + if k in _DERIVED or k.startswith(_DERIVED_PREFIXES): + continue + seed[k] = v + # Calls are seeded ahead of the replay clock (see PREFETCH). The orphan + # re-correlation sweep selects status=="ended" calls by ended_at, so a + # seeded call keeping its real status would be swept up as an "orphan" + # before its own turn. Its real status is restored when it is processed. + seed["status"] = "replay_pending" + return seed + + +async def _prepare(call: dict, mode: str, source_scenes: dict[str, dict]) -> dict: + """ + Everything per call that doesn't depend on other calls: seed the sandbox + doc, then transcribe and/or extract. Runs ahead of correlation. + Returns {"transcript", "scenes", "skip"} for the in-order stage. + """ + from app.internal import intelligence, talkgroups, transcription + + call_id = call["call_id"] + await fstore.doc_set("calls", call_id, _sandbox_seed(call, mode), merge=False) + + talkgroup_name = await talkgroups.resolve( + call.get("system_id"), call.get("talkgroup_id"), + hint=call.get("talkgroup_name"), call_doc=call, + ) + + transcript: Optional[str] = None + segments: list = [] + if mode == "audio": + if call.get("audio_gcs_uri"): + transcript, segments = await transcription.transcribe_call( + call_id, call["audio_gcs_uri"], talkgroup_name, + system_id=call.get("system_id"), talkgroup_id=call.get("talkgroup_id"), + ) + else: + transcript, segments = _stored_input(call) + + if mode == "reuse": + src = source_scenes.get(call_id) + if src is None: + return {"skip": "not_in_source_run", "talkgroup_name": talkgroup_name} + if src.get("call_fields"): + # skip_reason gates upload.py's no-scene fallback and the orphan + # sweep correlates from tags/units/location on the call doc, so + # extraction's call-level output comes across with its scenes. + await fstore.doc_set("calls", call_id, src["call_fields"]) + return {"transcript": transcript, "scenes": src.get("scenes") or [], + "talkgroup_name": talkgroup_name} + + scenes: list = [] + if transcript: + scenes = await intelligence.extract_scenes( + call_id, transcript, talkgroup_name, + talkgroup_id=call.get("talkgroup_id"), system_id=call.get("system_id"), + segments=segments, node_id=call.get("node_id"), + ) + return {"transcript": transcript, "scenes": scenes, "talkgroup_name": talkgroup_name} + + +async def _sweeps_until(t: datetime, state: dict) -> None: + """Run the live periodic sweeps (idle auto-resolve, orphan re-correlation) at every tick up to t.""" + from app.internal import recorrelation_sweep, summarizer + + interval = timedelta(minutes=settings.summary_interval_minutes) + if state["last_sweep"] is None: + state["last_sweep"] = t + return + while state["last_sweep"] + interval <= t: + state["last_sweep"] += interval + tok = clock.pin(state["last_sweep"]) + try: + await summarizer._resolve_stale_incidents() + await recorrelation_sweep._run_sweep_pass() + finally: + clock.unpin(tok) + + +async def _run(run_id: str, org_id: str, calls: list[dict], mode: str, + source_run_id: Optional[str]) -> None: + from app.routers.upload import _extract_and_correlate + + global _active_run_id + progress = {"total": len(calls), "done": 0, "errors": 0, "skipped": 0, + "extractions": 0, "audio_minutes": 0.0} + errors: list[str] = [] + status = "done" + + source_scenes: dict[str, dict] = {} + if mode == "reuse" and source_run_id: + rows = await fstore.collection_list(_scenes_coll(source_run_id)) + source_scenes = {r["call_id"]: r for r in rows if r.get("call_id")} + + sb_token = fstore.enter_sandbox(sandbox_root(run_id)) + fl_token = force_flags(_flags_for(mode)) + try: + sem = asyncio.Semaphore(PREFETCH) + + async def prep(call: dict): + async with sem: + tok = clock.pin(_pipeline_time(call)) + try: + return await _prepare(call, mode, source_scenes) + finally: + clock.unpin(tok) + + pending: dict[int, asyncio.Task] = {} + sweep_state = {"last_sweep": None} + last_t = None + for i, call in enumerate(calls): + for j in range(i, min(i + PREFETCH * 2, len(calls))): + if j not in pending: + pending[j] = asyncio.create_task(prep(calls[j])) + if run_id in _cancel: + status = "cancelled" + break + + t = _pipeline_time(call) + last_t = t + try: + prepared = await pending.pop(i) + await _sweeps_until(t, sweep_state) + if prepared.get("skip"): + progress["skipped"] += 1 + else: + tok = clock.pin(t) + try: + await fstore.doc_set("calls", call["call_id"], { + "status": call.get("status") or "ended", + "intelligence_started_at": t.isoformat(), + }) + _, _, scenes = await _extract_and_correlate( + call_id=call["call_id"], + node_id=call.get("node_id"), + system_id=call.get("system_id"), + talkgroup_id=call.get("talkgroup_id"), + talkgroup_name=prepared["talkgroup_name"], + transcript=prepared["transcript"], + scenes=prepared["scenes"], + ) + finally: + clock.unpin(tok) + # Kept whole (embedding included) so a later "reuse" run + # can correlate from exactly these scenes. + sb_call = await fstore.doc_get("calls", call["call_id"]) or {} + await fstore.doc_set(_scenes_coll(run_id), call["call_id"], { + "call_id": call["call_id"], + "scenes": scenes, + "call_fields": _extraction_fields(sb_call), + }, merge=False) + if prepared["transcript"] and mode != "reuse": + progress["extractions"] += 1 + if mode == "audio": + progress["audio_minutes"] += float(call.get("duration_s") or 0) / 60 + except Exception as e: + progress["errors"] += 1 + if len(errors) < 20: + errors.append(f"{call.get('call_id')}: {type(e).__name__}: {e}"[:300]) + logger.warning(f"Replay {run_id}: call {call.get('call_id')} failed: {e}") + progress["done"] = i + 1 + if (i + 1) % 25 == 0: + await fstore.doc_set(RUNS, run_id, {"progress": dict(progress), "errors": errors}) + + for task in pending.values(): + task.cancel() + + if status == "done" and last_t is not None: + # Let every incident age out exactly as it would have live. + await _sweeps_until( + last_t + timedelta(minutes=settings.incident_auto_resolve_minutes + + 2 * settings.summary_interval_minutes), + sweep_state, + ) + + incidents = await fstore.collection_list("incidents") + sb_calls = await fstore.collection_list("calls") + metrics = compute_metrics(incidents, sb_calls) + metrics["est_cost_usd"] = _running_cost(progress, metrics, mode) + except Exception as e: + status = "failed" + errors.append(f"run: {type(e).__name__}: {e}"[:300]) + metrics = None + logger.error(f"Replay {run_id} failed: {e}") + finally: + unforce_flags(fl_token) + fstore.exit_sandbox(sb_token) + _cancel.discard(run_id) + _active_run_id = None + + await fstore.doc_set(RUNS, run_id, { + "status": status, + "progress": progress, + "errors": errors, + "metrics": metrics, + "finished_at": datetime.now(timezone.utc).isoformat(), + }) + logger.info(f"Replay {run_id} {status}: {progress}") + + +def _running_cost(progress: dict, metrics: dict, mode: str) -> float: + usd = progress["audio_minutes"] * USD_WHISPER_PER_MIN + if mode == "audio": + usd += progress["extractions"] * USD_PER_CORRECTION + usd += progress["extractions"] * (USD_PER_EXTRACTION + USD_PER_GEOCODE / 3) + usd += metrics.get("llm_decisions", 0) * USD_PER_LLM_CORRELATE + return round(usd, 2) + + +# --------------------------------------------------------------------------- +# Scoring +# --------------------------------------------------------------------------- + +def compute_metrics(incidents: list[dict], calls: list[dict]) -> dict: + """ + The numbers that say whether incidents are being tracked, from one run's + sandbox. Same questions every correlation review has asked by hand, so two + runs over the same range compare directly. + """ + sizes = [len(i.get("call_ids") or []) for i in incidents] + resolved_via = Counter( + (i.get("resolved_via") or ("unknown" if i.get("status") == "resolved" else "still_active")) + for i in incidents + ) + corr_path: Counter = Counter() + consensus: Counter = Counter() + for c in calls: + scenes = c.get("scenes") or {} + records = [s.get("corr_debug") or {} for s in scenes.values()] if scenes else [c] + for r in records: + corr_path[r.get("corr_path") or "none"] += 1 + consensus[r.get("corr_consensus") or "none"] += 1 + linked = sum(1 for c in calls if c.get("incident_ids")) + llm = sum(n for k, n in consensus.items() if k not in ("none", "rules_only")) + return { + "calls": len(calls), + "calls_linked": linked, + "calls_orphaned": len(calls) - linked, + "incidents": len(incidents), + "single_call_incidents": sum(1 for s in sizes if s == 1), + "single_call_pct": round(100 * sum(1 for s in sizes if s == 1) / len(sizes), 1) if sizes else None, + "median_calls_per_incident": statistics.median(sizes) if sizes else None, + "max_calls_in_incident": max(sizes) if sizes else None, + "incidents_with_units_cleared": sum(1 for i in incidents if i.get("units_cleared")), + "incidents_with_coords": sum(1 for i in incidents if i.get("location_coords")), + "resolved_via": dict(resolved_via), + "corr_path": dict(corr_path), + "corr_consensus": dict(consensus), + "llm_decisions": llm, + } diff --git a/drb-c2-core/app/internal/summarizer.py b/drb-c2-core/app/internal/summarizer.py index 2d7f0cd..1211ea0 100644 --- a/drb-c2-core/app/internal/summarizer.py +++ b/drb-c2-core/app/internal/summarizer.py @@ -148,7 +148,8 @@ async def _resolve_stale_incidents() -> None: if not all_active: return - now = datetime.now(timezone.utc) + from app.internal import clock + now = clock.now() cutoff = timedelta(minutes=settings.incident_auto_resolve_minutes) count = 0 @@ -167,6 +168,7 @@ async def _resolve_stale_incidents() -> None: await fstore.doc_set("incidents", incident_id, { "status": "resolved", "resolved_at": now.isoformat(), + "resolved_via": "idle_timeout", }) from app.internal.incident_correlator import maybe_resolve_parent await maybe_resolve_parent(incident_id) diff --git a/drb-c2-core/app/main.py b/drb-c2-core/app/main.py index 0d76464..2923129 100644 --- a/drb-c2-core/app/main.py +++ b/drb-c2-core/app/main.py @@ -17,7 +17,7 @@ from app.internal.auth import ( require_node_service_or_firebase_token, ) from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users -from app.routers import enrollment, media, org, waitlist, telemetry +from app.routers import enrollment, media, org, waitlist, telemetry, replay from app.internal import dynsec from app.internal import firestore as fstore @@ -129,6 +129,7 @@ app.include_router(trips.router, dependencies=[Depends(require_service_or_fi app.include_router(places.router, dependencies=[Depends(require_service_or_firebase_token)]) app.include_router(upload.router) # auth is per-node, handled inline app.include_router(admin.router) # auth is per-endpoint (read: firebase, write: admin) +app.include_router(replay.router) # auth: admin only (every route spends or reads a replay run) app.include_router(users.router) # auth: admin only app.include_router(links.router) # auth is per-endpoint (generate: firebase, resolve: service key) app.include_router(enrollment.router) # public; auth is the enrollment/pickup-secret tokens, checked inline diff --git a/drb-c2-core/app/routers/calls.py b/drb-c2-core/app/routers/calls.py index 870400f..3a7ffcc 100644 --- a/drb-c2-core/app/routers/calls.py +++ b/drb-c2-core/app/routers/calls.py @@ -443,6 +443,7 @@ async def patch_transcript( "call_ids": [], "status": "resolved", "resolved_at": datetime.now(timezone.utc).isoformat(), + "resolved_via": "emptied_by_correction", "summary_stale": True, }) await fstore.doc_set("calls", call_id, {"incident_ids": [], "incident_id": None}) diff --git a/drb-c2-core/app/routers/incidents.py b/drb-c2-core/app/routers/incidents.py index 5ba509e..ab05aa4 100644 --- a/drb-c2-core/app/routers/incidents.py +++ b/drb-c2-core/app/routers/incidents.py @@ -170,6 +170,7 @@ async def unlink_call_from_incident(incident_id: str, call_id: str, _: dict = De if not remaining: updates["status"] = "resolved" updates["resolved_at"] = datetime.now(timezone.utc).isoformat() + updates["resolved_via"] = "emptied_by_admin" await fstore.doc_update("incidents", incident_id, updates) call = await fstore.doc_get("calls", call_id) diff --git a/drb-c2-core/app/routers/replay.py b/drb-c2-core/app/routers/replay.py new file mode 100644 index 0000000..30e5297 --- /dev/null +++ b/drb-c2-core/app/routers/replay.py @@ -0,0 +1,181 @@ +""" +Admin replay routes — the backend for the /admin Replay tab. + +See app/internal/replay.py for what a run is and why it exists. Every route is +admin-only: a run spends real AI credits. +""" +from datetime import datetime, timezone +from typing import Literal, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel + +from app.internal import replay +from app.internal.audit import write_audit +from app.internal.auth import describe_actor, require_admin_token, resolve_caller_org_id +from app.internal.logger import logger + +router = APIRouter(prefix="/admin/replay", tags=["admin"]) + + +def _parse_ts(value: Optional[str], field: str) -> datetime: + if not value: + raise HTTPException(400, f"{field} is required.") + try: + dt = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + raise HTTPException(400, f"{field} is not an ISO-8601 timestamp.") + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + + +async def _org(decoded: dict) -> str: + # Same fallback as /calls/search: a platform admin resolves to "every + # org", which is not a scope a replay can run in. + org_id = await resolve_caller_org_id(decoded) or decoded.get("org_id") + if not org_id: + raise HTTPException(403, "No organization scope for this caller.") + return org_id + + +async def _own_run(run_id: str, org_id: str) -> dict: + run = await replay.get_run(run_id) + if not run or run.get("org_id") != org_id: + raise HTTPException(404, f"Replay run '{run_id}' not found.") + return run + + +@router.get("/estimate") +async def estimate_run( + date_from: str = Query(...), + date_to: str = Query(...), + mode: Literal["audio", "transcripts", "reuse"] = Query("transcripts"), + system_ids: Optional[str] = Query(None, description="comma-separated"), + decoded: dict = Depends(require_admin_token), +): + """How many calls a run over this range would process, and a rough cost.""" + org_id = await _org(decoded) + sids = [s for s in (system_ids or "").split(",") if s] or None + calls, truncated = await replay.select_calls( + org_id, _parse_ts(date_from, "date_from"), _parse_ts(date_to, "date_to"), sids, + ) + return {**replay.estimate(calls, mode), "truncated": truncated, "max_calls": replay.MAX_CALLS} + + +class StartRun(BaseModel): + date_from: str + date_to: str + mode: Literal["audio", "transcripts", "reuse"] = "transcripts" + system_ids: Optional[list[str]] = None + source_run_id: Optional[str] = None + label: str = "" + + +@router.post("") +async def start_run(body: StartRun, decoded: dict = Depends(require_admin_token)): + org_id = await _org(decoded) + actor_uid, actor_email = describe_actor(decoded) + try: + run = await replay.start_run( + org_id=org_id, + date_from=_parse_ts(body.date_from, "date_from"), + date_to=_parse_ts(body.date_to, "date_to"), + mode=body.mode, + system_ids=body.system_ids or None, + source_run_id=body.source_run_id, + label=body.label[:120], + actor=actor_email or actor_uid, + ) + except replay.ReplayBusy as e: + raise HTTPException(409, str(e)) + except ValueError as e: + raise HTTPException(400, str(e)) + try: + await write_audit(actor_uid, actor_email, "replay.start", details={ + "run_id": run["run_id"], "mode": run["mode"], "calls": run["progress"]["total"], + "est_cost_usd": run["estimate"]["est_cost_usd"], + }) + except Exception as e: + logger.error(f"Replay: audit write failed ({e}) — run {run['run_id']} continues") + return run + + +@router.get("") +async def list_runs(decoded: dict = Depends(require_admin_token)): + org_id = await _org(decoded) + return {"runs": await replay.list_runs(org_id), "active_run_id": replay.active_run_id()} + + +@router.get("/{run_id}") +async def get_run(run_id: str, decoded: dict = Depends(require_admin_token)): + return await _own_run(run_id, await _org(decoded)) + + +@router.post("/{run_id}/cancel") +async def cancel_run(run_id: str, decoded: dict = Depends(require_admin_token)): + await _own_run(run_id, await _org(decoded)) + if not replay.request_cancel(run_id): + raise HTTPException(409, "That run is not running.") + return {"ok": True} + + +@router.delete("/{run_id}") +async def delete_run(run_id: str, decoded: dict = Depends(require_admin_token)): + await _own_run(run_id, await _org(decoded)) + try: + await replay.delete_run(run_id) + except replay.ReplayBusy as e: + raise HTTPException(409, str(e)) + return {"ok": True} + + +def _call_row(c: dict) -> dict: + scenes = c.get("scenes") or {} + paths = [((s.get("corr_debug") or {}).get("corr_path")) for _, s in sorted(scenes.items())] + return { + "call_id": c.get("call_id"), + "started_at": c.get("started_at"), + "talkgroup_name": c.get("talkgroup_name"), + "transcript": c.get("transcript_corrected") or c.get("transcript"), + "units": c.get("units"), + "cleared_units": c.get("cleared_units"), + "location": c.get("location"), + "skip_reason": c.get("skip_reason"), + "corr_path": [p for p in paths if p] or ([c["corr_path"]] if c.get("corr_path") else []), + "incident_ids": c.get("incident_ids") or [], + } + + +@router.get("/{run_id}/incidents") +async def run_incidents(run_id: str, decoded: dict = Depends(require_admin_token)): + """ + A run's sandbox, shaped for reading: every incident with its calls in + order, plus the calls that never linked. Embeddings stay out. + """ + await _own_run(run_id, await _org(decoded)) + incidents, calls = await replay.sandbox_contents(run_id) + by_id = {c.get("call_id"): _call_row(c) for c in calls} + out = [] + for inc in sorted(incidents, key=lambda i: str(i.get("started_at") or "")): + rows = [by_id[cid] for cid in (inc.get("call_ids") or []) if cid in by_id] + rows.sort(key=lambda r: str(r["started_at"] or "")) + out.append({ + "incident_id": inc.get("incident_id"), + "title": inc.get("title"), + "type": inc.get("type"), + "severity": inc.get("severity"), + "status": inc.get("status"), + "resolved_via": inc.get("resolved_via"), + "started_at": inc.get("started_at"), + "updated_at": inc.get("updated_at"), + "resolved_at": inc.get("resolved_at"), + "location": inc.get("location"), + "location_coords": inc.get("location_coords"), + "units": inc.get("units"), + "units_active": inc.get("units_active"), + "units_cleared": inc.get("units_cleared"), + "talkgroup_ids": inc.get("talkgroup_ids"), + "calls": rows, + }) + orphans = sorted((r for r in by_id.values() if not r["incident_ids"]), + key=lambda r: str(r["started_at"] or "")) + return {"incidents": out, "orphans": orphans} diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index 1ea9598..96e90c0 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/drb-c2-core/app/routers/upload.py @@ -1,11 +1,11 @@ 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 import clock from app.internal.logger import logger from app.config import settings @@ -140,7 +140,7 @@ def _recent_incident_on_same_talkgroup(ctx: dict) -> bool: 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) + now = ctx.get("now") or clock.now() 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 []): @@ -374,7 +374,8 @@ async def _run_extraction_pipeline( if scene["resolved"] and incident_id: await fstore.doc_set("incidents", incident_id, { "status": "resolved", - "resolved_at": datetime.now(timezone.utc).isoformat(), + "resolved_at": clock.now().isoformat(), + "resolved_via": "llm_closure", }) await incident_correlator.maybe_resolve_parent(incident_id) logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)") @@ -396,6 +397,113 @@ async def _run_extraction_pipeline( ) +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, incident_correlator + + # 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. + # A single recording can produce multiple incidents on a busy channel. + incident_ids: list[str] = [] + all_tags: list[str] = [] + # 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": clock.now().isoformat(), + "resolved_via": "llm_closure", + }) + await incident_correlator.maybe_resolve_parent(incident_id) + logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)") + + # Correlator also runs for calls with no scenes (unclassified) to attempt + # talkgroup-based linking even when no transcript could be produced. + # transcript_too_short (<=5 words: "10-8", "show me clear", a unit + # check-in) still carries a real transcript and talkgroup — exactly the + # brief follow-up/clearance traffic an incident needs, and the thin-path + # merge below already requires a same-talkgroup, recently-active + # incident before attaching anything, same guard already trusted for + # no-transcript calls. Previously excluded here, so these calls never + # attached to anything at all. garbage_transcript (Whisper + # hallucination) has no real content behind it and stays excluded. + if not scenes: + _call_doc = await fstore.doc_get("calls", call_id) + skip_reason = (_call_doc or {}).get("skip_reason") + if not skip_reason or skip_reason == "transcript_too_short": + incident_id = await _correlate_with_consensus( + call_id=call_id, + node_id=node_id, + system_id=system_id, + talkgroup_id=talkgroup_id, + talkgroup_name=talkgroup_name, + tags=[], + incident_type=None, + location=None, + location_coords=None, + ) + if incident_id: + incident_ids.append(incident_id) + + if incident_ids: + await fstore.doc_set("calls", call_id, {"incident_ids": incident_ids}) + + return incident_ids, all_tags, scenes + + async def _run_intelligence_pipeline( call_id: str, node_id: str, @@ -411,7 +519,7 @@ async def _run_intelligence_pipeline( 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 + 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). @@ -427,7 +535,7 @@ async def _run_intelligence_pipeline( # 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() + "intelligence_started_at": clock.now().isoformat() }) except Exception as e: logger.warning(f"Could not mark intelligence_started_at for call {call_id}: {e}") @@ -466,90 +574,22 @@ async def _run_intelligence_pipeline( 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. + # Steps 2-3: scene extraction + correlation. incident_ids: list[str] = [] all_tags: list[str] = [] if _flag("correlation_enabled"): - # server-26#96: scene_index is threaded through so each scene's - # corr_debug/transcript lands in its own entry of the call doc's - # `scenes` map instead of clobbering every other scene's write. - for scene_index, scene in enumerate(scenes): - all_tags.extend(scene["tags"]) - is_reassignment = bool(scene.get("reassignment")) - corr_units = [] if is_reassignment else scene.get("units") - incident_id = await _correlate_with_consensus( - call_id=call_id, - node_id=node_id, - system_id=system_id, - talkgroup_id=talkgroup_id, - talkgroup_name=talkgroup_name, - tags=scene["tags"], - incident_type=scene["incident_type"], - location=scene["location"], - location_coords=scene["location_coords"], - units=corr_units, - vehicles=scene.get("vehicles"), - cleared_units=scene.get("cleared_units"), - reassignment=is_reassignment, - embedding=scene.get("embedding"), - severity=scene.get("severity"), - transcript=scene.get("transcript"), - scene_index=scene_index, - ) - if incident_id and incident_id not in incident_ids: - incident_ids.append(incident_id) - if scene["resolved"] and incident_id: - await fstore.doc_set("incidents", incident_id, { - "status": "resolved", - "resolved_at": datetime.now(timezone.utc).isoformat(), - }) - await incident_correlator.maybe_resolve_parent(incident_id) - logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)") - - # Correlator also runs for calls with no scenes (unclassified) to attempt - # talkgroup-based linking even when no transcript could be produced. - # transcript_too_short (<=5 words: "10-8", "show me clear", a unit - # check-in) still carries a real transcript and talkgroup — exactly the - # brief follow-up/clearance traffic an incident needs, and the thin-path - # merge below already requires a same-talkgroup, recently-active - # incident before attaching anything, same guard already trusted for - # no-transcript calls. Previously excluded here, so these calls never - # attached to anything at all. garbage_transcript (Whisper - # hallucination) has no real content behind it and stays excluded. - if not scenes: - _call_doc = await fstore.doc_get("calls", call_id) - skip_reason = (_call_doc or {}).get("skip_reason") - if not skip_reason or skip_reason == "transcript_too_short": - incident_id = await _correlate_with_consensus( - call_id=call_id, - node_id=node_id, - system_id=system_id, - talkgroup_id=talkgroup_id, - talkgroup_name=talkgroup_name, - tags=[], - incident_type=None, - location=None, - location_coords=None, - ) - if incident_id: - incident_ids.append(incident_id) - - if incident_ids: - await fstore.doc_set("calls", call_id, {"incident_ids": incident_ids}) + 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, + ) + 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( diff --git a/drb-c2-core/tests/test_replay.py b/drb-c2-core/tests/test_replay.py new file mode 100644 index 0000000..49d7d04 --- /dev/null +++ b/drb-c2-core/tests/test_replay.py @@ -0,0 +1,371 @@ +""" +Replay (app/internal/replay.py): re-running the pipeline over past calls in a +sandbox. The properties that matter, in order: a replay never writes a live +call or incident; it runs the live correlation code with the clock pinned to +each call's own time; and a call seeded ahead of its turn is invisible to the +orphan sweep until it is processed. +""" +import asyncio +import copy +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +import pytest + +from app.internal import clock, replay +from app.internal import firestore as fstore +from app.internal.feature_flags import force_flags, resolve_flags, unforce_flags + + +# --------------------------------------------------------------------------- +# An in-memory Firestore that honours the sandbox redirect, so the real +# correlator can run against it. +# --------------------------------------------------------------------------- + +def _merge(dst: dict, src: dict) -> dict: + for k, v in src.items(): + if isinstance(v, dict) and isinstance(dst.get(k), dict): + _merge(dst[k], v) + else: + dst[k] = copy.deepcopy(v) + return dst + + +def _cmp(a, op, b) -> bool: + if a is None: + return False + if isinstance(a, str) and isinstance(b, datetime): + a = datetime.fromisoformat(a) + return {"==": a == b, ">=": a >= b, "<=": a <= b, ">": a > b, "<": a < b}[op] + + +class FakeStore: + def __init__(self): + self.data: dict[str, dict[str, dict]] = {} + + def coll(self, name: str) -> dict: + return self.data.setdefault(fstore._path(name), {}) + + async def doc_set(self, collection, doc_id, data, merge=True): + c = self.coll(collection) + if merge and doc_id in c: + _merge(c[doc_id], data) + else: + c[doc_id] = copy.deepcopy(data) + + async def doc_update(self, collection, doc_id, data): + await self.doc_set(collection, doc_id, data) + + async def doc_get(self, collection, doc_id): + d = self.coll(collection).get(doc_id) + return copy.deepcopy(d) if d is not None else None + + async def doc_get_cached(self, collection, doc_id, ttl=300.0): + return await self.doc_get(collection, doc_id) + + async def doc_delete(self, collection, doc_id): + self.coll(collection).pop(doc_id, None) + + async def collection_list(self, collection, **filters): + return [copy.deepcopy(d) for d in self.coll(collection).values() + if all(d.get(k) == v for k, v in filters.items())] + + async def collection_where(self, collection, conditions, order_by=None, + limit_to=None, start_after=None): + rows = [copy.deepcopy(d) for d in self.coll(collection).values() + if all(_cmp(d.get(f), op, v) for f, op, v in conditions)] + for field, direction in reversed(order_by or []): + rows.sort(key=lambda d: d.get(field), reverse=direction == "DESCENDING") + return rows[:limit_to] if limit_to else rows + + +@pytest.fixture +def store(): + s = FakeStore() + names = ("doc_set", "doc_update", "doc_get", "doc_get_cached", "doc_delete", + "collection_list", "collection_where") + patches = [patch.object(fstore, n, getattr(s, n)) for n in names] + for p in patches: + p.start() + replay._active_run_id = None + replay._active_task = None + yield s + for p in patches: + p.stop() + + +# --------------------------------------------------------------------------- +# The context-scoped pieces +# --------------------------------------------------------------------------- + +def test_sandbox_redirects_only_calls_and_incidents(): + assert fstore._path("calls") == "calls" + tok = fstore.enter_sandbox("replay_runs/r1") + try: + assert fstore._path("calls") == "replay_runs/r1/calls" + assert fstore._path("incidents") == "replay_runs/r1/incidents" + assert fstore._path("systems") == "systems" + assert fstore._path("config") == "config" + finally: + fstore.exit_sandbox(tok) + assert fstore._path("incidents") == "incidents" + + +@pytest.mark.asyncio +async def test_sandbox_and_clock_do_not_leak_into_a_concurrent_task(): + """A replay runs beside live uploads in one event loop. The live task + must see the real collections and the real clock.""" + pinned = datetime(2026, 9, 21, 12, 0, tzinfo=timezone.utc) + seen = {} + replay_entered = asyncio.Event() + live_checked = asyncio.Event() + + async def replay_task(): + fstore.enter_sandbox("replay_runs/r1") + clock.pin(pinned) + replay_entered.set() + await live_checked.wait() + seen["replay"] = (fstore._path("calls"), clock.now()) + + async def live_task(): + await replay_entered.wait() + seen["live"] = (fstore._path("calls"), clock.now()) + live_checked.set() + + await asyncio.gather(replay_task(), live_task()) + assert seen["replay"] == ("replay_runs/r1/calls", pinned) + assert seen["live"][0] == "calls" + assert seen["live"][1] != pinned + + +@pytest.mark.asyncio +async def test_forced_flags_override_global_switches(): + tok = force_flags({"correlation_enabled": True, "stt_enabled": False}) + try: + flags, flag = await resolve_flags("sys-1") + assert flag("correlation_enabled") is True + assert flag("stt_enabled") is False + assert flag("summaries_enabled") is False + finally: + unforce_flags(tok) + + +def test_sandbox_seed_strips_live_answers(): + call = { + "call_id": "c1", "org_id": "o", "talkgroup_id": 5, "srcaddr": 123, + "status": "ended", "transcript": "engine 5 responding", "segments": [{"t": 1}], + "incident_ids": ["live-inc"], "incident_id": "live-inc", "units": ["E5"], + "corr_path": "fast/thin", "scenes": {"0": {}}, "skip_reason": None, + "chatter_classifier_verdict": "x", "eval_transcript": "y", "embedding": [0.1], + } + seed = replay._sandbox_seed(call, "transcripts") + assert seed["transcript"] == "engine 5 responding" + assert seed["srcaddr"] == 123 + assert seed["status"] == "replay_pending" + for gone in ("incident_ids", "incident_id", "units", "corr_path", "scenes", + "chatter_classifier_verdict", "eval_transcript", "embedding"): + assert gone not in seed + assert "transcript" not in replay._sandbox_seed(call, "audio") + + +def test_compute_metrics_separates_timeout_from_real_clears(): + incidents = [ + {"call_ids": ["a"], "status": "resolved", "resolved_via": "idle_timeout"}, + {"call_ids": ["b", "c"], "status": "resolved", "resolved_via": "units_cleared", + "units_cleared": ["E5"]}, + {"call_ids": ["d", "e", "f"], "status": "active"}, + ] + calls = [ + {"call_id": "a", "incident_ids": ["1"], "scenes": {"0": {"corr_debug": { + "corr_path": "new", "corr_consensus": "rules_only"}}}}, + {"call_id": "z", "corr_path": "unlinked"}, + ] + m = replay.compute_metrics(incidents, calls) + assert m["incidents"] == 3 + assert m["single_call_incidents"] == 1 + assert m["resolved_via"] == {"idle_timeout": 1, "units_cleared": 1, "still_active": 1} + assert m["incidents_with_units_cleared"] == 1 + assert m["calls_orphaned"] == 1 + assert m["corr_path"] == {"new": 1, "unlinked": 1} + assert m["llm_decisions"] == 0 + + +# --------------------------------------------------------------------------- +# A whole run, through the real correlator +# --------------------------------------------------------------------------- + +T0 = datetime(2026, 9, 21, 14, 0, tzinfo=timezone.utc) + + +def _live_call(i: int, minute: int, transcript: str) -> dict: + return { + "call_id": f"call-{i}", "org_id": "org-1", "node_id": "node-1", + "system_id": "sys-1", "talkgroup_id": 100, "talkgroup_name": "Police Dispatch", + "started_at": T0 + timedelta(minutes=minute), + "ended_at": T0 + timedelta(minutes=minute, seconds=20), + "duration_s": 20, "status": "ended", + "transcript": transcript, + "incident_ids": ["LIVE-INCIDENT"], "corr_path": "fast/thin", + } + + +def _scene(transcript: str, units: list[str]) -> dict: + return { + "tags": ["mva"], "incident_type": "accident", "location": "Main Street", + "location_coords": None, "units": units, "vehicles": [], "cleared_units": [], + "reassignment": False, "embedding": None, "severity": "moderate", + "transcript": transcript, "resolved": False, + } + + +@pytest.mark.asyncio +async def test_run_writes_only_to_its_sandbox_and_pins_the_clock(store): + live = { + "call-1": _live_call(1, 0, "Car 12, MVA Main Street"), + "call-2": _live_call(2, 1, "Car 12 on scene Main Street"), + "call-3": _live_call(3, 300, "Car 40, alarm Oak Avenue"), + } + store.data["calls"] = copy.deepcopy(live) + store.data["incidents"] = {"LIVE-INCIDENT": {"incident_id": "LIVE-INCIDENT", "org_id": "org-1", + "status": "active", "call_ids": ["call-1"]}} + live_before = copy.deepcopy(store.data) + + extracted = [] + + async def fake_extract(call_id, transcript, talkgroup_name, **kw): + extracted.append(call_id) + # Prefetch seeds calls ahead of the clock; they must not look "ended" yet. + return [_scene(transcript, ["Car 12"] if "12" in transcript else ["Car 40"])] + + with patch("app.internal.intelligence.extract_scenes", fake_extract): + calls, truncated = await replay.select_calls( + "org-1", T0 - timedelta(hours=1), T0 + timedelta(hours=6)) + assert [c["call_id"] for c in calls] == ["call-1", "call-2", "call-3"] + assert not truncated + await replay.start_run( + org_id="org-1", date_from=T0 - timedelta(hours=1), + date_to=T0 + timedelta(hours=6), mode="transcripts", system_ids=None, + source_run_id=None, label="t", actor="test", + ) + await replay._active_task + + # Live collections are exactly as they were. + assert store.data["calls"] == live_before["calls"] + assert store.data["incidents"] == live_before["incidents"] + + run = next(iter(store.data["replay_runs"].values())) + assert run["status"] == "done", run["errors"] + root = f"replay_runs/{run['run_id']}" + sb_calls = store.data[f"{root}/calls"] + sb_incidents = store.data[f"{root}/incidents"] + assert sorted(extracted) == ["call-1", "call-2", "call-3"] + assert all(c["status"] == "ended" for c in sb_calls.values()) + assert "LIVE-INCIDENT" not in sb_incidents + + # Incident timestamps come from the replayed calls, not the wall clock. + for inc in sb_incidents.values(): + started = datetime.fromisoformat(inc["started_at"]) + assert T0 <= started <= T0 + timedelta(hours=6) + # The two Car 12 calls are one job; the Car 40 call five hours later is another. + groups = sorted(sorted(i["call_ids"]) for i in sb_incidents.values()) + assert groups == [["call-1", "call-2"], ["call-3"]] + # Each aged out on the replayed clock the way it would have live — + # incident_auto_resolve_minutes after its last activity, not "now". + assert run["metrics"]["resolved_via"] == {"idle_timeout": 2} + first = next(i for i in sb_incidents.values() if "call-1" in i["call_ids"]) + idle = datetime.fromisoformat(first["resolved_at"]) - datetime.fromisoformat(first["updated_at"]) + assert timedelta(minutes=90) < idle <= timedelta(minutes=95) + assert run["metrics"]["calls"] == 3 + assert set(store.data[f"{root}/scenes"]) == {"call-1", "call-2", "call-3"} + + +@pytest.mark.asyncio +async def test_reuse_mode_correlates_without_extracting(store): + store.data["calls"] = {"call-1": _live_call(1, 0, "Car 12, MVA Main Street")} + + async def fake_extract(call_id, transcript, talkgroup_name, **kw): + # What the real extract_scenes also does: write call-level fields. + await fstore.doc_set("calls", call_id, {"units": ["Car 12"], "tags": ["mva"]}) + return [_scene(transcript, ["Car 12"])] + + with patch("app.internal.intelligence.extract_scenes", fake_extract): + first = await replay.start_run( + org_id="org-1", date_from=T0 - timedelta(hours=1), date_to=T0 + timedelta(hours=1), + mode="transcripts", system_ids=None, source_run_id=None, label="", actor="t") + await replay._active_task + + async def must_not_extract(*a, **kw): + raise AssertionError("reuse mode re-ran extraction") + + with patch("app.internal.intelligence.extract_scenes", must_not_extract): + second = await replay.start_run( + org_id="org-1", date_from=T0 - timedelta(hours=1), date_to=T0 + timedelta(hours=1), + mode="reuse", system_ids=None, source_run_id=first["run_id"], label="", actor="t") + await replay._active_task + + run = store.data["replay_runs"][second["run_id"]] + assert run["status"] == "done", run["errors"] + assert run["progress"]["errors"] == 0 + assert run["metrics"]["calls_linked"] == 1 + # Extraction's call-level output came across too — the orphan sweep reads + # units/tags/location off the call doc, not off the scenes. + sb_call = store.data[f"replay_runs/{second['run_id']}/calls"]["call-1"] + assert sb_call["units"] == ["Car 12"] + assert sb_call["tags"] == ["mva"] + + +@pytest.mark.asyncio +async def test_one_run_at_a_time(store): + store.data["calls"] = {"call-1": _live_call(1, 0, "x")} + gate = asyncio.Event() + + async def slow_extract(*a, **kw): + await gate.wait() + return [] + + with patch("app.internal.intelligence.extract_scenes", slow_extract): + await replay.start_run( + org_id="org-1", date_from=T0 - timedelta(hours=1), date_to=T0 + timedelta(hours=1), + mode="transcripts", system_ids=None, source_run_id=None, label="", actor="t") + with pytest.raises(replay.ReplayBusy): + await replay.start_run( + org_id="org-1", date_from=T0 - timedelta(hours=1), date_to=T0 + timedelta(hours=1), + mode="transcripts", system_ids=None, source_run_id=None, label="", actor="t") + gate.set() + await replay._active_task + + +def test_stored_input_rebuilds_from_corrector_segments(): + """Live extraction overwrites transcript_corrected with scene 0's text; + the corrector's own output survives in segments_corrected.""" + call = { + "transcript": "raw whisper", + "transcript_corrected": "scene zero only", + "segments": [{"text": "raw a"}, {"text": "raw b"}], + "segments_corrected": [{"text": "fixed a"}, {"text": "fixed b"}], + } + text, segs = replay._stored_input(call) + assert text == "fixed a fixed b" + assert segs == call["segments_corrected"] + assert replay._stored_input({"transcript": "raw", "segments": []}) == ("raw", []) + assert replay._stored_input({"transcript": "hum", "transcript_not_speech": True}) == (None, []) + + +@pytest.mark.asyncio +async def test_replay_never_touches_live_ai_health_or_review_queue(): + from app.internal import ai_health, area_context + + before = ai_health.snapshot() + tok = fstore.enter_sandbox("replay_runs/r1") + try: + with patch.object(ai_health, "_post_webhook") as hook, \ + patch.object(fstore, "doc_get") as get: + for _ in range(10): + await ai_health.report_degraded("correlation_cheap", "gemini", "m", "429", "wait") + await ai_health.report_healthy("transcription") + assert await area_context.add_pending("sys-1", 5, [{"term": "x"}]) == 0 + hook.assert_not_called() + get.assert_not_called() + finally: + fstore.exit_sandbox(tok) + assert ai_health.snapshot() == before diff --git a/drb-frontend/app/admin/page.tsx b/drb-frontend/app/admin/page.tsx index 35d132b..df178c3 100644 --- a/drb-frontend/app/admin/page.tsx +++ b/drb-frontend/app/admin/page.tsx @@ -2,6 +2,7 @@ import { useAuth } from "@/components/AuthProvider"; import { c2api } from "@/lib/c2api"; +import { ReplayTab } from "@/components/admin/ReplayTab"; import { useEffect, useState, useRef, useCallback } from "react"; import { useRouter } from "next/navigation"; import type { UserRecord, AuditEntry, UserRole, CallRecord } from "@/lib/types"; @@ -1227,11 +1228,12 @@ function SttEvalTab() { // Main admin page // --------------------------------------------------------------------------- -type AdminTab = "features" | "correlation" | "users" | "audit" | "calls" | "eval"; +type AdminTab = "features" | "correlation" | "replay" | "users" | "audit" | "calls" | "eval"; const TAB_LABELS: { key: AdminTab; label: string }[] = [ { key: "features", label: "AI Features" }, { key: "correlation", label: "Correlation Debug" }, + { key: "replay", label: "Replay" }, { key: "calls", label: "Calls" }, { key: "eval", label: "STT Eval" }, { key: "users", label: "Users" }, @@ -1256,7 +1258,7 @@ export default function AdminPage() { if (!isAdmin) return null; // Users/Audit tabs benefit from full width; everything else is narrow - const wide = tab === "users" || tab === "audit"; + const wide = tab === "users" || tab === "audit" || tab === "replay"; return (
@@ -1278,6 +1280,7 @@ export default function AdminPage() { {tab === "features" && } {tab === "correlation" && } + {tab === "replay" && } {tab === "calls" && } {tab === "eval" && } {tab === "users" && } diff --git a/drb-frontend/components/admin/ReplayTab.tsx b/drb-frontend/components/admin/ReplayTab.tsx new file mode 100644 index 0000000..8f5363b --- /dev/null +++ b/drb-frontend/components/admin/ReplayTab.tsx @@ -0,0 +1,439 @@ +"use client"; + +// --------------------------------------------------------------------------- +// Replay — re-run the intelligence pipeline over a past time range into a +// sandbox, then compare runs. Backend: drb-c2-core/app/internal/replay.py. +// Nothing here touches a live call or incident; every run spends real AI +// credits, so the form estimates first and only then offers Start. +// --------------------------------------------------------------------------- + +import { useCallback, useEffect, useState } from "react"; +import { c2api } from "@/lib/c2api"; +import { useSystems } from "@/lib/useSystems"; +import type { + ReplayEstimate, ReplayIncident, ReplayIncidents, ReplayMode, ReplayRun, ReplayCallRow, +} from "@/lib/types"; + +const MODES: { key: ReplayMode; label: string; help: string }[] = [ + { key: "transcripts", label: "Saved transcripts", help: "Reuse each call's transcript; re-run extraction + correlation." }, + { key: "audio", label: "Re-transcribe audio", help: "Whisper the saved audio again, then extract + correlate. For ranges where AI was off." }, + { key: "reuse", label: "Correlation only", help: "Reuse an earlier run's extraction; re-run correlation only. Isolates a correlator change." }, +]; + +// Clears that came from the radio traffic itself, vs the 90-minute idle timer. +const SIGNAL_RESOLVES = ["units_cleared", "llm_closure", "reassignment", "children_resolved"]; + +const input = + "bg-gray-800 border border-gray-700 rounded-lg px-3 py-1.5 text-white text-sm font-mono focus:outline-none focus:border-indigo-500"; +const btn = + "bg-gray-800 hover:bg-gray-700 disabled:opacity-50 border border-gray-700 text-white text-sm font-mono px-4 py-1.5 rounded-lg transition-colors"; + +function toLocalInput(d: Date): string { + const pad = (n: number) => String(n).padStart(2, "0"); + return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}T${pad(d.getHours())}:${pad(d.getMinutes())}`; +} + +function fmtTime(iso: string | null | undefined): string { + if (!iso) return "—"; + return new Date(iso).toLocaleString([], { month: "short", day: "numeric", hour: "2-digit", minute: "2-digit" }); +} + +function sum(rec: Record | undefined, keys: string[]): number { + return keys.reduce((n, k) => n + (rec?.[k] ?? 0), 0); +} + +// --------------------------------------------------------------------------- + +function NewRunForm({ runs, busy, onStarted }: { runs: ReplayRun[]; busy: boolean; onStarted: () => void }) { + const { systems } = useSystems(); + const [from, setFrom] = useState(() => toLocalInput(new Date(Date.now() - 6 * 3600_000))); + const [to, setTo] = useState(() => toLocalInput(new Date())); + const [mode, setMode] = useState("transcripts"); + const [systemIds, setSystemIds] = useState([]); + const [sourceRun, setSourceRun] = useState(""); + const [label, setLabel] = useState(""); + const [est, setEst] = useState(null); + const [working, setWorking] = useState(false); + const [error, setError] = useState(null); + + // Any change to what would be replayed invalidates the estimate. + useEffect(() => { setEst(null); }, [from, to, mode, systemIds]); + + const sources = runs.filter((r) => r.status === "done" && r.mode !== "reuse"); + const range = () => ({ date_from: new Date(from).toISOString(), date_to: new Date(to).toISOString() }); + + async function estimate() { + setWorking(true); setError(null); + try { + setEst(await c2api.estimateReplay({ ...range(), mode, system_ids: systemIds })); + } catch (e) { setError(String(e)); } finally { setWorking(false); } + } + + async function start() { + setWorking(true); setError(null); + try { + await c2api.startReplay({ + ...range(), mode, system_ids: systemIds, label, + source_run_id: mode === "reuse" ? sourceRun : null, + }); + setEst(null); setLabel(""); + onStarted(); + } catch (e) { setError(String(e)); } finally { setWorking(false); } + } + + function toggleSystem(id: string) { + setSystemIds((s) => (s.includes(id) ? s.filter((x) => x !== id) : [...s, id])); + } + + const canStart = est && !est.truncated && est.calls > 0 && !busy && (mode !== "reuse" || sourceRun); + + return ( +
+
+
+ + setFrom(e.target.value)} className={input} /> +
+
+ + setTo(e.target.value)} className={input} /> +
+
+ + setLabel(e.target.value)} placeholder="what changed?" className={`${input} w-56`} /> +
+
+ +
+ {MODES.map((m) => ( + + ))} +
+ + {mode === "reuse" && ( +
+ + +

Set the range to match that run; calls outside it are skipped.

+
+ )} + + {systems.length > 1 && ( +
+ Systems (none = all): + {systems.map((s) => ( + + ))} +
+ )} + +
+ + + {busy && a run is in progress} +
+ + {est && ( +

+ {est.truncated ? ( + More than {est.max_calls} calls — narrow the range. + ) : ( + <> + {est.calls} calls ·{" "} + {est.calls_with_transcript} with transcripts · {est.audio_minutes} audio min ·{" "} + ~${est.est_cost_usd.toFixed(2)} est. + {mode === "transcripts" && est.calls_with_transcript < est.calls / 2 && ( + — most calls have no transcript; consider Re-transcribe audio. + )} + + )} +

+ )} + + {error &&

{error}

} +
+ ); +} + +// --------------------------------------------------------------------------- + +function RunsTable({ + runs, activeId, selected, onSelect, onChanged, +}: { + runs: ReplayRun[]; activeId: string | null; selected: string | null; + onSelect: (id: string) => void; onChanged: () => void; +}) { + async function cancel(id: string) { + await c2api.cancelReplay(id).catch(() => undefined); + onChanged(); + } + async function remove(id: string) { + if (!window.confirm("Delete this run and its sandbox incidents?")) return; + await c2api.deleteReplay(id).catch(() => undefined); + onChanged(); + } + + if (!runs.length) return

No runs yet.

; + + const th = "text-left text-xs text-gray-500 font-normal px-2 py-1.5 whitespace-nowrap"; + const td = "px-2 py-1.5 whitespace-nowrap"; + return ( +
+ + + + + + + + + + + + + + + + + {runs.map((r) => { + const m = r.metrics; + const running = r.run_id === activeId; + return ( + onSelect(r.run_id)} + className={`border-b border-gray-800/60 cursor-pointer ${selected === r.run_id ? "bg-gray-800/60" : "hover:bg-gray-900"}`} + > + + + + + + + + + + + + ); + })} + +
RunRangeStatusInc1-callOrphansReal clears / timeoutsLLMCost
+
{r.label || r.run_id}
+
{r.mode} · {r.git_sha?.slice(0, 7)} · {fmtTime(r.created_at)}
+ 		{fmtTime(r.date_from)} → {fmtTime(r.date_to)} + {running ? ( + {r.progress.done}/{r.progress.total} + ) : ( + {r.status} + )} + {r.progress.errors > 0 && · {r.progress.errors} err} + {m?.incidents ?? "—"}{m?.single_call_pct != null ? `${m.single_call_pct}%` : "—"}{m ? `${m.calls_orphaned}/${m.calls}` : "—"} + {m ? ( + <> + {sum(m.resolved_via, SIGNAL_RESOLVES)} + {" / "} + {m.resolved_via.idle_timeout ?? 0} + + ) : "—"} + {m?.llm_decisions ?? "—"}{m ? `$${m.est_cost_usd.toFixed(2)}` : `~$${r.estimate.est_cost_usd.toFixed(2)}`} e.stopPropagation()}> + {running ? ( + + ) : ( + + )} +
+
+ ); +} + +// --------------------------------------------------------------------------- + +function CallLine({ call }: { call: ReplayCallRow }) { + const [audio, setAudio] = useState(null); + async function play() { + try { + const c = await c2api.getCall(call.call_id); + setAudio(c.audio_url); + } catch { /* audio is a convenience */ } + } + return ( +
+
+ {fmtTime(call.started_at)} + {call.talkgroup_name} + {call.corr_path.map((p, i) => {p})} + {call.units?.length ? units {call.units.join(", ")} : null} + {call.cleared_units?.length ? cleared {call.cleared_units.join(", ")} : null} + {call.skip_reason && {call.skip_reason}} + {audio ? ( +
+
{call.transcript || (no transcript)}
+
+ ); +} + +function IncidentCard({ inc }: { inc: ReplayIncident }) { + const [open, setOpen] = useState(false); + const signal = inc.resolved_via && SIGNAL_RESOLVES.includes(inc.resolved_via); + return ( +
+ + {open &&
{inc.calls.map((c) => )}
} +
+ ); +} + +function RunDetail({ run }: { run: ReplayRun }) { + const [data, setData] = useState(null); + const [error, setError] = useState(null); + const [filter, setFilter] = useState<"all" | "multi" | "single">("all"); + const [showOrphans, setShowOrphans] = useState(false); + + useEffect(() => { + setData(null); setError(null); + if (run.status === "running") return; + c2api.getReplayIncidents(run.run_id).then(setData).catch((e) => setError(String(e))); + }, [run.run_id, run.status]); + + const m = run.metrics; + const shown = (data?.incidents ?? []).filter((i) => + filter === "all" ? true : filter === "multi" ? i.calls.length > 1 : i.calls.length === 1, + ); + + return ( +
+ {m && ( +
+ {[ + ["Linked calls", `${m.calls_linked}/${m.calls}`], + ["Median calls / incident", m.median_calls_per_incident ?? "—"], + ["With units cleared", m.incidents_with_units_cleared], + ["With map pin", m.incidents_with_coords], + ].map(([k, v]) => ( +
+
{k}
+
{v}
+
+ ))} +
+ )} + {m && ( +

+ resolved: {Object.entries(m.resolved_via).map(([k, v]) => `${k} ${v}`).join(" · ")} +
+ paths: {Object.entries(m.corr_path).map(([k, v]) => `${k} ${v}`).join(" · ")} +

+ )} + {run.errors?.length > 0 && ( +
+ {run.errors.length} error(s) + {run.errors.map((e, i) =>
{e}
)} +
+ )} + {run.status === "running" &&

Incidents appear when the run finishes.

} + {error &&

{error}

} + {data && ( + <> +
+ {(["all", "multi", "single"] as const).map((f) => ( + + ))} + +
+ {showOrphans ? ( +
+ {data.orphans.map((c) => )} +
+ ) : ( +
{shown.map((i) => )}
+ )} + + )} +
+ ); +} + +// --------------------------------------------------------------------------- + +export function ReplayTab() { + const [runs, setRuns] = useState([]); + const [activeId, setActiveId] = useState(null); + const [selected, setSelected] = useState(null); + const [error, setError] = useState(null); + + const load = useCallback(async () => { + try { + const res = await c2api.listReplays(); + setRuns(res.runs); + setActiveId(res.active_run_id); + setError(null); + } catch (e) { setError(String(e)); } + }, []); + + useEffect(() => { load(); }, [load]); + // Poll only while something is running. + useEffect(() => { + if (!activeId) return; + const t = setInterval(load, 4000); + return () => clearInterval(t); + }, [activeId, load]); + + const run = runs.find((r) => r.run_id === selected) ?? null; + + return ( +
+

+ Re-run the pipeline over a past time range, in the calls' original order, into a sandbox — live incidents are never + touched. Replay the same range after each change and compare the rows below. A run starts with no incidents + open, so its first ~90 minutes split more than live did — compare runs over the same range, not a run against live. +

+ + {error &&

{error}

} + + {run && } +
+ ); +} diff --git a/drb-frontend/lib/c2api.ts b/drb-frontend/lib/c2api.ts index 4a2b575..8a90286 100644 --- a/drb-frontend/lib/c2api.ts +++ b/drb-frontend/lib/c2api.ts @@ -233,6 +233,28 @@ export const c2api = { getCorrelationDebug: (limit: number, orphanHours: number) => request(`/admin/debug/correlation?limit=${limit}&orphan_hours=${orphanHours}`), + // Replay (admin) — re-run the pipeline over past calls into a sandbox. + // See drb-c2-core/app/internal/replay.py. + estimateReplay: (p: { date_from: string; date_to: string; mode: import("@/lib/types").ReplayMode; system_ids?: string[] }) => { + const qs = new URLSearchParams({ date_from: p.date_from, date_to: p.date_to, mode: p.mode }); + if (p.system_ids?.length) qs.set("system_ids", p.system_ids.join(",")); + return request(`/admin/replay/estimate?${qs}`); + }, + startReplay: (body: { + date_from: string; date_to: string; mode: import("@/lib/types").ReplayMode; + system_ids?: string[]; source_run_id?: string | null; label?: string; + }) => + request("/admin/replay", { method: "POST", body: JSON.stringify(body) }), + listReplays: () => + request<{ runs: import("@/lib/types").ReplayRun[]; active_run_id: string | null }>("/admin/replay"), + getReplay: (runId: string) => request(`/admin/replay/${runId}`), + cancelReplay: (runId: string) => + request<{ ok: boolean }>(`/admin/replay/${runId}/cancel`, { method: "POST" }), + deleteReplay: (runId: string) => + request<{ ok: boolean }>(`/admin/replay/${runId}`, { method: "DELETE" }), + getReplayIncidents: (runId: string) => + request(`/admin/replay/${runId}/incidents`), + // Preferred bot token per system setPreferredToken: (tokenId: string, systemId: string) => request<{ ok: boolean; preferred_for_system_id: string | null }>(`/tokens/${tokenId}/prefer/${systemId}`, { method: "PUT" }), diff --git a/drb-frontend/lib/types.ts b/drb-frontend/lib/types.ts index c202217..2408754 100644 --- a/drb-frontend/lib/types.ts +++ b/drb-frontend/lib/types.ts @@ -306,3 +306,93 @@ export interface TalkgroupPending { talkgroup_name?: string; pending: PendingLocalTerm[]; } + +// --------------------------------------------------------------------------- +// Replay (admin) — drb-c2-core/app/internal/replay.py +// --------------------------------------------------------------------------- + +export type ReplayMode = "audio" | "transcripts" | "reuse"; + +export interface ReplayEstimate { + calls: number; + calls_with_transcript: number; + calls_with_audio: number; + audio_minutes: number; + est_cost_usd: number; + truncated: boolean; + max_calls: number; +} + +export interface ReplayMetrics { + calls: number; + calls_linked: number; + calls_orphaned: number; + incidents: number; + single_call_incidents: number; + single_call_pct: number | null; + median_calls_per_incident: number | null; + max_calls_in_incident: number | null; + incidents_with_units_cleared: number; + incidents_with_coords: number; + resolved_via: Record; + corr_path: Record; + corr_consensus: Record; + llm_decisions: number; + est_cost_usd: number; +} + +export interface ReplayRun { + run_id: string; + label: string; + mode: ReplayMode; + source_run_id: string | null; + date_from: string; + date_to: string; + system_ids: string[]; + git_sha: string; + created_by: string; + created_at: string; + finished_at?: string; + status: "running" | "done" | "cancelled" | "failed" | "interrupted"; + estimate: Omit; + progress: { total: number; done: number; errors: number; skipped?: number }; + metrics: ReplayMetrics | null; + errors: string[]; +} + +export interface ReplayCallRow { + call_id: string; + started_at: string; + talkgroup_name: string | null; + transcript: string | null; + units: string[] | null; + cleared_units: string[] | null; + location: string | null; + skip_reason: string | null; + corr_path: string[]; + incident_ids: string[]; +} + +export interface ReplayIncident { + incident_id: string; + title: string | null; + type: string | null; + severity: string | null; + status: string; + resolved_via: string | null; + started_at: string; + updated_at: string | null; + resolved_at: string | null; + location: string | null; + location_coords: { lat: number; lng: number } | null; + units: string[] | null; + units_active: string[] | null; + units_cleared: string[] | null; + talkgroup_ids: number[] | null; + calls: ReplayCallRow[]; +} + +export interface ReplayIncidents { + incidents: ReplayIncident[]; + orphans: ReplayCallRow[]; +}