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.
108 lines
4.0 KiB
Python
108 lines
4.0 KiB
Python
"""
|
|
Global AI feature flags stored in Firestore at config/ai_features.
|
|
|
|
Defaults to all-on when the document does not exist yet. Uses a short
|
|
in-memory TTL cache so flag reads don't add a Firestore round-trip to every
|
|
call upload.
|
|
"""
|
|
import time
|
|
from typing import Any
|
|
from app.internal.logger import logger
|
|
from app.internal import firestore as fstore
|
|
|
|
_COLLECTION = "config"
|
|
_DOC_ID = "ai_features"
|
|
_TTL = 30.0 # seconds before re-reading from Firestore
|
|
|
|
_DEFAULTS: dict[str, bool] = {
|
|
"stt_enabled": True,
|
|
"correlation_enabled": True,
|
|
"summaries_enabled": True,
|
|
"vocabulary_learning_enabled": True,
|
|
# Transcript correction runs inside transcribe_call and spends Gemini
|
|
# tokens plus Places quota on every transcribed call. Until server-26#76
|
|
# it was reachable only through an env var and an ansible run, which meant
|
|
# an "STT-only" evaluation window was never STT-only and its cost could
|
|
# not be attributed (server-26#45).
|
|
#
|
|
# NOT a pure cost lever. The corrector is also the noise gate: it is what
|
|
# sets not_speech, and transcription.py returns nothing for a call it
|
|
# flags. _is_degenerate does not catch what the corrector catches, so with
|
|
# this off, recogniser noise reaches extraction as a real transcript, comes
|
|
# back with no units/tags/location, is judged thin, and auto-attaches to the
|
|
# most recent incident on the talkgroup with no fit check. Turning this off
|
|
# while correlation_enabled is on therefore pushes over-merging -- do not do
|
|
# it during an evaluation window.
|
|
"transcript_correction_enabled": True,
|
|
}
|
|
|
|
_cache: dict[str, Any] = {}
|
|
_cache_ts: float = 0.0
|
|
|
|
|
|
async def get_flags() -> dict[str, bool]:
|
|
"""Return the current feature flags, using the TTL cache when fresh."""
|
|
global _cache, _cache_ts
|
|
|
|
now = time.monotonic()
|
|
if _cache and (now - _cache_ts) < _TTL:
|
|
return dict(_cache)
|
|
|
|
try:
|
|
doc = await fstore.doc_get(_COLLECTION, _DOC_ID)
|
|
if doc:
|
|
merged = {**_DEFAULTS, **{k: bool(v) for k, v in doc.items() if k in _DEFAULTS}}
|
|
else:
|
|
merged = dict(_DEFAULTS)
|
|
except Exception as e:
|
|
logger.warning(f"Feature flags: could not read from Firestore ({e}), using defaults")
|
|
merged = dict(_DEFAULTS)
|
|
|
|
_cache = merged
|
|
_cache_ts = now
|
|
return dict(_cache)
|
|
|
|
|
|
async def set_flags(updates: dict[str, bool]) -> dict[str, bool]:
|
|
"""Write flag updates to Firestore and invalidate the cache."""
|
|
global _cache, _cache_ts
|
|
|
|
clean = {k: bool(v) for k, v in updates.items() if k in _DEFAULTS}
|
|
if not clean:
|
|
raise ValueError(f"No recognised flag keys in update: {list(updates)}")
|
|
|
|
await fstore.doc_set(_COLLECTION, _DOC_ID, clean)
|
|
_cache_ts = 0.0 # force re-read on next get_flags()
|
|
logger.info(f"Feature flags updated: {clean}")
|
|
return await get_flags()
|
|
|
|
|
|
async def resolve_flags(system_id: str | None):
|
|
"""
|
|
Resolve the AI feature flags for one radio system.
|
|
|
|
Returns ``(flags, flag)``: ``flags`` is the raw global config/ai_features
|
|
document, and ``flag(name)`` layers the system's own ``ai_flags`` on top of
|
|
it. A system flag of False beats a global True, but a global False beats
|
|
everything -- config/ai_features is the master switch, which is the whole
|
|
point of having one (server-26#75, server-26#76).
|
|
|
|
Every AI spend path resolves through here. A path that reads ``flags``
|
|
directly re-introduces #75; a path that reads neither re-introduces #76.
|
|
"""
|
|
from app.internal import firestore as _fstore
|
|
|
|
flags = await get_flags()
|
|
|
|
system_ai_flags: dict = {}
|
|
if system_id:
|
|
sys_doc = await _fstore.doc_get_cached("systems", system_id)
|
|
system_ai_flags = (sys_doc or {}).get("ai_flags") or {}
|
|
|
|
def flag(name: str) -> bool:
|
|
if not flags[name]: # global master off
|
|
return False
|
|
return system_ai_flags.get(name, True) # system override, else inherit
|
|
|
|
return flags, flag
|