Logan Cusano
c7be6416f2
/admin/debug/correlation stripped corr_consensus and the corr_llm_* fields
that upload.py's consensus correlator writes onto the call doc, making it
the one tool built to answer "is the LLM correlation tier alive" unable to
answer it (2026-08-19 dump had to infer LLM state from commit dates instead
of reading it off the data). admin.py's _call_summary() now includes
corr_consensus, corr_llm_reasoning, corr_llm_action, corr_rules_action.
The unit-continuity correlation path never wrote corr_matched_units, unlike
fast/single and fast/disambig, so the debug view showed null for a match
that was in fact unit-driven by construction. Now populated unconditionally
on that path (server-26#16).
Also traced the negative corr_incident_idle_min (-4.1 observed) to its root
cause: the re-correlation sweep anchors `now` to the linking call's own
started_at, and that back-dated value was being written straight into the
incident's updated_at, letting it land before the incident's own
started_at. Added _floor_at_started_at() so updated_at can never precede
started_at. (commit 33a247d already fixed the recency *gates* misreading
that negative value; this fixes the write that produced it.) Verified the
skip_reason filter in recorrelation_sweep.py:63 is already correct, no
change needed there.
Added tests for the debug endpoint's LLM field passthrough, the
unit-continuity corr_matched_units fix, and the updated_at floor — each
confirmed to fail when its fix is reverted. 148 passed, 0 failed.
Closes logan/server-26#24
Closes logan/server-26#16
227 lines
11 KiB
Python
227 lines
11 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"),
|
|
# LLM consensus tier fields — written by upload.py's
|
|
# _correlate_with_consensus / llm_correlator.py, but previously
|
|
# dropped here, making it impossible to tell from this endpoint
|
|
# whether the LLM correlation tier is actually running (server-26#24).
|
|
"corr_consensus": call.get("corr_consensus"),
|
|
"corr_llm_reasoning": call.get("corr_llm_reasoning"),
|
|
"corr_llm_action": call.get("corr_llm_action"),
|
|
"corr_rules_action": call.get("corr_rules_action"),
|
|
}
|
|
|
|
# ── 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))
|
|
# Key off the id we asked for, not doc["call_id"]. At least one stored call
|
|
# has no call_id field -- the document id is authoritative and always
|
|
# present, while the field is written by the upload path and evidently was
|
|
# not always there. Indexing the field raised KeyError and took the whole
|
|
# debug view down with a 500 over a single malformed document.
|
|
call_map: dict[str, dict] = {
|
|
cid: doc for cid, doc in zip(unique_call_ids, 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]
|