Merge pull request 'Admin Replay: re-run the pipeline over past calls in a sandbox' (#171) from feat/replay into main

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]

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)

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)

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 = {}

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/<id>") 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]:

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} "

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.

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,
+    }

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)

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

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

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)

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}

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,91 +574,23 @@ 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] = []
+    # Steps 2-3: scene extraction + correlation.
+    incident_ids: list[str] = []
+    all_tags: list[str] = []
     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,
+        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 3: Correlate each scene independently.
-    # A single recording can produce multiple incidents on a busy channel.
-    incident_ids: list[str] = []
-    all_tags: list[str] = []
-    if _flag("correlation_enabled"):
-        # server-26#96: scene_index is threaded through so each scene's
-        # corr_debug/transcript lands in its own entry of the call doc's
-        # `scenes` map instead of clobbering every other scene's write.
-        for scene_index, scene in enumerate(scenes):
-            all_tags.extend(scene["tags"])
-            is_reassignment = bool(scene.get("reassignment"))
-            corr_units = [] if is_reassignment else scene.get("units")
-            incident_id = await _correlate_with_consensus(
-                call_id=call_id,
-                node_id=node_id,
-                system_id=system_id,
-                talkgroup_id=talkgroup_id,
-                talkgroup_name=talkgroup_name,
-                tags=scene["tags"],
-                incident_type=scene["incident_type"],
-                location=scene["location"],
-                location_coords=scene["location_coords"],
-                units=corr_units,
-                vehicles=scene.get("vehicles"),
-                cleared_units=scene.get("cleared_units"),
-                reassignment=is_reassignment,
-                embedding=scene.get("embedding"),
-                severity=scene.get("severity"),
-                transcript=scene.get("transcript"),
-                scene_index=scene_index,
-            )
-            if incident_id and incident_id not in incident_ids:
-                incident_ids.append(incident_id)
-            if scene["resolved"] and incident_id:
-                await fstore.doc_set("incidents", incident_id, {
-                    "status": "resolved",
-                    "resolved_at": datetime.now(timezone.utc).isoformat(),
-                })
-                await incident_correlator.maybe_resolve_parent(incident_id)
-                logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)")
-
-        # Correlator also runs for calls with no scenes (unclassified) to attempt
-        # talkgroup-based linking even when no transcript could be produced.
-        # 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})
-
     # Step 4: Alert dispatch (always runs — talkgroup ID rules don't need a transcript)
     await alerter.check_and_dispatch(
         call_id=call_id,

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

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 (
     <div className={`space-y-6 ${wide ? "" : "max-w-2xl"}`}>
@@ -1278,6 +1280,7 @@ export default function AdminPage() {
 
       {tab === "features"    && <FeaturesTab />}
       {tab === "correlation" && <CorrelationDebugTab />}
+      {tab === "replay"      && <ReplayTab />}
       {tab === "calls"       && <StaleCallsTab />}
       {tab === "eval"        && <SttEvalTab />}
       {tab === "users"       && <UsersTab currentUid={user?.uid ?? ""} />}

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<string, number> | 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<ReplayMode>("transcripts");
+  const [systemIds, setSystemIds] = useState<string[]>([]);
+  const [sourceRun, setSourceRun] = useState("");
+  const [label, setLabel] = useState("");
+  const [est, setEst] = useState<ReplayEstimate | null>(null);
+  const [working, setWorking] = useState(false);
+  const [error, setError] = useState<string | null>(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 (
+    <div className="bg-gray-900 border border-gray-800 rounded-xl p-4 space-y-4">
+      <div className="flex flex-wrap items-end gap-4">
+        <div>
+          <label className="text-xs text-gray-400 block mb-1">From</label>
+          <input type="datetime-local" value={from} max={to} onChange={(e) => setFrom(e.target.value)} className={input} />
+        </div>
+        <div>
+          <label className="text-xs text-gray-400 block mb-1">To</label>
+          <input type="datetime-local" value={to} min={from} onChange={(e) => setTo(e.target.value)} className={input} />
+        </div>
+        <div>
+          <label className="text-xs text-gray-400 block mb-1">Label</label>
+          <input value={label} onChange={(e) => setLabel(e.target.value)} placeholder="what changed?" className={`${input} w-56`} />
+        </div>
+      </div>
+
+      <div className="space-y-1">
+        {MODES.map((m) => (
+          <label key={m.key} className="flex items-start gap-2 text-sm font-mono cursor-pointer">
+            <input type="radio" checked={mode === m.key} onChange={() => setMode(m.key)} className="mt-1" />
+            <span className="text-white">{m.label}</span>
+            <span className="text-gray-500 text-xs mt-0.5">{m.help}</span>
+          </label>
+        ))}
+      </div>
+
+      {mode === "reuse" && (
+        <div>
+          <label className="text-xs text-gray-400 block mb-1">Reuse extraction from</label>
+          <select value={sourceRun} onChange={(e) => setSourceRun(e.target.value)} className={input}>
+            <option value="">— pick a finished run —</option>
+            {sources.map((r) => (
+              <option key={r.run_id} value={r.run_id}>
+                {r.label || r.run_id} · {fmtTime(r.date_from)} → {fmtTime(r.date_to)}
+              </option>
+            ))}
+          </select>
+          <p className="text-xs text-gray-500 mt-1">Set the range to match that run; calls outside it are skipped.</p>
+        </div>
+      )}
+
+      {systems.length > 1 && (
+        <div className="flex flex-wrap gap-3">
+          <span className="text-xs text-gray-400">Systems (none = all):</span>
+          {systems.map((s) => (
+            <label key={s.system_id} className="flex items-center gap-1 text-xs font-mono text-gray-300 cursor-pointer">
+              <input type="checkbox" checked={systemIds.includes(s.system_id)} onChange={() => toggleSystem(s.system_id)} />
+              {s.name}
+            </label>
+          ))}
+        </div>
+      )}
+
+      <div className="flex flex-wrap items-center gap-3">
+        <button onClick={estimate} disabled={working} className={btn}>{working && !est ? "Counting…" : "Estimate"}</button>
+        <button
+          onClick={start}
+          disabled={!canStart || working}
+          className="bg-indigo-600 hover:bg-indigo-500 disabled:opacity-50 text-white text-sm font-mono px-4 py-1.5 rounded-lg transition-colors"
+        >
+          Start run
+        </button>
+        {busy && <span className="text-xs text-amber-400 font-mono">a run is in progress</span>}
+      </div>
+
+      {est && (
+        <p className="text-sm font-mono text-gray-300">
+          {est.truncated ? (
+            <span className="text-red-400">More than {est.max_calls} calls — narrow the range.</span>
+          ) : (
+            <>
+              <span className="text-white">{est.calls}</span> calls ·{" "}
+              {est.calls_with_transcript} with transcripts · {est.audio_minutes} audio min ·{" "}
+              <span className="text-amber-400">~${est.est_cost_usd.toFixed(2)}</span> est.
+              {mode === "transcripts" && est.calls_with_transcript < est.calls / 2 && (
+                <span className="text-amber-400"> — most calls have no transcript; consider Re-transcribe audio.</span>
+              )}
+            </>
+          )}
+        </p>
+      )}
+
+      {error && <p className="text-red-400 text-sm font-mono">{error}</p>}
+    </div>
+  );
+}
+
+// ---------------------------------------------------------------------------
+
+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 <p className="text-sm text-gray-500 font-mono">No runs yet.</p>;
+
+  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 (
+    <div className="overflow-x-auto">
+      <table className="w-full text-sm font-mono">
+        <thead>
+          <tr className="border-b border-gray-800">
+            <th className={th}>Run</th>
+            <th className={th}>Range</th>
+            <th className={th}>Status</th>
+            <th className={th} title="Incidents created">Inc</th>
+            <th className={th} title="Share of incidents that are a single call">1-call</th>
+            <th className={th} title="Calls that never linked to an incident">Orphans</th>
+            <th className={th} title="Closed by radio traffic (units cleared, closure, reassignment) vs the idle timer">Real clears / timeouts</th>
+            <th className={th} title="Correlation decisions the LLM took part in">LLM</th>
+            <th className={th}>Cost</th>
+            <th className={th}></th>
+          </tr>
+        </thead>
+        <tbody>
+          {runs.map((r) => {
+            const m = r.metrics;
+            const running = r.run_id === activeId;
+            return (
+              <tr
+                key={r.run_id}
+                onClick={() => 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"}`}
+              >
+                <td className={td}>
+                  <div className="text-white">{r.label || r.run_id}</div>
+                  <div className="text-xs text-gray-500">{r.mode} · {r.git_sha?.slice(0, 7)} · {fmtTime(r.created_at)}</div>
+                </td>
+                <td className={`${td} text-xs text-gray-400`}>{fmtTime(r.date_from)} → {fmtTime(r.date_to)}</td>
+                <td className={td}>
+                  {running ? (
+                    <span className="text-amber-400">{r.progress.done}/{r.progress.total}</span>
+                  ) : (
+                    <span className={r.status === "done" ? "text-green-400" : "text-red-400"}>{r.status}</span>
+                  )}
+                  {r.progress.errors > 0 && <span className="text-red-400 text-xs"> · {r.progress.errors} err</span>}
+                </td>
+                <td className={td}>{m?.incidents ?? "—"}</td>
+                <td className={td}>{m?.single_call_pct != null ? `${m.single_call_pct}%` : "—"}</td>
+                <td className={td}>{m ? `${m.calls_orphaned}/${m.calls}` : "—"}</td>
+                <td className={td}>
+                  {m ? (
+                    <>
+                      <span className="text-green-400">{sum(m.resolved_via, SIGNAL_RESOLVES)}</span>
+                      {" / "}
+                      <span className="text-gray-400">{m.resolved_via.idle_timeout ?? 0}</span>
+                    </>
+                  ) : "—"}
+                </td>
+                <td className={td}>{m?.llm_decisions ?? "—"}</td>
+                <td className={td}>{m ? `$${m.est_cost_usd.toFixed(2)}` : `~$${r.estimate.est_cost_usd.toFixed(2)}`}</td>
+                <td className={td} onClick={(e) => e.stopPropagation()}>
+                  {running ? (
+                    <button onClick={() => cancel(r.run_id)} className="text-xs text-amber-400 hover:text-amber-300">cancel</button>
+                  ) : (
+                    <button onClick={() => remove(r.run_id)} className="text-xs text-gray-500 hover:text-red-400">delete</button>
+                  )}
+                </td>
+              </tr>
+            );
+          })}
+        </tbody>
+      </table>
+    </div>
+  );
+}
+
+// ---------------------------------------------------------------------------
+
+function CallLine({ call }: { call: ReplayCallRow }) {
+  const [audio, setAudio] = useState<string | null>(null);
+  async function play() {
+    try {
+      const c = await c2api.getCall(call.call_id);
+      setAudio(c.audio_url);
+    } catch { /* audio is a convenience */ }
+  }
+  return (
+    <div className="py-1.5 border-t border-gray-800/60 text-xs font-mono">
+      <div className="flex flex-wrap gap-2 text-gray-500">
+        <span>{fmtTime(call.started_at)}</span>
+        <span>{call.talkgroup_name}</span>
+        {call.corr_path.map((p, i) => <span key={i} className="text-indigo-400">{p}</span>)}
+        {call.units?.length ? <span>units {call.units.join(", ")}</span> : null}
+        {call.cleared_units?.length ? <span className="text-green-400">cleared {call.cleared_units.join(", ")}</span> : null}
+        {call.skip_reason && <span className="text-gray-600">{call.skip_reason}</span>}
+        {audio ? (
+          <audio src={audio} controls autoPlay className="h-6" />
+        ) : (
+          <button onClick={play} className="text-gray-400 hover:text-white">▶ audio</button>
+        )}
+      </div>
+      <div className="text-gray-200 mt-0.5">{call.transcript || <span className="text-gray-600">(no transcript)</span>}</div>
+    </div>
+  );
+}
+
+function IncidentCard({ inc }: { inc: ReplayIncident }) {
+  const [open, setOpen] = useState(false);
+  const signal = inc.resolved_via && SIGNAL_RESOLVES.includes(inc.resolved_via);
+  return (
+    <div className="bg-gray-900 border border-gray-800 rounded-lg px-3 py-2">
+      <button onClick={() => setOpen(!open)} className="w-full text-left flex flex-wrap items-center gap-x-3 gap-y-1 text-sm font-mono">
+        <span className="text-gray-500">{open ? "▾" : "▸"}</span>
+        <span className="text-white">{inc.title || inc.incident_id}</span>
+        <span className="text-gray-500 text-xs">{inc.calls.length} call{inc.calls.length !== 1 ? "s" : ""}</span>
+        <span className="text-gray-500 text-xs">{fmtTime(inc.started_at)} → {fmtTime(inc.resolved_at)}</span>
+        <span className={`text-xs ${signal ? "text-green-400" : "text-gray-500"}`}>{inc.resolved_via ?? inc.status}</span>
+        {inc.location_coords && <span className="text-xs text-indigo-400">📍 {inc.location}</span>}
+      </button>
+      {open && <div className="mt-2">{inc.calls.map((c) => <CallLine key={c.call_id} call={c} />)}</div>}
+    </div>
+  );
+}
+
+function RunDetail({ run }: { run: ReplayRun }) {
+  const [data, setData] = useState<ReplayIncidents | null>(null);
+  const [error, setError] = useState<string | null>(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 (
+    <div className="space-y-3">
+      {m && (
+        <div className="grid grid-cols-2 sm:grid-cols-4 gap-2 text-xs font-mono">
+          {[
+            ["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]) => (
+            <div key={k as string} className="bg-gray-900 border border-gray-800 rounded-lg p-2">
+              <div className="text-gray-500">{k}</div>
+              <div className="text-white text-base">{v}</div>
+            </div>
+          ))}
+        </div>
+      )}
+      {m && (
+        <p className="text-xs font-mono text-gray-500">
+          resolved: {Object.entries(m.resolved_via).map(([k, v]) => `${k} ${v}`).join(" · ")}
+          <br />
+          paths: {Object.entries(m.corr_path).map(([k, v]) => `${k} ${v}`).join(" · ")}
+        </p>
+      )}
+      {run.errors?.length > 0 && (
+        <details className="text-xs font-mono text-red-400">
+          <summary>{run.errors.length} error(s)</summary>
+          {run.errors.map((e, i) => <div key={i}>{e}</div>)}
+        </details>
+      )}
+      {run.status === "running" && <p className="text-sm text-gray-500 font-mono">Incidents appear when the run finishes.</p>}
+      {error && <p className="text-red-400 text-sm font-mono">{error}</p>}
+      {data && (
+        <>
+          <div className="flex gap-1 text-xs font-mono">
+            {(["all", "multi", "single"] as const).map((f) => (
+              <button
+                key={f}
+                onClick={() => setFilter(f)}
+                className={`px-3 py-1 rounded-md ${filter === f ? "bg-gray-800 text-white" : "text-gray-500 hover:text-gray-300"}`}
+              >
+                {f === "all" ? `All (${data.incidents.length})` : f === "multi" ? "Multi-call" : "Single-call"}
+              </button>
+            ))}
+            <button
+              onClick={() => setShowOrphans(!showOrphans)}
+              className={`px-3 py-1 rounded-md ${showOrphans ? "bg-gray-800 text-white" : "text-gray-500 hover:text-gray-300"}`}
+            >
+              Orphans ({data.orphans.length})
+            </button>
+          </div>
+          {showOrphans ? (
+            <div className="bg-gray-900 border border-gray-800 rounded-lg px-3 py-2">
+              {data.orphans.map((c) => <CallLine key={c.call_id} call={c} />)}
+            </div>
+          ) : (
+            <div className="space-y-1.5">{shown.map((i) => <IncidentCard key={i.incident_id} inc={i} />)}</div>
+          )}
+        </>
+      )}
+    </div>
+  );
+}
+
+// ---------------------------------------------------------------------------
+
+export function ReplayTab() {
+  const [runs, setRuns] = useState<ReplayRun[]>([]);
+  const [activeId, setActiveId] = useState<string | null>(null);
+  const [selected, setSelected] = useState<string | null>(null);
+  const [error, setError] = useState<string | null>(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 (
+    <div className="space-y-5">
+      <p className="text-xs text-gray-500 font-mono">
+        Re-run the pipeline over a past time range, in the calls&apos; 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.
+      </p>
+      <NewRunForm runs={runs} busy={!!activeId} onStarted={load} />
+      {error && <p className="text-red-400 text-sm font-mono">{error}</p>}
+      <RunsTable runs={runs} activeId={activeId} selected={selected} onSelect={setSelected} onChanged={load} />
+      {run && <RunDetail run={run} />}
+    </div>
+  );
+}

drb-frontend/lib/c2api.ts

@@ -233,6 +233,28 @@ export const c2api = {
   getCorrelationDebug: (limit: number, orphanHours: number) =>
     request<unknown>(`/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<import("@/lib/types").ReplayEstimate>(`/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<import("@/lib/types").ReplayRun>("/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<import("@/lib/types").ReplayRun>(`/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<import("@/lib/types").ReplayIncidents>(`/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" }),

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<string, number>;
+  corr_path: Record<string, number>;
+  corr_consensus: Record<string, number>;
+  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<ReplayEstimate, "truncated" | "max_calls">;
+  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[];
+}