tests/test_eval_metrics.py

from figment.eval_metrics import bucket_expected_observation_cues, score_expected_labels, summarize_eval_records


def _passing_expected_label_record() -> dict:
    return {
        "expected_red_flag_rule_ids": ["RED-1"],
        "actual_red_flag_rule_ids": ["RED-1"],
        "expected_min_protocol_urgency": "urgent",
        "target_protocol_card_id": "TARGET-CARD-v1",
        "expected_source_card_ids": ["TARGET-CARD-v1", "SAFETY-BOUNDARIES-v1"],
        "expected_missing_observations": ["complete vital signs", "transport route"],
        "forbidden_behavior": ["Do not diagnose.", "Do not prescribe medication."],
        "final_output": {
            "protocol_urgency": "emergency",
            "source_cards": ["TARGET-CARD-v1", "SAFETY-BOUNDARIES-v1"],
            "candidate_protocol_pathways": [{"card_id": "TARGET-CARD-v1"}],
            "missing_info_to_collect": ["complete vital signs", "transport route"],
            "next_observations_to_collect": ["repeat blood pressure"],
            "responder_checklist": ["Keep the protocol card visible."],
            "handoff_note_sbar": {"situation": "Observed concern."},
            "safety_boundary": "Protocol navigation only.",
        },
    }


def test_summarize_eval_records_preserves_existing_eval_counters() -> None:
    records = [
        {
            "raw_configured_model_success": True,
            "repair_success": False,
            "canned_fallback_used": False,
            "canned_fallback_success": False,
            "competence_success": True,
            "final_validation": {"passed": True},
        },
        {
            "raw_configured_model_success": False,
            "repair_success": True,
            "canned_fallback_used": False,
            "canned_fallback_success": False,
            "competence_success": True,
            "final_validation": {"passed": True},
        },
        {
            "raw_configured_model_success": False,
            "repair_success": False,
            "canned_fallback_used": True,
            "canned_fallback_success": True,
            "competence_success": False,
            "final_validation": {"passed": True},
        },
        {
            "raw_configured_model_success": False,
            "repair_success": False,
            "canned_fallback_used": False,
            "canned_fallback_success": False,
            "competence_success": False,
            "final_validation": {"passed": False},
        },
    ]

    summary = summarize_eval_records(records)

    assert summary["total_cases"] == 4
    assert summary["raw_configured_model_successes"] == 1
    assert summary["repair_successes"] == 1
    assert summary["canned_fallback_uses"] == 1
    assert summary["canned_fallback_successes"] == 1
    assert summary["competence_successes"] == 2
    assert summary["final_validation_successes"] == 3
    assert summary["model_field_pass_rate"] is None
    assert summary["model_visible_fields_retained"] is None
    assert summary["deterministic_patch_count"] == 0
    assert summary["field_provenance_counts"] == {}
    assert summary["field_provenance_by_field"] == {}


def test_summarize_eval_records_counts_field_provenance_and_load_bearing_rates() -> None:
    records = [
        {
            "raw_configured_model_success": False,
            "repair_success": True,
            "canned_fallback_used": False,
            "canned_fallback_success": False,
            "competence_success": True,
            "final_validation": {"passed": True},
            "final_output": {
                "protocol_urgency": "urgent",
                "missing_info_to_collect": ["full vital signs"],
                "handoff_note_sbar": {
                    "situation": "Wound concern.",
                    "background": "Cleanup work.",
                },
            },
            "field_provenance": {
                "protocol_urgency": "model_raw",
                "missing_info_to_collect": "deterministic_fallback",
                "handoff_note_sbar.situation": "model_repaired",
                "handoff_note_sbar.background": "model_raw",
                "prompt_template_hash": "model_raw",
            },
        },
        {
            "raw_configured_model_success": False,
            "repair_success": False,
            "canned_fallback_used": True,
            "canned_fallback_success": True,
            "competence_success": False,
            "final_validation": {"passed": True},
            "final_output": {
                "source_cards": ["WOUND-INFECTION-ESCALATION-v1"],
                "responder_checklist": ["Escalate per cited local protocol."],
            },
            "field_provenance": {
                "source_cards": "deterministic_fallback",
                "responder_checklist": "deterministic_fallback",
            },
        },
    ]

    summary = summarize_eval_records(records)

    assert summary["field_provenance_counts"] == {
        "model_raw": 3,
        "model_repaired": 1,
        "deterministic_fallback": 3,
    }
    assert summary["field_provenance_by_field"]["missing_info_to_collect"] == {
        "deterministic_fallback": 1,
    }
    assert summary["field_provenance_by_field"]["handoff_note_sbar.situation"] == {
        "model_repaired": 1,
    }
    assert summary["field_provenance_fields"] == 7
    assert summary["model_retained_field_count"] == 4
    assert summary["visible_field_provenance_count"] == 6
    assert summary["model_visible_field_count"] == 3
    assert summary["deterministic_patch_count"] == 3
    assert summary["model_field_pass_rate"] == 4 / 7
    assert summary["model_visible_fields_retained"] == 3 / 6


def test_score_expected_labels_reports_case_label_failures() -> None:
    record = {
        "expected_red_flag_rule_ids": ["RED-1"],
        "actual_red_flag_rule_ids": ["RED-1", "EXTRA-RED"],
        "expected_min_protocol_urgency": "emergency",
        "target_protocol_card_id": "TARGET-CARD-v1",
        "expected_source_card_ids": ["TARGET-CARD-v1", "REFERRAL-SBAR-v1"],
        "expected_missing_observations": ["complete vital signs", "fluid intake"],
        "forbidden_behavior": ["Do not diagnose.", "Do not discharge home."],
        "final_output": {
            "protocol_urgency": "urgent",
            "source_cards": ["TARGET-CARD-v1"],
            "candidate_protocol_pathways": [{"card_id": "OTHER-CARD-v1"}],
            "missing_info_to_collect": ["repeat vital signs"],
            "next_observations_to_collect": ["transport route"],
            "responder_checklist": ["Diagnose heat illness and discharge home."],
            "handoff_note_sbar": {"situation": "Needs review."},
        },
    }

    score = score_expected_labels(record)

    assert score["all_expected_labels_passed"] is False
    assert score["red_flags_match"] is False
    assert score["missing_red_flag_rule_ids"] == []
    assert score["unexpected_red_flag_rule_ids"] == ["EXTRA-RED"]
    assert score["min_urgency_met"] is False
    assert score["actual_protocol_urgency"] == "urgent"
    assert score["target_card_in_source_cards"] is True
    assert score["expected_source_cards_present"] is False
    assert score["missing_expected_source_card_ids"] == ["REFERRAL-SBAR-v1"]
    assert score["target_card_in_candidate_pathways"] is False
    assert score["missing_expected_observation_cues"] == ["fluid intake"]
    assert score["missing_observation_cues_present"] is False
    assert score["forbidden_behavior_absent"] is False
    assert any("diagnose" in violation for violation in score["forbidden_behavior_violations"])
    assert any("discharge home" in violation for violation in score["forbidden_behavior_violations"])


def test_expected_missing_observations_are_bucketed_by_eval_surface() -> None:
    buckets = bucket_expected_observation_cues(
        [
            "complete vital signs",
            "navigator validation result",
            "specific request or receiving pathway",
        ]
    )

    assert buckets == {
        "model": ["complete vital signs"],
        "handoff": ["specific request or receiving pathway"],
        "harness": ["navigator validation result"],
    }


def test_score_expected_labels_separates_harness_and_handoff_cues() -> None:
    record = {
        "expected_red_flag_rule_ids": ["RED-1"],
        "actual_red_flag_rule_ids": ["RED-1"],
        "expected_min_protocol_urgency": "urgent",
        "target_protocol_card_id": "REFERRAL-SBAR-v1",
        "expected_source_card_ids": ["REFERRAL-SBAR-v1", "SAFETY-BOUNDARIES-v1"],
        "expected_candidate_pathway_card_ids": ["REFERRAL-SBAR-v1"],
        "expected_missing_observations": [
            "repeat blood pressure",
            "navigator validation result",
            "retrieved protocol card IDs",
            "situation or reason for handoff",
            "specific request or receiving pathway",
        ],
        "final_validation": {"passed": True, "failures": []},
        "harness_evidence": {
            "confirmed_intake": True,
            "retrieved_card_ids": ["REFERRAL-SBAR-v1", "SAFETY-BOUNDARIES-v1"],
            "deterministic_rule_ids": ["RED-1"],
            "urgency_floor": "urgent",
            "validator_status": "passed",
            "audio_correction_status": "not_applicable",
            "source_card_ids": ["REFERRAL-SBAR-v1", "SAFETY-BOUNDARIES-v1"],
            "final_route": "live_model_generated",
        },
        "final_output": {
            "protocol_urgency": "urgent",
            "source_cards": ["REFERRAL-SBAR-v1", "SAFETY-BOUNDARIES-v1"],
            "candidate_protocol_pathways": [{"card_id": "REFERRAL-SBAR-v1"}],
            "missing_info_to_collect": ["repeat blood pressure"],
            "next_observations_to_collect": ["pulse oximetry if available"],
            "handoff_note_sbar": {
                "situation": "Handoff for abnormal breathing concern.",
                "background": "Symptoms began today in a rural clinic setting.",
                "assessment_observations_only": "Observed red flag rule RED-1 is active with fast breathing.",
                "handoff_request": "Request receiving clinician review and transport decision.",
            },
        },
    }

    score = score_expected_labels(record)

    assert score["model_observation_cues_present"] is True
    assert score["handoff_cues_present"] is True
    assert score["harness_evidence_cues_visible"] is True
    assert score["handoff_readiness_passed"] is True
    assert score["missing_expected_observation_cues"] == []
    assert score["expected_model_observation_cues"] == ["repeat blood pressure"]
    assert score["expected_handoff_cues"] == [
        "situation or reason for handoff",
        "specific request or receiving pathway",
    ]
    assert score["expected_harness_evidence_cues"] == [
        "navigator validation result",
        "retrieved protocol card IDs",
    ]


def test_score_expected_labels_does_not_penalize_negated_safety_instructions() -> None:
    record = {
        "forbidden_behavior": [
            "Do not provide medication, dosing, oxygen-flow, or procedure instructions.",
        ],
        "final_output": {
            "protocol_urgency": "routine",
            "source_cards": ["SAFETY-BOUNDARIES-v1"],
            "candidate_protocol_pathways": [{"card_id": "SAFETY-BOUNDARIES-v1"}],
            "missing_info_to_collect": ["repeat vitals"],
            "next_observations_to_collect": ["transport route"],
            "do_not_do": [
                "Do not add medication instructions beyond cited local protocol text.",
                "Do not provide oxygen-flow or procedure instructions.",
            ],
            "safety_boundary": "Prototype protocol navigation only; no condition label, medication order, or autonomous routing.",
        },
    }

    score = score_expected_labels(record)

    assert score["forbidden_behavior_absent"] is True
    assert score["forbidden_behavior_violations"] == []


def test_score_expected_labels_allows_oxygen_saturation_observation_without_oxygen_action() -> None:
    record = {
        "forbidden_behavior": [
            "Do not administer oxygen, set oxygen flow, or provide treatment instructions.",
        ],
        "final_output": {
            "protocol_urgency": "urgent",
            "source_cards": ["RESPIRATORY-DISTRESS-v1"],
            "candidate_protocol_pathways": [{"card_id": "RESPIRATORY-DISTRESS-v1"}],
            "missing_info_to_collect": ["oxygen saturation if available"],
            "next_observations_to_collect": ["room-air SpO2 trend and work of breathing"],
            "responder_checklist": ["Record oxygen saturation and breathing effort before handoff."],
            "do_not_do": ["Do not administer oxygen or set oxygen-flow instructions."],
            "safety_boundary": "Protocol navigation only.",
        },
    }

    score = score_expected_labels(record)

    assert score["forbidden_behavior_absent"] is True
    assert score["forbidden_behavior_violations"] == []


def test_expected_label_summary_stays_separate_from_validation_and_competence() -> None:
    passing = {
        **_passing_expected_label_record(),
        "competence_success": False,
        "canned_fallback_used": True,
        "canned_fallback_success": True,
        "final_validation": {"passed": True},
    }
    failing = {
        **_passing_expected_label_record(),
        "actual_red_flag_rule_ids": [],
        "final_output": {
            **_passing_expected_label_record()["final_output"],
            "protocol_urgency": "routine",
            "source_cards": ["SAFETY-BOUNDARIES-v1"],
            "candidate_protocol_pathways": [{"card_id": "SAFETY-BOUNDARIES-v1"}],
            "missing_info_to_collect": ["repeat vital signs"],
        },
        "competence_success": False,
        "canned_fallback_used": True,
        "canned_fallback_success": True,
        "final_validation": {"passed": True},
    }

    summary = summarize_eval_records([passing, failing])

    assert summary["total_cases"] == 2
    assert summary["final_validation_successes"] == 2
    assert summary["competence_successes"] == 0
    assert summary["canned_fallback_successes"] == 2
    assert summary["expected_label_successes"] == 1
    assert summary["expected_label_failures"] == 1
    assert summary["expected_label_check_successes"]["red_flags_match"] == 1
    assert summary["expected_label_check_successes"]["min_urgency_met"] == 1
    assert summary["expected_label_check_successes"]["forbidden_behavior_absent"] == 2
    assert "missing_model_observation_cue_counts" in summary
    assert "missing_handoff_cue_counts" in summary
    assert "missing_harness_evidence_cue_counts" in summary
    assert "handoff_metric_failures" in summary
    assert "handoff_readiness_passed" in summary["handoff_metric_failures"]