workspace/tests/test_metrics.py

from __future__ import annotations

import json
import math
from pathlib import Path

import pytest

from cil.metrics import (
    MetricInputError,
    any_unsafe,
    candidate_diversity,
    collapse_rate,
    mean_nearest_distance_to_set,
    normalized_causal_action_regret,
    outcome_safety_violation,
    outcome_ptr_at_k,
    proxy_positive_tangent_coverage_at_k,
    proxy_support_distance,
    selector_regret_at_k,
    selected_unsafe,
    safety_label_coverage,
    unsafe_rate,
)
from scripts import eval_metrics


def test_outcome_ptr_refuses_proxy_inputs() -> None:
    with pytest.raises(MetricInputError, match="OutcomePTR"):
        outcome_ptr_at_k([0.5, 0.7], 0.6, candidates_evaluated=False)

    assert outcome_ptr_at_k([0.5, 0.7], 0.6, candidates_evaluated=True) == 1.0


def test_selector_regret_refuses_unevaluated_candidates() -> None:
    with pytest.raises(MetricInputError, match="SelectorRegret"):
        selector_regret_at_k([0.5, 0.7], candidates_evaluated=False)


def test_normalized_causal_action_regret() -> None:
    assert normalized_causal_action_regret(2.0, 1.25, 1.0) == pytest.approx(0.75)
    assert normalized_causal_action_regret(2.0, 0.5, 1.0) > 1.0


def test_proxy_positive_tangent_coverage_is_not_outcome_ptr() -> None:
    generated = [[0.0, 0.1], [2.0, 2.0]]
    positives = [[0.0, 0.0]]

    assert proxy_positive_tangent_coverage_at_k(
        generated,
        positives,
        threshold=0.08,
        k=2,
    ) == 1.0
    assert math.isclose(proxy_support_distance(generated, positives, k=2), math.sqrt(0.005))
    assert math.isclose(
        mean_nearest_distance_to_set(generated, positives, k=2),
        (math.sqrt(0.005) + 2.0) / 2.0,
    )
    assert candidate_diversity(generated, k=2) > 0.0
    assert collapse_rate([[1.0, 1.0], [1.0, 1.0], [2.0, 2.0]], k=3) == pytest.approx(
        2.0 / 3.0
    )


def test_safety_metrics_preserve_unknown_labels() -> None:
    outcomes = [
        {"safety_violation": False},
        {"safety_violation": "yes"},
        {"safety_violation": None},
    ]

    assert outcome_safety_violation(outcomes[0]) is False
    assert outcome_safety_violation(outcomes[1]) is True
    assert outcome_safety_violation(outcomes[2]) is None
    assert safety_label_coverage(outcomes, k=3) == pytest.approx(2.0 / 3.0)
    assert unsafe_rate(outcomes, k=3) == pytest.approx(0.5)
    assert any_unsafe(outcomes, k=3) == 1.0
    assert selected_unsafe(outcomes, selected_index=1, k=3) == 1.0
    assert unsafe_rate([{"safety_violation": None}], k=1) is None


def test_eval_metrics_measured_mode_requires_evaluated_candidates(tmp_path: Path) -> None:
    input_path = tmp_path / "rows.json"
    input_path.write_text(
        json.dumps(
            [
                {
                    "task_id": "pick",
                    "seed": 0,
                    "candidates_evaluated": False,
                    "base_utility": 0.2,
                    "generated_utilities": [0.3, 0.4],
                }
            ]
        )
    )

    with pytest.raises(MetricInputError, match="candidates_evaluated=true"):
        eval_metrics.main(["--input", str(input_path), "--out-dir", str(tmp_path / "out"), "--mode", "measured"])


def test_eval_metrics_measured_exports_success_decomposition(tmp_path: Path) -> None:
    input_path = tmp_path / "rows.json"
    input_path.write_text(
        json.dumps(
            {
                "rows": [
                    {
                        "chart_id": "c0",
                        "task_id": "pick",
                        "seed": 0,
                        "candidates_evaluated": True,
                        "base_utility": 0.3,
                        "base_success": False,
                        "base_outcome": {"success": False, "safety_violation": False},
                        "generated_utilities": [0.2, 1.1, 0.5],
                        "candidate_success": [False, True, False],
                        "candidate_outcomes": [
                            {"success": False, "safety_violation": True},
                            {"success": True, "safety_violation": False},
                            {"success": False, "safety_violation": None},
                        ],
                        "selected_index": 0,
                        "hidden_chart_utilities": [0.4, 1.5],
                        "predicted_scores": [0.9, 0.8, 0.1],
                    }
                ]
            }
        )
    )
    out_dir = tmp_path / "measured_metrics"

    assert (
        eval_metrics.main(
            [
                "--input",
                str(input_path),
                "--out-dir",
                str(out_dir),
                "--mode",
                "measured",
                "--k",
                "3",
                "--bootstrap-samples",
                "20",
            ]
        )
        == 0
    )

    metrics = json.loads((out_dir / "metrics.json").read_text())
    row = metrics["rows"][0]
    assert row["selected_success_at_3"] == 0.0
    assert row["proposal_oracle_success_at_3"] == 1.0
    assert row["hidden_chart_oracle_success_at_3"] == 1.0
    assert row["success_support_gap_at_3"] == 0.0
    assert row["success_selector_gap_at_3"] == 1.0
    assert row["ncar_to_proposal_oracle_at_3"] == pytest.approx(1.125)
    assert row["ncar_to_hidden_chart_oracle_at_3"] == pytest.approx(1.0833333333)
    assert row["support_gap_fraction_to_hidden_at_3"] == pytest.approx(0.3333333333)
    assert row["selector_gap_fraction_to_hidden_at_3"] == pytest.approx(0.75)
    assert row["proposal_oracle_utility_gain_over_base_at_3"] == pytest.approx(0.8)
    assert row["base_safety_label_known"] == 1.0
    assert row["base_unsafe_known"] == 0.0
    assert row["generated_safety_label_coverage_at_3"] == pytest.approx(2.0 / 3.0)
    assert row["generated_unsafe_rate_known_at_3"] == pytest.approx(0.5)
    assert row["any_generated_unsafe_known_at_3"] == 1.0
    assert row["selected_safety_label_known_at_3"] == 1.0
    assert row["selected_unsafe_known_at_3"] == 1.0
    assert row["proposal_oracle_safety_label_known_at_3"] == 1.0
    assert row["proposal_oracle_unsafe_known_at_3"] == 0.0


def test_eval_metrics_measured_exports_safety_coverage_without_false_zero(
    tmp_path: Path,
) -> None:
    input_path = tmp_path / "rows.json"
    input_path.write_text(
        json.dumps(
            {
                "rows": [
                    {
                        "chart_id": "c0",
                        "task_id": "pick",
                        "seed": 0,
                        "candidates_evaluated": True,
                        "base_utility": 0.3,
                        "base_success": False,
                        "base_outcome": {"success": False, "safety_violation": None},
                        "generated_utilities": [0.2, 0.4],
                        "candidate_success": [False, False],
                        "candidate_outcomes": [
                            {"success": False, "safety_violation": None},
                            {"success": False, "safety_violation": None},
                        ],
                        "selected_index": 0,
                    }
                ]
            }
        )
    )
    out_dir = tmp_path / "measured_metrics_null_safety"

    assert (
        eval_metrics.main(
            [
                "--input",
                str(input_path),
                "--out-dir",
                str(out_dir),
                "--mode",
                "measured",
                "--k",
                "2",
                "--bootstrap-samples",
                "20",
            ]
        )
        == 0
    )

    metrics = json.loads((out_dir / "metrics.json").read_text())
    row = metrics["rows"][0]
    assert row["base_safety_label_known"] == 0.0
    assert row["generated_safety_label_coverage_at_2"] == 0.0
    assert row["selected_safety_label_known_at_2"] == 0.0
    assert row["proposal_oracle_safety_label_known_at_2"] == 0.0
    assert "base_unsafe_known" not in row
    assert "generated_unsafe_rate_known_at_2" not in row
    assert "any_generated_unsafe_known_at_2" not in row
    assert "selected_unsafe_known_at_2" not in row
    assert "proposal_oracle_unsafe_known_at_2" not in row


def test_eval_metrics_omits_unstable_ncar_when_oracle_matches_base(tmp_path: Path) -> None:
    input_path = tmp_path / "rows.json"
    input_path.write_text(
        json.dumps(
            {
                "rows": [
                    {
                        "chart_id": "c0",
                        "task_id": "pick",
                        "seed": 0,
                        "candidates_evaluated": True,
                        "base_utility": 0.5,
                        "generated_utilities": [0.4, 0.5],
                        "selected_index": 0,
                    }
                ]
            }
        )
    )
    out_dir = tmp_path / "measured_metrics_unstable_ncar"

    assert (
        eval_metrics.main(
            [
                "--input",
                str(input_path),
                "--out-dir",
                str(out_dir),
                "--mode",
                "measured",
                "--k",
                "2",
                "--bootstrap-samples",
                "20",
            ]
        )
        == 0
    )

    row = json.loads((out_dir / "metrics.json").read_text())["rows"][0]
    assert "ncar_to_proposal_oracle_at_2" not in row


def test_eval_metrics_proxy_exports_json_and_latex(tmp_path: Path) -> None:
    input_path = tmp_path / "rows.json"
    input_path.write_text(
        json.dumps(
            {
                "rows": [
                    {
                        "chart_id": "c0",
                        "task_id": "pick",
                        "seed": 0,
                        "generated_tangents": [[0.0, 0.1], [1.0, 1.0]],
                        "positive_tangents": [[0.0, 0.0]],
                        "negative_tangents": [[2.0, 2.0]],
                    }
                ]
            }
        )
    )
    out_dir = tmp_path / "proxy_metrics"

    assert (
        eval_metrics.main(
            [
                "--input",
                str(input_path),
                "--out-dir",
                str(out_dir),
                "--mode",
                "proxy",
                "--k",
                "2",
                "--thresholds",
                "0.20,0.40",
                "--bootstrap-samples",
                "20",
            ]
        )
        == 0
    )
    metrics = json.loads((out_dir / "metrics.json").read_text())
    assert metrics["rows"][0]["pptc_at_2_thr_0p20"] == 1.0
    assert "proxy_support_distance_at_2" in metrics["rows"][0]
    assert "mean_positive_distance_at_2" in metrics["rows"][0]
    assert "mean_negative_distance_at_2" in metrics["rows"][0]
    assert (out_dir / "table.tex").exists()
    assert (out_dir / "report.md").exists()
    for filename in (
        "config.yaml",
        "command.txt",
        "git_hash.txt",
        "data_hash.txt",
        "split_hash.txt",
        "train.log",
        "eval.log",
        "metrics_by_task.json",
        "metrics_by_seed.json",
    ):
        assert (out_dir / filename).exists()


def test_eval_metrics_no_markdown_report_removes_stale_report(tmp_path: Path) -> None:
    input_path = tmp_path / "rows.json"
    input_path.write_text(
        json.dumps(
            {
                "split_hash": "split-demo",
                "rows": [
                    {
                        "chart_id": "c0",
                        "task_id": "pick",
                        "seed": 0,
                        "generated_tangents": [[0.0, 0.1]],
                        "positive_tangents": [[0.0, 0.0]],
                    }
                ],
            }
        )
    )
    out_dir = tmp_path / "proxy_metrics"
    (out_dir).mkdir()
    (out_dir / "report.md").write_text("stale\n")

    assert (
        eval_metrics.main(
            [
                "--input",
                str(input_path),
                "--out-dir",
                str(out_dir),
                "--mode",
                "proxy",
                "--k",
                "1",
                "--bootstrap-samples",
                "20",
                "--no-markdown-report",
            ]
        )
        == 0
    )

    assert not (out_dir / "report.md").exists()
    assert (out_dir / "config.yaml").exists()
    assert (out_dir / "command.txt").exists()
    assert (out_dir / "git_hash.txt").exists()
    assert (out_dir / "data_hash.txt").exists()
    assert (out_dir / "split_hash.txt").read_text().strip() == "split-demo"