Logan CusanoandClaude Opus 5
90a0412066
/admin/debug/correlation read every incident ever created, sorted them in Python and kept 20, and separately pulled every call in the orphan window with no cap. That worked while the collections were small. They are not small now: Firestore kills an unbounded scan with a 503 and the request never returns, so the debug view simply spins -- which is also what made the org backfill script fail earlier tonight, same cause, different caller. Incidents now come back pre-sorted from Firestore with a limit, and the orphan scan is capped at 3000 documents. Both queries order on the single field they already filter or sort by (updated_at, ended_at), so neither needs a composite index -- worth preserving, since the index file from the tenancy work has not been deployed. Capping introduces a way to be wrong quietly: a truncated window looks exactly like a quiet night. The payload now carries incidents_window_exhausted and orphan_scan_truncated so a short result announces itself instead of being read as a correlation improvement. The AI-system filter still runs in Python, so the incident window is 10x the requested limit rather than the limit itself. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
212 lines
9.8 KiB
Python
212 lines
9.8 KiB
Python
import asyncio
|
|
from datetime import datetime, timezone, timedelta
|
|
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
|
|
|
|
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."""
|
|
global_stt = global_flags.get("stt_enabled", True)
|
|
global_corr = global_flags.get("correlation_enabled", True)
|
|
all_systems = await fstore.collection_list("systems")
|
|
enabled: set[str] = set()
|
|
for system in all_systems:
|
|
sid = system.get("system_id")
|
|
if not sid:
|
|
continue
|
|
ai_flags = system.get("ai_flags") or {}
|
|
if ai_flags.get("stt_enabled", global_stt) or ai_flags.get("correlation_enabled", global_corr):
|
|
enabled.add(sid)
|
|
return enabled
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"])
|
|
|
|
|
|
@router.get("/features")
|
|
async def get_feature_flags(_=Depends(require_admin_token)):
|
|
"""
|
|
Return the current AI feature flag state. Admin-only (SAAS_PLAN.md B2c) —
|
|
was previously any authenticated user via require_firebase_token, which
|
|
handed platform-wide AI configuration state to every signed-in viewer
|
|
regardless of org.
|
|
"""
|
|
return await get_flags()
|
|
|
|
|
|
@router.put("/features")
|
|
async def update_feature_flags(body: dict, _=Depends(require_admin_token)):
|
|
"""Update one or more AI feature flags. Admin only."""
|
|
return await set_flags(body)
|
|
|
|
|
|
@router.get("/debug/correlation")
|
|
async def debug_correlation(
|
|
limit: int = Query(20, ge=1, le=100),
|
|
orphan_hours: int = Query(48, ge=1, le=168),
|
|
_=Depends(require_admin_token),
|
|
):
|
|
"""
|
|
Return the last N incidents with full correlation debug detail, plus recent orphaned calls.
|
|
|
|
Each incident includes a calls_detail array with per-call corr_* fields so you can see
|
|
exactly which correlation path fired (or didn't) for every call in the incident.
|
|
|
|
Embeddings are stripped — they're large float arrays and unreadable.
|
|
|
|
Query params:
|
|
limit — number of incidents to return, sorted by updated_at desc (default 20, max 100)
|
|
orphan_hours — how far back to scan for orphaned calls (default 48h, max 168h / 1 week)
|
|
"""
|
|
def _strip(doc: dict) -> dict:
|
|
return {k: v for k, v in doc.items() if k != "embedding"}
|
|
|
|
def _call_summary(call: dict) -> dict:
|
|
return {
|
|
"call_id": call.get("call_id"),
|
|
"started_at": call.get("started_at"),
|
|
"ended_at": call.get("ended_at"),
|
|
"duration_s": call.get("duration_s"),
|
|
"talkgroup_id": call.get("talkgroup_id"),
|
|
"talkgroup_name": call.get("talkgroup_name"),
|
|
"system_id": call.get("system_id"),
|
|
"node_id": call.get("node_id"),
|
|
"incident_type": call.get("incident_type"),
|
|
"tags": call.get("tags"),
|
|
"location": call.get("location"),
|
|
"location_coords": call.get("location_coords"),
|
|
"units": call.get("units"),
|
|
"vehicles": call.get("vehicles"),
|
|
"cleared_units": call.get("cleared_units"),
|
|
"severity": call.get("severity"),
|
|
"transcript": call.get("transcript_corrected") or call.get("transcript"),
|
|
# Correlation decision fields written back by incident_correlator
|
|
"corr_path": call.get("corr_path"),
|
|
"corr_incident_idle_min": call.get("corr_incident_idle_min"),
|
|
"corr_distance_km": call.get("corr_distance_km"),
|
|
"corr_score": call.get("corr_score"),
|
|
"corr_candidates": call.get("corr_candidates"),
|
|
"corr_shared_units": call.get("corr_shared_units"),
|
|
"corr_fit_signal": call.get("corr_fit_signal"),
|
|
"corr_matched_units": call.get("corr_matched_units"),
|
|
"corr_sweep_count": call.get("corr_sweep_count"),
|
|
"skip_reason": call.get("skip_reason"),
|
|
}
|
|
|
|
# ── Determine which systems have AI active ────────────────────────────────
|
|
global_flags = await get_flags()
|
|
ai_systems = await _get_ai_enabled_system_ids(global_flags)
|
|
|
|
# ── 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,
|
|
# which stopped returning at all once the collection grew — Firestore kills
|
|
# an unbounded scan with a 503 and the request just hangs. Ordering on the
|
|
# single field updated_at needs no composite index.
|
|
#
|
|
# The AI-system filter runs in Python (it's a membership test against a set
|
|
# the flags decide), so the window has to be wider than `limit` or filtering
|
|
# could empty it. 10x with a floor of 200 covers a debug view; if a fetch
|
|
# still comes back short, incidents_window_exhausted says so in the payload
|
|
# rather than quietly looking like "no incidents".
|
|
window = max(limit * 10, 200)
|
|
all_incidents = await fstore.collection_where(
|
|
"incidents", [],
|
|
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 []))
|
|
]
|
|
incidents = ai_incidents[:limit]
|
|
incidents_window_exhausted = len(all_incidents) >= window and len(ai_incidents) < limit
|
|
|
|
# ── Fetch all linked call docs in parallel ────────────────────────────────
|
|
all_call_ids: list[str] = []
|
|
for inc in incidents:
|
|
all_call_ids.extend(inc.get("call_ids") or [])
|
|
unique_call_ids = list(dict.fromkeys(all_call_ids)) # dedupe, preserve order
|
|
|
|
call_docs = await asyncio.gather(*(fstore.doc_get("calls", cid) for cid in unique_call_ids))
|
|
call_map: dict[str, dict] = {doc["call_id"]: doc for doc in call_docs if doc}
|
|
|
|
# ── Build incident debug records ──────────────────────────────────────────
|
|
incident_records = []
|
|
for inc in incidents:
|
|
rec = _strip(inc)
|
|
rec["calls_detail"] = [
|
|
_call_summary(call_map[cid])
|
|
for cid in (inc.get("call_ids") or [])
|
|
if cid in call_map
|
|
]
|
|
incident_records.append(rec)
|
|
|
|
# ── Recent orphaned calls (AI-enabled systems only) ───────────────────────
|
|
# Use a single-field range query to avoid requiring a composite Firestore index;
|
|
# filter status and system in Python.
|
|
cutoff = datetime.now(timezone.utc) - timedelta(hours=orphan_hours)
|
|
# Bounded for the same reason as the incident read above. The range and the
|
|
# sort are both on ended_at, which is what keeps this a single-field query
|
|
# needing no composite index.
|
|
_ORPHAN_SCAN_CAP = 3000
|
|
recent_calls = await fstore.collection_where(
|
|
"calls",
|
|
[("ended_at", ">=", cutoff)],
|
|
order_by=[("ended_at", "DESCENDING")],
|
|
limit_to=_ORPHAN_SCAN_CAP,
|
|
)
|
|
orphan_scan_truncated = len(recent_calls) >= _ORPHAN_SCAN_CAP
|
|
orphans = [
|
|
_call_summary(c) for c in recent_calls
|
|
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
|
|
]
|
|
orphans.sort(key=lambda c: c.get("started_at", ""), reverse=True)
|
|
|
|
# Summarise orphans by talkgroup so the volume and source are immediately visible.
|
|
orphans_by_tg: dict[str, dict] = {}
|
|
for o in orphans:
|
|
tg_key = str(o.get("talkgroup_id") or "unknown")
|
|
if tg_key not in orphans_by_tg:
|
|
orphans_by_tg[tg_key] = {
|
|
"talkgroup_id": o.get("talkgroup_id"),
|
|
"talkgroup_name": o.get("talkgroup_name") or "unknown",
|
|
"count": 0,
|
|
"no_type_count": 0,
|
|
"sweep_exhausted_count": 0,
|
|
}
|
|
orphans_by_tg[tg_key]["count"] += 1
|
|
if not o.get("incident_type") and not o.get("tags"):
|
|
orphans_by_tg[tg_key]["no_type_count"] += 1
|
|
if (o.get("corr_sweep_count") or 0) >= 3:
|
|
orphans_by_tg[tg_key]["sweep_exhausted_count"] += 1
|
|
|
|
return {
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
# 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,
|
|
"orphan_scan_truncated": orphan_scan_truncated,
|
|
"orphan_scan_cap": _ORPHAN_SCAN_CAP,
|
|
"incident_count": len(incident_records),
|
|
"orphaned_call_count": len(orphans),
|
|
"orphans_by_talkgroup": sorted(orphans_by_tg.values(), key=lambda x: x["count"], reverse=True),
|
|
"incidents": incident_records,
|
|
"orphaned_calls": orphans[:250],
|
|
}
|
|
|
|
|
|
@router.get("/audit")
|
|
async def get_audit_log(
|
|
limit: int = Query(50, ge=1, le=200),
|
|
offset: int = Query(0, ge=0),
|
|
_=Depends(require_admin_token),
|
|
):
|
|
"""Return paginated audit log entries, most recent first."""
|
|
entries = await fstore.collection_list("audit_log")
|
|
entries.sort(key=lambda e: e.get("timestamp", ""), reverse=True)
|
|
return entries[offset: offset + limit]
|