From 140dfbfc749e319d7488424a5d7640219448c072 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 23 Aug 2026 12:33:52 -0400 Subject: [PATCH] Give the archive a real read, and the debug view a verdict MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three backend pieces the /calls page needs, plus the fix for a debug view that hid its data exactly when it was wanted. GET /calls/search — paged, filterable call archive. GET /calls returns every call in one unordered shot: fine for a node's handful of active calls, useless as an archive. Only the org scope and the started_at ordering go to Firestore, since that pair is the one composite index that exists; the rest filters in Python over a bounded window, the same shape admin.py's debug route uses. The cursor advances over the scanned window rather than the returned page, or a sparse filter would re-scan from the same place forever. Manual attribution. POST /incidents/{id}/calls/{id} only ever wrote the legacy scalar incident_id, never incident_ids -- which is what the correlator writes and what the frontend queries with array-contains. A manually attached call was therefore invisible on the incident page it had just been attached to. It now maintains both and marks the summary stale. DELETE is new: there was no way to undo an attachment at all, so a wrong link was permanent. The debug view no longer filters to AI-enabled systems by default. That filter emptied the view the moment the flags went off, which is precisely when a window gets reviewed -- on 2026-08-23 it fell from 100 incidents to 6 between switching correlation off and opening the tab. ai_systems_only=true restores it. It also returns a summary block now: corr_path / fit_signal / consensus / llm_action tallies, transcript coverage on both linked and orphaned calls, single-call and median-calls-per-incident for fragmentation, max span and anything past the server-26#22 caps for merging, and the count of incidents still carrying a fallback "— TGID" title. All of it was being recomputed by hand from the raw payload on every review. Co-Authored-By: Claude Opus 5 --- drb-c2-core/app/routers/admin.py | 91 ++++++++++++++++++++++++++-- drb-c2-core/app/routers/calls.py | 90 +++++++++++++++++++++++++++ drb-c2-core/app/routers/incidents.py | 61 ++++++++++++++++++- 3 files changed, 234 insertions(+), 8 deletions(-) diff --git a/drb-c2-core/app/routers/admin.py b/drb-c2-core/app/routers/admin.py index d9012c4..25519cd 100644 --- a/drb-c2-core/app/routers/admin.py +++ b/drb-c2-core/app/routers/admin.py @@ -4,6 +4,7 @@ from fastapi import APIRouter, Depends, Query from app.internal.auth import require_admin_token from app.internal.feature_flags import get_flags, set_flags from app.internal import firestore as fstore +from app.config import settings async def _get_ai_enabled_system_ids(global_flags: dict) -> set[str]: """Return system_ids where at least one AI function (STT or correlation) is effectively on.""" @@ -44,6 +45,7 @@ async def update_feature_flags(body: dict, _=Depends(require_admin_token)): async def debug_correlation( limit: int = Query(20, ge=1, le=100), orphan_hours: int = Query(48, ge=1, le=168), + ai_systems_only: bool = Query(False, description="Restrict to systems with STT or correlation currently enabled"), _=Depends(require_admin_token), ): """ @@ -102,9 +104,19 @@ async def debug_correlation( } # ── Determine which systems have AI active ──────────────────────────────── + # NOT a filter by default. Restricting to AI-enabled systems meant the view + # emptied itself the moment the flags went off — which is precisely when a + # window gets reviewed. On 2026-08-23 it dropped from 100 incidents to 6 + # between switching correlation off and opening the tab. Pass + # ai_systems_only=true to get the old behaviour. global_flags = await get_flags() ai_systems = await _get_ai_enabled_system_ids(global_flags) + def _in_scope(system_ids: list) -> bool: + if not ai_systems_only: + return True + return any(sid in ai_systems for sid in system_ids) + # ── Fetch recent incidents (AI-enabled systems only) ────────────────────── # Read a bounded, already-sorted window rather than the whole collection. # This route used to pull every incident ever created and sort in Python, @@ -123,10 +135,7 @@ async def debug_correlation( order_by=[("updated_at", "DESCENDING")], limit_to=window, ) - ai_incidents = [ - i for i in all_incidents - if any(sid in ai_systems for sid in (i.get("system_ids") or [])) - ] + ai_incidents = [i for i in all_incidents if _in_scope(i.get("system_ids") or [])] incidents = ai_incidents[:limit] incidents_window_exhausted = len(all_incidents) >= window and len(ai_incidents) < limit @@ -177,7 +186,7 @@ async def debug_correlation( if c.get("status") == "ended" and not c.get("incident_ids") and not c.get("incident_id") and not c.get("duplicate_of") # another node's copy — never meant to correlate - and c.get("system_id") in ai_systems + and _in_scope([c.get("system_id")]) ] orphans.sort(key=lambda c: c.get("started_at", ""), reverse=True) @@ -199,8 +208,80 @@ async def debug_correlation( if (o.get("corr_sweep_count") or 0) >= 3: orphans_by_tg[tg_key]["sweep_exhausted_count"] += 1 + # ── Summary ─────────────────────────────────────────────────────────────── + # Everything below was being recomputed by hand from the raw payload on + # every review — path counts, how much of the run the LLM tier actually saw, + # how many incidents ended up with the "Ems — TGID 9048" fallback name, and + # whether anything blew past the server-26#22 caps. Compute it once, here, + # where the data already is. + def _tally(values) -> dict: + out: dict[str, int] = {} + for v in values: + k = str(v) if v is not None else "none" + out[k] = out.get(k, 0) + 1 + return dict(sorted(out.items(), key=lambda kv: kv[1], reverse=True)) + + linked = [c for inc in incident_records for c in (inc.get("calls_detail") or [])] + call_counts = [len(inc.get("call_ids") or []) for inc in incident_records] + + def _span_minutes(inc: dict) -> float: + stamps = sorted( + s for s in ((c.get("started_at") or "") for c in (inc.get("calls_detail") or [])) if s + ) + if len(stamps) < 2: + return 0.0 + try: + first = datetime.fromisoformat(str(stamps[0]).replace("Z", "+00:00")) + last = datetime.fromisoformat(str(stamps[-1]).replace("Z", "+00:00")) + return round((last - first).total_seconds() / 60, 1) + except ValueError: + return 0.0 + + spans = [_span_minutes(inc) for inc in incident_records] + with_transcript = sum(1 for c in linked if (c.get("transcript") or "").strip()) + fallback_titles = sum( + 1 for inc in incident_records + if " — TGID " in (inc.get("title") or "") or (inc.get("title") or "").endswith("Unknown Talkgroup") + ) + over_cap = [ + {"incident_id": inc.get("incident_id"), "title": inc.get("title"), + "calls": len(inc.get("call_ids") or []), "span_minutes": _span_minutes(inc)} + for inc in incident_records + if len(inc.get("call_ids") or []) > settings.incident_max_calls + or _span_minutes(inc) > settings.incident_max_duration_minutes + ] + + summary = { + "ai_systems_only": ai_systems_only, + "ai_enabled_system_ids": sorted(ai_systems), + "linked_call_count": len(linked), + "corr_path": _tally(c.get("corr_path") for c in linked), + "corr_fit_signal": _tally(c.get("corr_fit_signal") for c in linked), + "corr_consensus": _tally(c.get("corr_consensus") for c in linked), + "corr_llm_action": _tally(c.get("corr_llm_action") for c in linked), + # STT coverage: correlation quality is capped by this, so it belongs in + # the same view rather than a separate investigation. + "linked_calls_with_transcript": with_transcript, + "linked_calls_without_transcript": len(linked) - with_transcript, + "orphans_with_transcript": sum(1 for o in orphans if (o.get("transcript") or "").strip()), + # Fragmentation vs merging, the two failure directions. + "single_call_incidents": sum(1 for n in call_counts if n == 1), + "median_calls_per_incident": sorted(call_counts)[len(call_counts) // 2] if call_counts else 0, + "max_calls_in_one_incident": max(call_counts) if call_counts else 0, + "max_span_minutes": max(spans) if spans else 0.0, + "incidents_over_cap": over_cap, + "caps": { + "incident_max_calls": settings.incident_max_calls, + "incident_max_duration_minutes": settings.incident_max_duration_minutes, + }, + # Titling health — server-26#34. + "fallback_titled_incidents": fallback_titles, + "titled_incidents": len(incident_records) - fallback_titles, + } + return { "generated_at": datetime.now(timezone.utc).isoformat(), + "summary": summary, # Both reads are capped, so say plainly when a cap was hit — otherwise a # truncated window is indistinguishable from a quiet night. "incidents_window_exhausted": incidents_window_exhausted, diff --git a/drb-c2-core/app/routers/calls.py b/drb-c2-core/app/routers/calls.py index 3f6ae2a..1a70e80 100644 --- a/drb-c2-core/app/routers/calls.py +++ b/drb-c2-core/app/routers/calls.py @@ -40,6 +40,96 @@ async def list_calls( return [with_playback_url(c) for c in calls] +@router.get("/search") +async def search_calls( + limit: int = Query(50, ge=1, le=200), + cursor: Optional[str] = Query(None, description="started_at of the last row of the previous page"), + system_id: Optional[str] = Query(None), + node_id: Optional[str] = Query(None), + talkgroup_id: Optional[int] = Query(None), + link: str = Query("any", pattern="^(any|orphan|linked)$"), + transcript: str = Query("any", pattern="^(any|yes|no)$"), + q: Optional[str] = Query(None, description="case-insensitive substring of the transcript"), + decoded: dict = Depends(require_admin_token), +): + """ + Paged, filterable call archive — the backend for the /calls page. + + `GET /calls` returns every call in one unordered shot, which is fine for a + node's handful of active calls and useless as an archive: no order, no + paging, no way to find the orphans. This route is the archive read. + + Only the org scope and the started_at ordering go to Firestore, because + that pair is the one composite index that exists (infra/firestore/ + firestore.indexes.json). Every other filter runs in Python over a bounded + window, the same shape admin.py's correlation debug route uses — adding a + composite index per filter combination would be a worse trade than reading + 10x the page and discarding most of it. + + `window_exhausted` says the scan hit its cap before filling the page, so an + empty result means "not in this window", not "none exist". + """ + org_id = await resolve_caller_org_id(decoded) + if org_id is None: + # resolve_caller_org_id lets platform admins see every org (server-26#4). + # A new browse surface shouldn't widen that, so fall back to the + # caller's own org claim when they have one. + org_id = decoded.get("org_id") + if not org_id: + raise HTTPException(403, "No organization scope for this caller.") + + window = max(limit * 10, 200) + rows = await fstore.collection_where( + "calls", + [("org_id", "==", org_id)], + order_by=[("started_at", "DESCENDING")], + limit_to=window, + start_after={"started_at": cursor} if cursor else None, + ) + + needle = (q or "").strip().lower() + + def _keep(c: dict) -> bool: + if system_id and c.get("system_id") != system_id: + return False + if node_id and c.get("node_id") != node_id: + return False + if talkgroup_id is not None and c.get("talkgroup_id") != talkgroup_id: + return False + linked = bool(c.get("incident_ids") or c.get("incident_id")) + if link == "orphan" and linked: + return False + if link == "linked" and not linked: + return False + text = c.get("transcript_corrected") or c.get("transcript") or "" + if transcript == "yes" and not text: + return False + if transcript == "no" and text: + return False + if needle and needle not in text.lower(): + return False + return True + + matches = [c for c in rows if _keep(c)] + page = matches[:limit] + + # Cursor advances over the SCANNED window, not the filtered page — otherwise + # a page whose last match sits early in the window would re-scan everything + # after it on the next request and loop forever on a sparse filter. + next_cursor = None + if len(rows) == window: + last_scanned = rows[-1].get("started_at") + next_cursor = last_scanned.isoformat() if hasattr(last_scanned, "isoformat") else last_scanned + + return { + "calls": [with_playback_url(c) for c in page], + "next_cursor": next_cursor, + "scanned": len(rows), + "matched": len(matches), + "window_exhausted": len(rows) == window, + } + + @router.get("/{call_id}") async def get_call(call_id: str, decoded: dict = Depends(require_service_or_firebase_token)): call = await fstore.doc_get("calls", call_id) diff --git a/drb-c2-core/app/routers/incidents.py b/drb-c2-core/app/routers/incidents.py index 75959b9..d2b1eba 100644 --- a/drb-c2-core/app/routers/incidents.py +++ b/drb-c2-core/app/routers/incidents.py @@ -112,15 +112,70 @@ async def summarize_incident( @router.post("/{incident_id}/calls/{call_id}") async def link_call_to_incident(incident_id: str, call_id: str, _: dict = Depends(require_admin_token)): + """Manually attach a call to an incident (the /calls page's attribution action).""" doc = await fstore.doc_get("incidents", incident_id) if not doc: raise HTTPException(404, f"Incident '{incident_id}' not found.") - call_ids = doc.get("call_ids", []) + call = await fstore.doc_get("calls", call_id) + if not call: + raise HTTPException(404, f"Call '{call_id}' not found.") + + call_ids = list(doc.get("call_ids") or []) if call_id not in call_ids: call_ids.append(call_id) await fstore.doc_update("incidents", incident_id, { "call_ids": call_ids, "updated_at": datetime.now(timezone.utc).isoformat(), + # A manually attached call changes what the incident is about. + "summary_stale": True, }) - await fstore.doc_update("calls", call_id, {"incident_id": incident_id}) - return {"ok": True} + + # incident_ids is the canonical link — it is what the correlator writes and + # what the frontend queries with array-contains. This route only ever set + # the legacy scalar incident_id, so a manually attached call stayed + # invisible on the incident's own page. + incident_ids = list(call.get("incident_ids") or ([call["incident_id"]] if call.get("incident_id") else [])) + if incident_id not in incident_ids: + incident_ids.append(incident_id) + await fstore.doc_update("calls", call_id, { + "incident_ids": incident_ids, + "incident_id": incident_id, + "corr_path": "manual", + }) + return {"ok": True, "incident_ids": incident_ids} + + +@router.delete("/{incident_id}/calls/{call_id}") +async def unlink_call_from_incident(incident_id: str, call_id: str, _: dict = Depends(require_admin_token)): + """ + Detach a call from an incident — the other half of manual attribution. + + An incident left with no calls is resolved rather than deleted, matching + what calls.py's transcript correction does when it empties one. + """ + doc = await fstore.doc_get("incidents", incident_id) + if not doc: + raise HTTPException(404, f"Incident '{incident_id}' not found.") + + remaining = [c for c in (doc.get("call_ids") or []) if c != call_id] + updates: dict = { + "call_ids": remaining, + "updated_at": datetime.now(timezone.utc).isoformat(), + "summary_stale": True, + } + if not remaining: + updates["status"] = "resolved" + updates["resolved_at"] = datetime.now(timezone.utc).isoformat() + await fstore.doc_update("incidents", incident_id, updates) + + call = await fstore.doc_get("calls", call_id) + if call: + incident_ids = [ + i for i in (call.get("incident_ids") or ([call["incident_id"]] if call.get("incident_id") else [])) + if i != incident_id + ] + await fstore.doc_update("calls", call_id, { + "incident_ids": incident_ids, + "incident_id": incident_ids[0] if incident_ids else None, + }) + return {"ok": True, "incident_emptied": not remaining}