server-26/drb-c2-core/app/internal/chatter_classifier.py

"""
Upstream dispatch-vs-chatter classifier — SHADOW MODE (server-26#115 follow-up).

Three live measurement windows (CORRELATION_REVIEW_0907.md, _0907b.md, _0912.md)
and two consensus-layer fixes (#125, #126) all converged on the same conclusion:
the actual non-event-promotion problem lives upstream of correlation entirely.
Radio housekeeping — unit check-ins, roll call, bare 10-4/10-8/98 acknowledgements
— has no incident content for `intelligence.extract_scenes` to find, but nothing
stops it from being sent to the scene-extraction LLM and coming out the other end
as a thin "scene" for the correlator to then judge. See CORRELATION_REVIEW_0912.md
("Reminder: the real fix is still unscoped") and issue #115.

This module is that classifier. It is a PURE function of the transcript text —
no Firestore, no LLM call, no side effects — so it is cheap to run on every
transcript and cheap to test against real dumps offline.

SHADOW MODE ONLY. As of this module's introduction, nothing skips scene
extraction based on this verdict. `intelligence.extract_scenes` calls
`classify_chatter` purely to record the verdict on the call doc
(`chatter_classifier_verdict` / `chatter_classifier_reason`) so it becomes
observable in the next `/admin` correlation-debug dump, exactly like
`corr_gate_veto` (server-26#115 / PR #126). See the TODO at that call site for
what has to be true before this flips live.

Precision over recall, deliberately. A false positive here — flagging a REAL
event as chatter — would, once live, silently mean that event never gets a
scene, never gets tags/location/severity, and never has a chance to become an
incident. That is a much bigger, harder-to-notice failure than a false
negative (a housekeeping call that still goes through the existing expensive
pipeline and gets judged "not an incident" the same way it is today). When a
transcript doesn't clearly match one of the shapes below, this returns
(False, None) and the existing pipeline runs exactly as it does today.

Patterns are drawn from hand-labeled examples in CORRELATION_REVIEW_0907b.md
and CORRELATION_REVIEW_0912.md, cross-referenced against the real transcripts
in corr_dump_9-7_0437am.json / corr_dump_9-7_pm.json / corr_dump_9-12.json —
not invented regexes. See the backtest script referenced in the PR for the
per-dump catch rate and false-positive count.
"""
import re
from typing import Optional

# Police/law-enforcement phonetic alphabet words (APCO + NATO). Deliberately
# duplicated from intelligence.py's `_PHONETIC_ALPHA_WORDS` rather than
# imported — intelligence.py imports this module (to write the shadow-mode
# verdict onto the call doc), so importing back would be circular. Keep the
# two sets in sync if either changes; they're small and rarely touched.
_PHONETIC_ALPHA_WORDS = frozenset({
    # APCO (law enforcement)
    "adam", "baker", "charles", "david", "edward", "frank", "george", "henry",
    "ida", "john", "king", "lincoln", "mary", "nora", "ocean", "paul", "queen",
    "robert", "sam", "tom", "union", "victor", "william", "x-ray", "young", "zebra",
    # NATO
    "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel",
    "india", "juliet", "kilo", "lima", "mike", "november", "oscar", "papa",
    "quebec", "romeo", "sierra", "tango", "uniform", "whiskey", "yankee", "zulu",
})

_TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9\-]*")

# Bare radio-procedure words that carry zero incident content by themselves.
# Deliberately small and literal — this is not a general stopword list, it's
# the exact vocabulary observed in hand-labeled chatter transcripts. Words
# that are ambiguous outside a pure-procedure context (e.g. "location",
# "call", "phone", "number", "go") are left OUT on purpose: including them
# risks reducing a real, substantive transcript down to nothing.
_FILLER_WORDS = frozenset({
    "to", "this", "is", "the", "a", "and", "for", "you", "can", "i", "in",
    "on", "of", "that", "just", "from", "out", "ok", "okay", "at", "be",
    "show", "me", "mark", "marked", "charge", "standby", "stand", "by",
    "clear", "available", "affirm", "affirmative", "negative", "copy",
    "copies", "received", "roger",
})

# Agency/procedural designators — who's being addressed, not what happened.
_RADIO_DESIGNATORS = frozenset({
    "central", "dispatch", "headquarters", "hq", "post", "unit", "sergeant",
    "sgt", "metro", "mta", "division", "county",
})

_ROLL_CALL_RE = re.compile(r"\broll\s*call\b")


def _tokenize(transcript: str) -> list[str]:
    return _TOKEN_RE.findall(transcript.lower())


def _is_filler_token(token: str) -> bool:
    # Any token starting with a digit is a unit ID, 10-code, badge/post
    # number, or call-number fragment ("10-4", "6-8", "72-holland",
    # "11-victor", "98", "114") — procedural, not incident content. This is
    # deliberately broad: a real event transcript that happens to include a
    # digit-led token (an address number, a case number) still has other,
    # non-digit descriptive words left over, so this alone never reduces a
    # real transcript to nothing. See the backtest for confirmation.
    if token[0].isdigit():
        return True
    return (
        token in _FILLER_WORDS
        or token in _RADIO_DESIGNATORS
        or token in _PHONETIC_ALPHA_WORDS
    )


def classify_chatter(transcript: Optional[str]) -> tuple[bool, Optional[str]]:
    """
    Pure classification of a transcript as non-event radio housekeeping.

    Returns (is_chatter, reason):
      (True,  "roll_call")            — contains a roll-call announcement
      (True,  "bare_acknowledgement") — every token is a callsign/10-code/
                                         procedural filler word; nothing else
      (False, None)                   — not confidently chatter; let the
                                         existing pipeline run as today

    Takes only the transcript. Other call metadata (talkgroup, severity, tags)
    doesn't exist yet at the point this needs to run — this classifier is
    upstream of the scene-extraction call that produces those fields — so it
    deliberately doesn't take them as input.
    """
    if not transcript or not transcript.strip():
        return False, None

    lowered = transcript.lower()
    if _ROLL_CALL_RE.search(lowered):
        return True, "roll_call"

    tokens = _tokenize(transcript)
    if not tokens:
        return False, None

    if any(not _is_filler_token(t) for t in tokens):
        return False, None

    return True, "bare_acknowledgement"