STT-Assistant/tests/test_triage_eval.py

"""Tests for the triage calibration report. Pure, hardware-free."""

import json
from types import SimpleNamespace

from stt_daemon.assistant.triage import TriageAnalyzer
from stt_daemon.triage_eval import (
    analyze_states,
    format_report,
    load_records,
    replay_states,
    suggest_thresholds,
)


def _analyzer(tmp_path, questions):
    path = tmp_path / "questions.json"
    path.write_text(json.dumps(questions), encoding="utf-8")
    return TriageAnalyzer(
        SimpleNamespace(
            questions_file=str(path),
            state_char_budget=1200,
            confidence_field="answer_confidence",
            escalate_threshold=0.6,
            choice_min_confidence=0.85,
        )
    )


def _questions():
    return {
        "blocker": {"type": "noul", "instructions": "Something blocking was raised."},
        "decision_status": {
            "type": "choice",
            "instructions": "Where is the decision?",
            "criteria": {"none": "nothing", "made": "settled"},
        },
    }


def _record(text, speaker="Efren Yevale", status="matched"):
    return {"text": text, "speaker": speaker, "speaker_status": status}


def _write_session(tmp_path, records):
    session = tmp_path / "2026-01-01-00-00-00"
    session.mkdir()
    lines = [json.dumps(record) for record in records]
    (session / "transcripts.jsonl").write_text("\n".join(lines) + "\n", encoding="utf-8")
    return session


def test_load_records_skips_missing_bad_and_empty(tmp_path):
    session = _write_session(tmp_path, [_record("hola")])
    with open(session / "transcripts.jsonl", "a", encoding="utf-8") as handle:
        handle.write("not json\n")
        handle.write(json.dumps({"text": "   "}) + "\n")
    records = load_records(session)
    assert [r["text"] for r in records] == ["hola"]
    assert load_records(tmp_path / "absent") == []


def test_replay_states_labels_speaker_and_honours_stride(tmp_path):
    analyzer = _analyzer(tmp_path, _questions())
    records = [_record(f"line {i}") for i in range(5)]
    states = replay_states(records, analyzer, window=2, stride=2)
    assert states == [
        "- Efren: line 0\n- Efren: line 1",
        "- Efren: line 2\n- Efren: line 3",
        "- Efren: line 4",
    ]


def test_replay_states_marks_unmatched_as_unknown(tmp_path):
    analyzer = _analyzer(tmp_path, _questions())
    states = replay_states([_record("hola", status="unknown")], analyzer, window=4)
    assert states == ["- Unknown: hola"]


def test_analyze_states_reports_fire_rates_and_choice_labels(tmp_path):
    analyzer = _analyzer(tmp_path, _questions())
    payloads = iter(
        [
            {
                "blocker": {"noul": 0.9, "confidence": 0.9, "answer_confidence": 0.9},
                "decision_status": {
                    "choice": "made",
                    "confidence": 0.9,
                    "answer_confidence": 0.9,
                },
            },
            {
                "blocker": {"noul": 0.1, "confidence": 0.9, "answer_confidence": 0.9},
                "decision_status": {
                    "choice": "none",
                    "confidence": 0.9,
                    "answer_confidence": 0.9,
                },
            },
        ]
    )
    reports = analyze_states(
        ["s1", "s2"], analyzer, lambda state: next(payloads), thresholds=(0.5, 0.8)
    )
    by_qid = {report.qid: report for report in reports}
    assert by_qid["blocker"].windows == 2
    assert by_qid["blocker"].fires[0.5] == 1
    assert by_qid["blocker"].fires[0.8] == 1
    assert round(by_qid["blocker"].mean_p, 3) == 0.5
    assert by_qid["decision_status"].labels == {"made": 1, "none": 1}


def test_analyze_states_survives_a_failing_predictor(tmp_path):
    analyzer = _analyzer(tmp_path, _questions())

    def boom(state):
        raise RuntimeError("model gone")

    assert analyze_states(["s1"], analyzer, boom) == []


def test_format_report_prints_fire_rates_and_suggestions(tmp_path):
    analyzer = _analyzer(tmp_path, _questions())
    payload = {
        "blocker": {"noul": 0.70, "confidence": 0.70, "answer_confidence": 0.70},
    }
    reports = analyze_states(["s1", "s2"], analyzer, lambda state: payload, thresholds=(0.5, 0.9))
    text = format_report(reports, 2, thresholds=(0.5, 0.9))
    assert "windows replayed: 2" in text
    assert "fires@0.50" in text
    assert "100%" in text
    assert "suggested per-question thresholds" in text
    assert "blocker 0.90" in text


def test_suggest_thresholds_picks_the_least_restrictive_cut_that_is_rare(tmp_path):
    payloads = iter(
        [
            {"q": {"noul": p, "confidence": p, "answer_confidence": p}}
            for p in (0.95, 0.55, 0.55, 0.55)
        ]
    )
    reports = analyze_states(
        ["s1", "s2", "s3", "s4"],
        _analyzer(tmp_path, {"q": {"type": "noul", "instructions": "x"}}),
        lambda state: next(payloads),
        thresholds=(0.5, 0.7, 0.9),
    )
    assert suggest_thresholds(reports, (0.5, 0.7, 0.9), target=0.25) == {"q": 0.7}