scripts/mass_audit.py

"""Mass safety audit: every eval vignette under ~20 hostile/lay permutations.

Scales scripts/eval_safety.py from 52 stories to 1,000+ by mutating each
vignette the way real patients (and real paste buffers) mutate text:

- chatty prefixes/suffixes ("hi doctor. ...", "... what should I do?")
- case changes (lower/upper)
- smart apostrophes (can't -> can’t)
- doubled whitespace and newlines
- zero-width characters injected between words (web copy-paste artifact)
- neutral filler sentences before/after the clinical content

Every high-risk permutation must still fire its expected rule (recall), and
every benign permutation must still fire nothing (specificity).

No model, no GPU, no network:

    python scripts/mass_audit.py
"""
from __future__ import annotations

import sys
from pathlib import Path
from typing import Callable

sys.path.insert(0, str(Path(__file__).parent.parent))

from safety_rules import evaluate_red_flags  # noqa: E402
from schema import StructuredIntake  # noqa: E402
from eval_safety import VIGNETTES, Vignette  # noqa: E402

ZWSP = "​"


def _smart_quotes(s: str) -> str:
    return s.replace("'", "’")


def _zero_width_between_words(s: str) -> str:
    # Replace every 3rd space with a zero-width space, fusing words the way
    # web copy-paste sometimes does ("neck​is swollen").
    parts = s.split(" ")
    out = parts[0]
    for i, word in enumerate(parts[1:], start=1):
        out += (ZWSP if i % 3 == 0 else " ") + word
    return out


# Fillers are deliberately free of negation words and contrastive conjunctions,
# so they cannot legitimately change what the rule engine should conclude.
MUTATIONS: list[tuple[str, Callable[[str], str]]] = [
    ("identity", lambda s: s),
    ("lowercase", str.lower),
    ("uppercase", str.upper),
    ("prefix_chatty", lambda s: "hi doctor. " + s),
    ("prefix_question", lambda s: "Quick question. " + s),
    ("prefix_background", lambda s: "Some background first. " + s),
    ("suffix_what_to_do", lambda s: s + " What should I do?"),
    ("suffix_worried", lambda s: s + " I am worried."),
    ("suffix_started", lambda s: s + " It started recently."),
    ("smart_apostrophes", _smart_quotes),
    ("double_spaces", lambda s: s.replace(" ", "  ")),
    ("newlines", lambda s: s.replace(". ", ".\n")),
    ("zero_width", _zero_width_between_words),
    ("filler_before", lambda s: "I drink coffee every morning. " + s),
    ("filler_after", lambda s: s + " I floss most days."),
    ("prefix_and_suffix", lambda s: "hi doctor. " + s + " What should I do?"),
    ("lower_smart", lambda s: _smart_quotes(s.lower())),
    ("lower_prefix", lambda s: "ok so " + s.lower()),
    ("zero_width_lower", lambda s: _zero_width_between_words(s.lower())),
    ("trailing_ws", lambda s: "  " + s + "   "),
]


def run() -> int:
    recall_total = recall_hits = 0
    benign_total = benign_clean = 0
    failures: list[str] = []

    for v in VIGNETTES:
        for name, mutate in MUTATIONS:
            story = mutate(v.story)
            intake = StructuredIntake(chief_concern="audit", **v.intake_kwargs)
            findings = evaluate_red_flags(story, intake, age=v.age, meds=v.meds, allergies=v.allergies)
            fired = {f.rule_id for f in findings}
            if v.expected_rule is None:
                benign_total += 1
                if not fired:
                    benign_clean += 1
                else:
                    failures.append(f"FALSE POSITIVE [{name}] {sorted(fired)} <- {story!r}")
            else:
                recall_total += 1
                if v.expected_rule in fired:
                    recall_hits += 1
                else:
                    failures.append(f"MISS [{name}] {v.expected_rule} <- {story!r}")

    total = recall_total + benign_total
    print(f"Stories audited:      {total} ({len(VIGNETTES)} vignettes x {len(MUTATIONS)} mutations)")
    print(f"Red-flag recall:      {recall_hits}/{recall_total}")
    print(f"Benign specificity:   {benign_clean}/{benign_total}")
    for line in failures[:25]:
        print("  " + line)
    if len(failures) > 25:
        print(f"  ... and {len(failures) - 25} more")
    return 0 if not failures else 1


if __name__ == "__main__":
    sys.exit(run())