diff --git a/drb-c2-core/app/internal/feature_flags.py b/drb-c2-core/app/internal/feature_flags.py index 9fcb7e1..16c650f 100644 --- a/drb-c2-core/app/internal/feature_flags.py +++ b/drb-c2-core/app/internal/feature_flags.py @@ -19,6 +19,21 @@ _DEFAULTS: dict[str, bool] = { "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] = {} @@ -60,3 +75,33 @@ async def set_flags(updates: dict[str, bool]) -> dict[str, bool]: _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 diff --git a/drb-c2-core/app/internal/summarizer.py b/drb-c2-core/app/internal/summarizer.py index a77f6ea..0cb8914 100644 --- a/drb-c2-core/app/internal/summarizer.py +++ b/drb-c2-core/app/internal/summarizer.py @@ -25,9 +25,14 @@ async def summarizer_loop() -> None: flags = await get_flags() if flags["summaries_enabled"]: await _run_summary_pass() - await _resolve_stale_incidents() else: - logger.info("Summaries disabled — skipping summary pass and stale incident sweep") + logger.info("Summaries disabled — skipping summary pass") + # Deliberately outside the flag. Auto-resolving a quiet incident is + # pure Firestore with no model call in it, and gating it behind the + # AI kill switch meant nothing ever auto-resolved in the standing + # flags-off configuration — leaving every incident "active" forever + # and growing the candidate set every correlation reads. + await _resolve_stale_incidents() except Exception as e: logger.error(f"Summarizer pass failed: {e}") @@ -43,10 +48,17 @@ async def _run_summary_pass() -> None: async def _summarize_incident(inc: dict) -> None: + from app.internal.feature_flags import get_flags + incident_id = inc.get("incident_id") if not incident_id: return + flags = await get_flags() + if not flags["summaries_enabled"]: + logger.info(f"Summaries disabled — skipping summary for incident {incident_id}") + return + call_ids: list[str] = inc.get("call_ids", []) if not call_ids: return diff --git a/drb-c2-core/app/internal/transcription.py b/drb-c2-core/app/internal/transcription.py index d26354b..fc7a604 100644 --- a/drb-c2-core/app/internal/transcription.py +++ b/drb-c2-core/app/internal/transcription.py @@ -205,12 +205,25 @@ async def transcribe_call( # correlation all consume the transcript — correcting it afterwards # (which is where it used to live, inside the extraction prompt) meant # every one of them reasoned over known-bad text. server-26#36. - corrected, corrected_segments, not_speech = await transcript_correction.correct( - call_id, transcript, segments, - system_id=system_id, - talkgroup_id=talkgroup_id, - talkgroup_name=talkgroup_name, - ) + # Correction is a second model call plus a Places lookup per proposed + # location, so it is real spend that used to be reachable only through + # an env var and an ansible run. That made an "STT-only" evaluation + # window not STT-only, and its cost unattributable (server-26#76, #45). + from app.internal.feature_flags import resolve_flags + _, _ai_flag = await resolve_flags(system_id) + + corrected, corrected_segments, not_speech = (None, None, False) + if _ai_flag("transcript_correction_enabled"): + corrected, corrected_segments, not_speech = await transcript_correction.correct( + call_id, transcript, segments, + system_id=system_id, + talkgroup_id=talkgroup_id, + talkgroup_name=talkgroup_name, + ) + else: + logger.info( + f"Transcript correction disabled — saving raw transcript for call {call_id}" + ) if corrected_segments: # Raw stays as evidence; the corrected copy is what extraction reads. updates["segments_corrected"] = corrected_segments diff --git a/drb-c2-core/app/routers/calls.py b/drb-c2-core/app/routers/calls.py index 1a70e80..9663981 100644 --- a/drb-c2-core/app/routers/calls.py +++ b/drb-c2-core/app/routers/calls.py @@ -229,10 +229,26 @@ async def patch_transcript( _: 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, { @@ -271,7 +287,7 @@ async def patch_transcript( # 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: + 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) diff --git a/drb-c2-core/app/routers/incidents.py b/drb-c2-core/app/routers/incidents.py index d2b1eba..77305a1 100644 --- a/drb-c2-core/app/routers/incidents.py +++ b/drb-c2-core/app/routers/incidents.py @@ -101,13 +101,17 @@ async def summarize_incident( ): """Immediately run the summarizer for a specific incident.""" from app.internal.summarizer import _summarize_incident + from app.internal.feature_flags import get_flags inc = await fstore.doc_get("incidents", incident_id) if not inc: raise HTTPException(404, f"Incident '{incident_id}' not found.") + flags = await get_flags() + if not flags["summaries_enabled"]: + return {"ok": False, "incident_id": incident_id, "summaries_enabled": False} # Rate limit by incident ID to prevent repeated expensive LLM calls summarize_limiter.check(incident_id) background_tasks.add_task(_summarize_incident, inc) - return {"ok": True, "incident_id": incident_id} + return {"ok": True, "incident_id": incident_id, "summaries_enabled": True} @router.post("/{incident_id}/calls/{call_id}") diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index b0de9a3..a661016 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/drb-c2-core/app/routers/upload.py @@ -159,6 +159,19 @@ async def _correlate_with_consensus( return await incident_correlator.apply_correlation({"decision": final, "ctx": ctx}) +async def _resolve_flags(system_id: Optional[str]): + """ + Resolve AI feature flags for a given system. + + Thin alias for `feature_flags.resolve_flags` — the resolver lives there + because transcription and the calls router need the same answer, and three + copies of it is how server-26#75 happened in the first place. + """ + from app.internal.feature_flags import resolve_flags + + return await resolve_flags(system_id) + + async def _run_extraction_pipeline( call_id: str, node_id: str, @@ -172,48 +185,55 @@ async def _run_extraction_pipeline( """Run steps 2-4 of the intelligence pipeline using an existing transcript.""" from app.internal import intelligence, incident_correlator, alerter - # Step 2: Scene detection + intelligence extraction. - # Returns one scene per distinct incident detected in the recording. - scenes = await intelligence.extract_scenes( - call_id, transcript, talkgroup_name, - talkgroup_id=talkgroup_id, system_id=system_id, segments=segments, - node_id=node_id, - preserve_transcript_correction=preserve_transcript_correction, - ) + flags, _flag = await _resolve_flags(system_id) - # Step 3: Correlate each scene to an incident independently. incident_ids: list[str] = [] all_tags: list[str] = [] - for scene in scenes: - all_tags.extend(scene["tags"]) - # When dispatch is pulling a unit to a NEW call (reassignment), suppress unit - # overlap so the new scene doesn't chain into the unit's previous incident. - is_reassignment = bool(scene.get("reassignment")) - corr_units = [] if is_reassignment else scene.get("units") - incident_id = await _correlate_with_consensus( - call_id=call_id, + + if _flag("correlation_enabled"): + # Step 2: Scene detection + intelligence extraction. + # Returns one scene per distinct incident detected in the recording. + scenes = await intelligence.extract_scenes( + call_id, transcript, talkgroup_name, + talkgroup_id=talkgroup_id, system_id=system_id, segments=segments, node_id=node_id, - system_id=system_id, - talkgroup_id=talkgroup_id, - talkgroup_name=talkgroup_name, - tags=scene["tags"], - incident_type=scene["incident_type"], - location=scene["location"], - location_coords=scene["location_coords"], - units=corr_units, - vehicles=scene.get("vehicles"), - cleared_units=scene.get("cleared_units"), - reassignment=is_reassignment, + preserve_transcript_correction=preserve_transcript_correction, ) - if incident_id and incident_id not in incident_ids: - incident_ids.append(incident_id) - if scene["resolved"] and incident_id: - await fstore.doc_set("incidents", incident_id, { - "status": "resolved", - "resolved_at": datetime.now(timezone.utc).isoformat(), - }) - await incident_correlator.maybe_resolve_parent(incident_id) - logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)") + + # Step 3: Correlate each scene to an incident independently. + for scene in scenes: + all_tags.extend(scene["tags"]) + # When dispatch is pulling a unit to a NEW call (reassignment), suppress unit + # overlap so the new scene doesn't chain into the unit's previous incident. + is_reassignment = bool(scene.get("reassignment")) + corr_units = [] if is_reassignment else scene.get("units") + incident_id = await _correlate_with_consensus( + call_id=call_id, + node_id=node_id, + system_id=system_id, + talkgroup_id=talkgroup_id, + talkgroup_name=talkgroup_name, + tags=scene["tags"], + incident_type=scene["incident_type"], + location=scene["location"], + location_coords=scene["location_coords"], + units=corr_units, + vehicles=scene.get("vehicles"), + cleared_units=scene.get("cleared_units"), + reassignment=is_reassignment, + ) + if incident_id and incident_id not in incident_ids: + incident_ids.append(incident_id) + if scene["resolved"] and incident_id: + await fstore.doc_set("incidents", incident_id, { + "status": "resolved", + "resolved_at": datetime.now(timezone.utc).isoformat(), + }) + await incident_correlator.maybe_resolve_parent(incident_id) + logger.info(f"Auto-resolved incident {incident_id} (LLM closure detection)") + else: + scope = "globally" if not flags["correlation_enabled"] else f"system {system_id}" + logger.info(f"Correlation disabled ({scope}) — skipping scene extraction and correlation for call {call_id} (reprocess)") if incident_ids: await fstore.doc_set("calls", call_id, {"incident_ids": incident_ids}) @@ -245,7 +265,6 @@ async def _run_intelligence_pipeline( 4. Check alert rules and dispatch notifications """ from app.internal import transcription, intelligence, incident_correlator, alerter, talkgroups - from app.internal.feature_flags import get_flags # The node only sends talkgroup_name when OP25 had it in the loaded tags # file, so it arrives empty for exactly the talkgroups C2 can name from the @@ -265,19 +284,7 @@ async def _run_intelligence_pipeline( except Exception as e: logger.warning(f"Could not backfill talkgroup_name on call {call_id}: {e}") - flags = await get_flags() - - # Resolve per-system overrides: system flag=False beats global flag=True, - # but global flag=False beats everything (master switch). - 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, default inherit + flags, _flag = await _resolve_flags(system_id) transcript: Optional[str] = None segments: list[dict] = [] @@ -310,7 +317,7 @@ async def _run_intelligence_pipeline( # A single recording can produce multiple incidents on a busy channel. incident_ids: list[str] = [] all_tags: list[str] = [] - if flags["correlation_enabled"]: + if _flag("correlation_enabled"): for scene in scenes: all_tags.extend(scene["tags"]) is_reassignment = bool(scene.get("reassignment")) diff --git a/drb-c2-core/tests/test_ai_flag_gating.py b/drb-c2-core/tests/test_ai_flag_gating.py new file mode 100644 index 0000000..b7524c0 --- /dev/null +++ b/drb-c2-core/tests/test_ai_flag_gating.py @@ -0,0 +1,252 @@ +""" +The AI feature flags have to be an enforceable statement about the system, +not just about the ingest path (server-26#75, server-26#76). + +Three defects motivate these tests: + + #75 Correlation read the raw global config/ai_features flag instead of the + per-system resolution, so a system that had opted out via its own + ai_flags still correlated -- with empty tags, down the thin/recency + path, blindly attaching to whatever incident was most recent. + + #76 Transcript correction and the transcript-PATCH extraction path checked + no Firestore flag at all, so "AI is off" still spent money. + + Plus the destructive half of PATCH /calls/{id}/transcript, which wipes a + call's intelligence fields on the promise that re-extraction rebuilds them. + +Firestore and the lazily-imported pipeline modules are fully mocked; the +functions are called directly rather than through FastAPI. +""" +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from fastapi import HTTPException + +from app.routers import upload, calls +from app.internal import summarizer, transcription + + +ALL_ON = { + "stt_enabled": True, + "correlation_enabled": True, + "summaries_enabled": True, + "vocabulary_learning_enabled": True, + "transcript_correction_enabled": True, +} + + +def _flags(**overrides): + return {**ALL_ON, **overrides} + + +def _system(ai_flags): + return {"system_id": "sys-1", "ai_flags": ai_flags or {}} + + +def _patch_flags(global_flags, system_ai_flags): + """Patch the two reads resolve_flags() makes: the global doc and the system doc.""" + return ( + patch("app.internal.feature_flags.get_flags", AsyncMock(return_value=global_flags)), + patch("app.internal.firestore.doc_get_cached", + AsyncMock(return_value=_system(system_ai_flags))), + ) + + +# -------------------------------------------------------------------------- +# resolve_flags: global master off beats everything, system false beats +# global true, absent system key inherits global. +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize( + "global_on, system_ai_flags, expected", + [ + (True, {"correlation_enabled": False}, False), # #75: system opt-out holds + (True, {}, True), # absent -> inherit global + (True, {"correlation_enabled": True}, True), + (False, {"correlation_enabled": True}, False), # global is the master switch + (False, {}, False), + ], +) +@pytest.mark.asyncio +async def test_resolve_flags_precedence(global_on, system_ai_flags, expected): + g, sysdoc = _patch_flags(_flags(correlation_enabled=global_on), system_ai_flags) + with g, sysdoc: + _, flag = await upload._resolve_flags("sys-1") + + assert flag("correlation_enabled") is expected + + +@pytest.mark.asyncio +async def test_resolve_flags_without_a_system_id_does_not_read_the_system_doc(): + with patch("app.internal.feature_flags.get_flags", AsyncMock(return_value=_flags())), \ + patch("app.internal.firestore.doc_get_cached", AsyncMock()) as cached: + _, flag = await upload._resolve_flags(None) + + assert flag("correlation_enabled") is True + cached.assert_not_awaited() + + +# -------------------------------------------------------------------------- +# The ingest path. This is the exact shape of #75: with the global on and the +# system opted out, extraction was skipped but the empty-scenes fallback still +# ran, correlating the call with no tags and attaching it to whatever incident +# was most recent on that system. +# -------------------------------------------------------------------------- + +async def _run_ingest(global_correlation, system_ai_flags): + g, sysdoc = _patch_flags( + _flags(correlation_enabled=global_correlation), system_ai_flags + ) + with g, sysdoc, \ + patch.object(upload, "fstore") as fs, \ + patch.object(upload, "_correlate_with_consensus", AsyncMock(return_value=None)) as corr, \ + patch("app.internal.transcription.transcribe_call", + AsyncMock(return_value=("units respond to main street", []))), \ + patch("app.internal.intelligence.extract_scenes", AsyncMock(return_value=[])) as scenes, \ + patch("app.internal.alerter.check_and_dispatch", AsyncMock()): + fs.doc_get = AsyncMock(return_value={}) + fs.doc_set = AsyncMock() + await upload._run_intelligence_pipeline( + call_id="call-1", + node_id="node-1", + system_id="sys-1", + talkgroup_id=101, + talkgroup_name="PD Dispatch", + gcs_uri="gs://bucket/call-1.mp3", + ) + return scenes, corr + + +@pytest.mark.asyncio +async def test_per_system_opt_out_blocks_the_blind_recency_fallback_too(): + scenes, corr = await _run_ingest(True, {"correlation_enabled": False}) + + scenes.assert_not_awaited() + # The regression that mattered: the no-scenes fallback correlating on empty tags. + corr.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_ingest_correlates_when_the_system_has_not_opted_out(): + scenes, corr = await _run_ingest(True, {}) + + scenes.assert_awaited_once() + corr.assert_awaited_once() + + +# -------------------------------------------------------------------------- +# _run_extraction_pipeline -- the transcript-PATCH path (#76). +# -------------------------------------------------------------------------- + +async def _run_extraction(global_correlation, system_ai_flags=None): + g, sysdoc = _patch_flags( + _flags(correlation_enabled=global_correlation), system_ai_flags + ) + with g, sysdoc, \ + patch.object(upload, "fstore") as fs, \ + patch("app.internal.intelligence.extract_scenes", AsyncMock(return_value=[])) as scenes, \ + patch("app.internal.alerter.check_and_dispatch", AsyncMock()) as alert: + fs.doc_set = AsyncMock() + await upload._run_extraction_pipeline( + call_id="call-1", + node_id="node-1", + system_id="sys-1", + talkgroup_id=101, + talkgroup_name="PD Dispatch", + transcript="units respond to main street", + ) + return scenes, alert, fs + + +@pytest.mark.asyncio +async def test_extraction_does_not_spend_when_correlation_is_off(): + scenes, alert, fs = await _run_extraction(False) + + scenes.assert_not_awaited() + # No incidents produced, so nothing may be stamped onto the call doc. + fs.doc_set.assert_not_awaited() + # Alerting is rule-based and free -- it still runs. + alert.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_extraction_respects_a_per_system_opt_out(): + scenes, _alert, _fs = await _run_extraction(True, {"correlation_enabled": False}) + + scenes.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_extraction_runs_when_the_flag_is_on(): + scenes, _alert, _fs = await _run_extraction(True) + + scenes.assert_awaited_once() + + +# -------------------------------------------------------------------------- +# PATCH /calls/{id}/transcript is destructive before it is constructive. +# With correlation off it must refuse rather than blank the call out. +# -------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_transcript_patch_refuses_when_correlation_is_off(): + g, sysdoc = _patch_flags(_flags(correlation_enabled=False), {}) + with g, sysdoc, patch.object(calls, "fstore") as fs: + fs.doc_get = AsyncMock(return_value={"call_id": "call-1", "system_id": "sys-1"}) + fs.doc_set = AsyncMock() + with pytest.raises(HTTPException) as exc: + await calls.patch_transcript( + call_id="call-1", + body=MagicMock(transcript="corrected text"), + background_tasks=MagicMock(), + _={}, + ) + + assert exc.value.status_code == 409 + # The refusal has to land before the first write, or the call is already ruined. + fs.doc_set.assert_not_awaited() + + +# -------------------------------------------------------------------------- +# Transcript correction is a second model call plus a Places lookup (#76). +# -------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_transcript_correction_is_skipped_when_its_flag_is_off(): + g, sysdoc = _patch_flags(_flags(transcript_correction_enabled=False), {}) + with g, sysdoc, \ + patch.object(transcription, "fstore") as fs, \ + patch.object(transcription, "ai_health") as health, \ + patch.object(transcription, "transcript_correction") as tc, \ + patch("asyncio.to_thread", AsyncMock(return_value=("units respond", [], False))): + fs.doc_set = AsyncMock() + health.report_healthy = AsyncMock() + health.report_failure = AsyncMock() + tc.correct = AsyncMock() + await transcription.transcribe_call( + "call-1", "gs://bucket/call-1.mp3", "PD Dispatch", system_id="sys-1" + ) + + tc.correct.assert_not_awaited() + + +# -------------------------------------------------------------------------- +# Summarizer: the flag guards model spend, not the free Firestore sweep. +# -------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_summarize_incident_is_a_no_op_when_summaries_are_off(): + with patch("app.internal.feature_flags.get_flags", + AsyncMock(return_value=_flags(summaries_enabled=False))), \ + patch.object(summarizer, "fstore") as fs, \ + patch.object(summarizer, "_sync_summarize") as sync: + fs.doc_get = AsyncMock() + fs.doc_set = AsyncMock() + await summarizer._summarize_incident( + {"incident_id": "inc-1", "call_ids": ["call-1"]} + ) + + sync.assert_not_called() + fs.doc_get.assert_not_awaited() + fs.doc_set.assert_not_awaited()