Logan CusanoandClaude Opus 5.5
ec91a9175f
First replay (290 calls, 09-22 10:00-12:00 ET) produced 0 incidents and no errors: every gpt-4o-mini extraction failed and _sync_extract swallowed it as "no scenes". Same shape as #169 — and the live extraction tier in /health/ai had no reporter at all, so this has been invisible in production too. - intelligence: API failures propagate out of _sync_extract; extract_scenes reports them to ai_health ("extraction" tier, billing/dead-model classified) and still returns [] so the pipeline degrades as before. - ai_health: inside a replay sandbox, failures go to the run's own sink instead of being dropped. - replay: aborts after 5 permanent failures on a tier, naming the cause; run metrics carry ai_failures; UI shows them. - replay estimate: audio minutes from started_at/ended_at (no duration field exists on call docs). - ReplayTab exposes the loaded run on window.__drbReplay for in-page analysis. c2-core: 458 pass. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
413 lines
17 KiB
Python
413 lines
17 KiB
Python
"""
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_aborts_when_an_ai_account_is_dead(store):
|
|
"""An unfunded OpenAI account made the first smoke run a sandbox of 290
|
|
orphans that looked like a result. A permanently failing tier now stops
|
|
the run and names the cause."""
|
|
store.data["calls"] = {
|
|
f"call-{i}": _live_call(i, i, "Car 12 responding to an MVA on Main Street") for i in range(1, 30)
|
|
}
|
|
|
|
def broke(*a, **kw):
|
|
raise RuntimeError("Error code: 429 - You exceeded your current quota (insufficient_quota)")
|
|
|
|
with patch("app.internal.intelligence._sync_extract", broke), \
|
|
patch("app.internal.intelligence.classify_chatter", return_value=(False, None)):
|
|
run = 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
|
|
|
|
run = store.data["replay_runs"][run["run_id"]]
|
|
assert run["status"] == "failed"
|
|
assert any("out of credit" in e for e in run["errors"])
|
|
assert run["progress"]["done"] < 29
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_live_extraction_failure_reports_to_ai_health():
|
|
from app.internal import ai_health, intelligence
|
|
|
|
def broke(*a, **kw):
|
|
raise RuntimeError("insufficient_quota")
|
|
|
|
with patch.object(intelligence, "_sync_extract", broke), \
|
|
patch.object(ai_health, "report_degraded") as degraded, \
|
|
patch.object(fstore, "doc_set"), patch.object(fstore, "doc_get_cached", return_value=None):
|
|
scenes = await intelligence.extract_scenes("c1", "Car 12 responding to an MVA on Main Street")
|
|
assert scenes == []
|
|
assert degraded.call_args.args[0] == "extraction"
|
|
assert degraded.call_args.kwargs["permanent"] is True
|