Logan Cusano
d18e4f0743
config/ai_features was not the switch it was documented to be. Three paths spent money with it off, and one path read it wrong, so per-system opt-outs did not opt anything out. - Correlation in the ingest pipeline tested the raw global flag instead of the per-system resolution. With a system opted out, extraction was skipped but the no-scenes fallback still correlated the call with empty tags, taking the thin/recency path and attaching it to whatever incident was most recent on that system. The opt-out did not disable correlation, it disabled good correlation and left the worst kind running. (#75) - Transcript correction ran on every transcribed call gated only by an env var, spending Gemini tokens and a Places lookup per proposed location. An "STT-only" window was never STT-only and its cost could not be attributed. Now behind transcript_correction_enabled. (#76) - _run_extraction_pipeline and the vocabulary learner, both reachable from PATCH /calls/{id}/transcript, checked no flags at all. (#76, #81) The flag resolver now lives in feature_flags.resolve_flags() rather than as a local helper in upload.py. Three copies of that logic is how #75 happened. PATCH /calls/{id}/transcript now refuses with 409 when correlation is off. That route wipes tags, severity, location, units, embedding and unlinks the call from every incident before queueing re-extraction. Gating extraction alone would have made it destructive-only in the standing flags-off configuration: the call left blank and orphaned forever, with the route still answering 200. The wipe and the rebuild are one transaction in intent, so it refuses before the first write. Also: the summarizer's stale-incident sweep is no longer behind summaries_enabled. It is pure Firestore with no model call in it, and gating it meant nothing auto-resolved while AI was off - so every incident stayed active forever and the candidate set every correlation reads kept growing. transcript_correction_enabled is documented as NOT a pure cost lever. The corrector is also the noise gate that sets not_speech; with it off, recogniser noise reaches extraction as a real transcript, comes back thin, and auto-attaches. Never open an evaluation window with correction off and correlation on. 14 tests added covering flag precedence, both pipeline paths, the 409, the correction gate and the summarizer no-op. Suite: 264 passed. Refs #75, #76, #81, #45.
307 lines
12 KiB
Python
307 lines
12 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."""
|
|
from app.internal.feature_flags import resolve_flags
|
|
|
|
call = await fstore.doc_get("calls", call_id)
|
|
if not call:
|
|
raise HTTPException(404, f"Call '{call_id}' not found.")
|
|
|
|
# This route is destructive before it is constructive: it wipes the call's
|
|
# tags, severity, location, units and embedding and unlinks it from every
|
|
# incident, on the promise that re-extraction will rebuild all of it. With
|
|
# correlation off that promise cannot be kept, and the call would be left
|
|
# permanently blank and orphaned while the route still answered 200.
|
|
# Refuse before the first write rather than half-run (server-26#76).
|
|
_, flag = await resolve_flags(call.get("system_id"))
|
|
if not flag("correlation_enabled"):
|
|
raise HTTPException(
|
|
409,
|
|
"Correlation is disabled, so the re-extraction this correction depends on "
|
|
"cannot run. The transcript was not changed. Enable correlation and retry.",
|
|
)
|
|
|
|
# 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 and flag("vocabulary_learning_enabled"):
|
|
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}
|