Logan CusanoandClaude Opus 5
140dfbfc74
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 <noreply@anthropic.com>
291 lines
11 KiB
Python
291 lines
11 KiB
Python
from datetime import datetime, timezone, timedelta
|
|
from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, Depends
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
from app.internal import firestore as fstore
|
|
from app.internal.auth import (
|
|
require_admin_token,
|
|
require_service_or_firebase_token,
|
|
resolve_caller_org_id,
|
|
reprocess_limiter,
|
|
)
|
|
from app.internal.storage import gcs_uri_for_call, with_playback_url
|
|
|
|
|
|
class TranscriptUpdate(BaseModel):
|
|
transcript: str
|
|
|
|
router = APIRouter(prefix="/calls", tags=["calls"])
|
|
|
|
|
|
@router.get("")
|
|
async def list_calls(
|
|
node_id: Optional[str] = Query(None),
|
|
status: Optional[str] = Query(None),
|
|
system_id: Optional[str] = Query(None),
|
|
decoded: dict = Depends(require_service_or_firebase_token),
|
|
):
|
|
filters = {}
|
|
if node_id:
|
|
filters["node_id"] = node_id
|
|
if status:
|
|
filters["status"] = status
|
|
if system_id:
|
|
filters["system_id"] = system_id
|
|
org_id = await resolve_caller_org_id(decoded)
|
|
if org_id is not None: # service key / platform admin stay unrestricted
|
|
filters["org_id"] = org_id
|
|
calls = await fstore.collection_list("calls", **filters)
|
|
# audio_url is not stored — it's a short-lived signed link minted per read.
|
|
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)
|
|
if not call:
|
|
raise HTTPException(404, f"Call '{call_id}' not found.")
|
|
org_id = await resolve_caller_org_id(decoded)
|
|
if org_id is not None and call.get("org_id") != org_id:
|
|
raise HTTPException(404, f"Call '{call_id}' not found.")
|
|
return with_playback_url(call)
|
|
|
|
|
|
@router.post("/{call_id}/reprocess")
|
|
async def reprocess_call(
|
|
call_id: str,
|
|
background_tasks: BackgroundTasks,
|
|
_: dict = Depends(require_admin_token),
|
|
):
|
|
"""
|
|
Re-run the full intelligence pipeline (transcription -> extraction ->
|
|
correlation) for a call. Admin-only (SAAS_PLAN.md B2c) — this was
|
|
previously gated only by "any valid Firebase token", which meant any
|
|
signed-in viewer could loop it and burn the owner's OpenAI/Gemini
|
|
credits (DEFERRED.md, calls.py:42). The rate limiter below is a second
|
|
guard against the same thing happening from a compromised/careless
|
|
admin session, not the primary fix.
|
|
"""
|
|
call = await fstore.doc_get("calls", call_id)
|
|
if not call:
|
|
raise HTTPException(404, f"Call '{call_id}' not found.")
|
|
reprocess_limiter.check(call_id)
|
|
|
|
from app.routers.upload import _run_intelligence_pipeline
|
|
|
|
gcs_uri = gcs_uri_for_call(call)
|
|
|
|
background_tasks.add_task(
|
|
_run_intelligence_pipeline,
|
|
call_id=call_id,
|
|
node_id=call.get("node_id"),
|
|
system_id=call.get("system_id"),
|
|
talkgroup_id=call.get("talkgroup_id"),
|
|
talkgroup_name=call.get("talkgroup_name"),
|
|
gcs_uri=gcs_uri,
|
|
)
|
|
return {"ok": True, "call_id": call_id}
|
|
|
|
|
|
@router.post("/close-stale")
|
|
async def close_stale_calls(
|
|
older_than_minutes: int = Query(30, ge=1, le=1440, description="Close active calls started more than this many minutes ago."),
|
|
dry_run: bool = Query(False, description="If true, return what would be closed without writing."),
|
|
_: dict = Depends(require_admin_token),
|
|
):
|
|
"""
|
|
Find and close calls stuck in 'active' status — e.g. because a node rebooted
|
|
before sending an end-call event. Returns the list of affected call IDs.
|
|
"""
|
|
cutoff = datetime.now(timezone.utc) - timedelta(minutes=older_than_minutes)
|
|
active_calls = await fstore.collection_list("calls", status="active")
|
|
|
|
stale = []
|
|
for call in active_calls:
|
|
started_raw = call.get("started_at")
|
|
if not started_raw:
|
|
continue
|
|
if isinstance(started_raw, datetime):
|
|
started = started_raw if started_raw.tzinfo else started_raw.replace(tzinfo=timezone.utc)
|
|
else:
|
|
try:
|
|
started = datetime.fromisoformat(str(started_raw).replace("Z", "+00:00"))
|
|
except Exception:
|
|
continue
|
|
if started < cutoff:
|
|
stale.append(call)
|
|
|
|
if not dry_run:
|
|
now_iso = datetime.now(timezone.utc).isoformat()
|
|
for call in stale:
|
|
await fstore.doc_set("calls", call["call_id"], {
|
|
"status": "ended",
|
|
"ended_at": now_iso,
|
|
})
|
|
|
|
return {
|
|
"dry_run": dry_run,
|
|
"older_than_minutes": older_than_minutes,
|
|
"count": len(stale),
|
|
"call_ids": [c["call_id"] for c in stale],
|
|
}
|
|
|
|
|
|
@router.patch("/{call_id}/transcript")
|
|
async def patch_transcript(
|
|
call_id: str,
|
|
body: TranscriptUpdate,
|
|
background_tasks: BackgroundTasks,
|
|
_: dict = Depends(require_admin_token),
|
|
):
|
|
"""Overwrite a call's transcript and re-run intelligence extraction."""
|
|
call = await fstore.doc_get("calls", call_id)
|
|
if not call:
|
|
raise HTTPException(404, f"Call '{call_id}' not found.")
|
|
|
|
# Save user correction as transcript_corrected; leave original transcript intact.
|
|
# Clear stale intelligence fields so re-extraction runs fresh.
|
|
await fstore.doc_set("calls", call_id, {
|
|
"transcript_corrected": body.transcript,
|
|
"tags": [],
|
|
"severity": "unknown",
|
|
"location": None,
|
|
"units": [],
|
|
"vehicles": [],
|
|
"embedding": None,
|
|
})
|
|
|
|
# Unlink from ALL current incidents so re-correlation starts clean.
|
|
# Handles both old single incident_id and new incident_ids list.
|
|
old_ids: list[str] = call.get("incident_ids") or (
|
|
[call["incident_id"]] if call.get("incident_id") else []
|
|
)
|
|
for old_incident_id in old_ids:
|
|
old_incident = await fstore.doc_get("incidents", old_incident_id)
|
|
if old_incident:
|
|
remaining = [c for c in (old_incident.get("call_ids") or []) if c != call_id]
|
|
if remaining:
|
|
await fstore.doc_set("incidents", old_incident_id, {
|
|
"call_ids": remaining,
|
|
"summary_stale": True,
|
|
})
|
|
else:
|
|
await fstore.doc_set("incidents", old_incident_id, {
|
|
"call_ids": [],
|
|
"status": "resolved",
|
|
"resolved_at": datetime.now(timezone.utc).isoformat(),
|
|
"summary_stale": True,
|
|
})
|
|
await fstore.doc_set("calls", call_id, {"incident_ids": [], "incident_id": None})
|
|
|
|
# Learn from the correction: diff original → corrected and add new tokens to vocabulary
|
|
system_id = call.get("system_id")
|
|
original_text = call.get("transcript_corrected") or call.get("transcript") or ""
|
|
if system_id and original_text:
|
|
from app.internal.vocabulary_learner import learn_from_correction
|
|
await learn_from_correction(system_id, original_text, body.transcript)
|
|
|
|
from app.routers.upload import _run_extraction_pipeline
|
|
background_tasks.add_task(
|
|
_run_extraction_pipeline,
|
|
call_id=call_id,
|
|
node_id=call.get("node_id"),
|
|
system_id=call.get("system_id"),
|
|
talkgroup_id=call.get("talkgroup_id"),
|
|
talkgroup_name=call.get("talkgroup_name"),
|
|
transcript=body.transcript,
|
|
segments=call.get("segments"),
|
|
preserve_transcript_correction=True,
|
|
)
|
|
return {"ok": True, "call_id": call_id}
|