server-26/drb-c2-core/tests/test_chatter_gate.py

"""
server-26#298 — the chatter pre-filter gates every paid call.

Radio housekeeping (roll call, a bare "Post 4, 10-8, 10-4") used to go through
transcript correction, scene extraction, an embedding per scene and a geocode,
only to come back as a thin routine scene. Now it pays for none of them; the
free rule paths (clearance parse, self-initiated/plate backstop) still run.
"""
from unittest.mock import AsyncMock, patch

import pytest

from app.config import settings
from app.internal import intelligence, transcription
from app.internal.chatter_classifier import gate_chatter


@pytest.fixture(autouse=True)
def _gate_on():
    """The gate ships OFF (shadow) until a live window clears it (#127);
    these tests exercise it switched on."""
    with patch.object(settings, "chatter_gate_enabled", True):
        yield


class _NoPaidCalls:
    """Fail the test if any paid extraction/embedding/geocode call happens."""

    def __enter__(self):
        def _boom(*a, **k):
            raise AssertionError("a paid AI call was made for chatter")

        async def _aboom(*a, **k):
            raise AssertionError("a paid geocode was made for chatter")

        self.ps = [
            patch.object(intelligence, "_sync_extract", _boom),
            patch.object(intelligence, "_sync_embed", _boom),
            patch.object(intelligence, "_geocode_lookup", _aboom),
        ]
        for p in self.ps:
            p.start()
        return self

    def __exit__(self, *a):
        for p in self.ps:
            p.stop()


async def _extract(transcript, talkgroup_name="PD Dispatch"):
    saved: dict = {}

    async def _doc_set(coll, doc_id, data, merge=True):
        saved.update(data)

    with patch.object(intelligence.fstore, "doc_get_cached", AsyncMock(return_value=None)), \
         patch.object(intelligence.fstore, "doc_set", _doc_set):
        scenes = await intelligence.extract_scenes("call-1", transcript, talkgroup_name)
    return scenes, saved


@pytest.mark.asyncio
async def test_bare_acknowledgement_over_five_words_skips_extraction():
    with _NoPaidCalls():
        scenes, saved = await _extract("10-4, 10-4 Central, 98. 10-4, 98.")
    assert scenes == []
    assert saved["skip_reason"] == "non_event_chatter"
    assert saved["severity"] == "routine"
    assert saved["chatter_classifier_verdict"] is True


@pytest.mark.asyncio
async def test_whole_roll_call_skips_extraction():
    roll = ("Car 100, roll call. Henry 1. Henry 1. Sam 1. Sam 1. 45 Baker. 45 Baker. "
            "11 Adam. Adam. 11 Baker. Baker. Staff 1. 1. Staff 2.")
    with _NoPaidCalls():
        scenes, saved = await _extract(roll)
    assert scenes == [] and saved["skip_reason"] == "non_event_chatter"


@pytest.mark.asyncio
async def test_roll_call_mentioned_on_the_way_to_a_real_job_still_extracts():
    """The #127 review's false-positive shape must still reach the model."""
    text = "Delayed for roll call, then respond to a structure fire at 12 Main Street"
    assert gate_chatter(text) == (False, None)
    with patch.object(intelligence, "_sync_extract", return_value=[]) as ext:
        await _extract(text)
    ext.assert_called_once()


@pytest.mark.asyncio
async def test_plate_read_chatter_still_becomes_a_stop_for_free():
    text = "Lincoln Charlie Robert 7-4-0-7, Lincoln Charlie Robert 7-4-0-7."
    with _NoPaidCalls():
        scenes, saved = await _extract(text)
    assert len(scenes) == 1 and scenes[0]["tags"] == ["traffic-stop"]
    assert saved["severity"] == "minor"


@pytest.mark.asyncio
async def test_gate_off_restores_shadow_mode():
    with patch.object(settings, "chatter_gate_enabled", False), \
         patch.object(intelligence, "_sync_extract", return_value=[]) as ext:
        _, saved = await _extract("10-4, 10-4 Central, 98. 10-4, 98.")
    ext.assert_called_once()
    assert saved["chatter_classifier_verdict"] is True, "the shadow verdict is still recorded"


@pytest.mark.asyncio
async def test_real_event_is_not_gated():
    with patch.object(intelligence, "_sync_extract", return_value=[]) as ext:
        await _extract("Shots fired, officer down, requesting backup immediately at the school.")
    ext.assert_called_once()


@pytest.mark.asyncio
async def test_transcription_skips_correction_for_chatter():
    flags = {"transcript_correction_enabled": True}
    with patch.object(settings, "openai_api_key", "k"), \
         patch.object(settings, "groq_api_key", None), \
         patch.object(transcription, "_download", lambda uri: "/nonexistent.flac"), \
         patch.object(transcription, "_sync_transcribe_file", lambda *a, **k: {
             "text": "Post 4, 10-8. 10-4. Copy, 98.", "segments": [],
             "degenerate": False, "duration": 2.0, "confidence": None}), \
         patch.object(transcription.fstore, "doc_set", AsyncMock()), \
         patch.object(transcription, "ai_health") as health, \
         patch.object(transcription.transcript_correction, "correct", AsyncMock()) as correct, \
         patch("app.internal.feature_flags.resolve_flags",
               AsyncMock(return_value=(flags, lambda k: flags.get(k, True)))):
        health.report_healthy = AsyncMock()
        text, _ = await transcription.transcribe_call("call-1", "gs://b/c.flac", "PD Dispatch")
    assert text == "Post 4, 10-8. 10-4. Copy, 98."
    correct.assert_not_awaited()


@pytest.mark.asyncio
async def test_chatter_still_reaches_the_free_thin_correlation_path():
    """non_event_chatter joins transcript_too_short in upload.py's no-scene
    fallback (rules-only for a thin call — no LLM)."""
    from app.routers import upload

    with patch.object(upload, "fstore") as fs, \
         patch.object(upload, "_correlate_with_consensus", AsyncMock(return_value=None)) as corr:
        fs.doc_get = AsyncMock(return_value={"skip_reason": "non_event_chatter", "severity": "routine"})
        fs.doc_set = AsyncMock()
        await upload._extract_and_correlate(
            call_id="c1", node_id="n1", system_id="s1", talkgroup_id=1,
            talkgroup_name="PD Dispatch", transcript="10-4 Central, 98.", scenes=[],
        )
    corr.assert_awaited_once()


@pytest.mark.asyncio
async def test_no_scene_fallback_passes_the_call_doc_severity():
    """server-26#104: a reprocessed call whose doc says 'major' must not be
    re-correlated as 'routine' (thin, no fit check)."""
    from app.routers import upload

    with patch.object(upload, "fstore") as fs, \
         patch.object(upload, "_correlate_with_consensus", AsyncMock(return_value=None)) as corr:
        fs.doc_get = AsyncMock(return_value={"severity": "major"})
        fs.doc_set = AsyncMock()
        await upload._extract_and_correlate(
            call_id="c1", node_id="n1", system_id="s1", talkgroup_id=1,
            talkgroup_name="PD Dispatch", transcript="whatever", scenes=[],
        )
    assert corr.await_args.kwargs["severity"] == "major"


@pytest.mark.asyncio
async def test_gate_off_still_corrects_chatter():
    """With the gate off, transcription must not skip correction either."""
    flags = {"transcript_correction_enabled": True}
    with patch.object(settings, "chatter_gate_enabled", False), \
         patch.object(settings, "openai_api_key", "k"), \
         patch.object(settings, "groq_api_key", None), \
         patch.object(transcription, "_download", lambda uri: "/nonexistent.flac"), \
         patch.object(transcription, "_sync_transcribe_file", lambda *a, **k: {
             "text": "Post 4, 10-8. 10-4. Copy, 98.", "segments": [],
             "degenerate": False, "duration": 2.0, "confidence": None}), \
         patch.object(transcription.fstore, "doc_set", AsyncMock()), \
         patch.object(transcription, "ai_health") as health, \
         patch.object(transcription.transcript_correction, "correct",
                      AsyncMock(return_value=(None, None, False))) as correct, \
         patch("app.internal.feature_flags.resolve_flags",
               AsyncMock(return_value=(flags, lambda k: flags.get(k, True)))):
        health.report_healthy = AsyncMock()
        await transcription.transcribe_call("call-1", "gs://b/c.flac", "PD Dispatch")
    correct.assert_awaited_once()


def test_gate_ships_off():
    from app.config import Settings
    assert Settings.model_fields["chatter_gate_enabled"].default is False