workspace/scripts/eval_dominance_selector.py

#!/usr/bin/env python
from __future__ import annotations

import argparse
import json
import math
import os
import subprocess
import sys
from collections import defaultdict
from pathlib import Path
from typing import Any

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

import torch  # noqa: E402

from cil.metrics import (  # noqa: E402
    macro_micro_summary,
    outcome_safety_violation,
    pairwise_causal_dominance_ece,
)
from cil.models import CTTConfig, ChartEncoder, TangentNormalizer, UtilityEnergy  # noqa: E402
from scripts.eval_ctt_generated_rollout import load_chart_items  # noqa: E402

try:
    torch.set_num_threads(int(os.environ.get("DOVLA_TORCH_THREADS", "1")))
except (RuntimeError, ValueError):
    pass


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Calibrate a causal-dominance fallback rule on measured generated "
            "candidate rollouts and evaluate it on a held-out measured rollout set."
        )
    )
    parser.add_argument("--calibration-input", type=Path, required=True)
    parser.add_argument("--calibration-target-index", type=Path, required=True)
    parser.add_argument("--eval-input", type=Path, required=True)
    parser.add_argument("--eval-target-index", type=Path, required=True)
    parser.add_argument(
        "--checkpoint-template",
        default="runs/ctt_residual_full_seed{seed}/model.pt",
        help="Template used to load the train-seed utility-energy checkpoint.",
    )
    parser.add_argument(
        "--score-source",
        choices=("row", "checkpoint"),
        default="row",
        help=(
            "Use row predicted_scores from the measured rollout, or recompute "
            "candidate scores from the checkpoint utility-energy model."
        ),
    )
    parser.add_argument("--out-dir", type=Path, default=Path("runs/ctt_dominance_val_to_test"))
    parser.add_argument("--alpha", type=float, default=0.1)
    parser.add_argument(
        "--tau",
        default="auto",
        help=(
            "Dominance threshold. Use a float, or 'auto' to choose the threshold "
            "that maximizes selected success on the calibration split after "
            "conformal residual subtraction."
        ),
    )
    parser.add_argument("--k", type=int, default=8)
    parser.add_argument("--bootstrap-samples", type=int, default=1000)
    parser.add_argument(
        "--no-markdown-report",
        action="store_true",
        help="Do not write report.md; persistent prose lives in README.md.",
    )
    args = parser.parse_args(argv)

    if not 0.0 < args.alpha < 1.0:
        parser.error("--alpha must be in (0, 1)")
    if args.k <= 0:
        parser.error("--k must be positive")

    out_dir = args.out_dir
    out_dir.mkdir(parents=True, exist_ok=True)
    _write_provenance(out_dir, args)

    calibrator = _DominanceScorer(args.checkpoint_template, score_source=args.score_source)
    calibration_rows = _rows(json.loads(args.calibration_input.read_text()))
    eval_rows = _rows(json.loads(args.eval_input.read_text()))
    chart_feature_mode = calibrator.chart_feature_mode(_first_train_seed(calibration_rows + eval_rows))
    calibration_charts, calibration_index = _chart_map(
        args.calibration_target_index,
        chart_feature_mode=chart_feature_mode,
    )
    eval_charts, eval_index = _chart_map(
        args.eval_target_index,
        chart_feature_mode=chart_feature_mode,
    )

    calibration_cases = [
        _dominance_case(row, calibration_charts, scorer=calibrator, k=args.k)
        for row in calibration_rows
    ]
    residual_quantile = _conformal_quantile(
        [abs(case["measured_margin"] - case["predicted_margin"]) for case in calibration_cases],
        alpha=args.alpha,
    )
    tau = (
        _choose_tau(calibration_cases, residual_quantile=residual_quantile)
        if args.tau == "auto"
        else float(args.tau)
    )

    eval_cases = [
        _dominance_case(row, eval_charts, scorer=calibrator, k=args.k)
        for row in eval_rows
    ]
    eval_pairwise = _pairwise_calibration_summary(eval_cases)
    calibration_pairwise = _pairwise_calibration_summary(calibration_cases)
    evaluated_cases = [
        _evaluate_case(
            case,
            residual_quantile=residual_quantile,
            tau=tau,
            pairwise_calibration=eval_pairwise["rows"].get(index, {}),
        )
        for index, case in enumerate(eval_cases)
    ]
    calibration_eval_cases = [
        _evaluate_case(
            case,
            residual_quantile=residual_quantile,
            tau=tau,
            pairwise_calibration=calibration_pairwise["rows"].get(index, {}),
        )
        for index, case in enumerate(calibration_cases)
    ]
    metric_names = sorted(
        {
            key
            for row in evaluated_cases
            for key, value in row.items()
            if key not in {"chart_id", "task_id", "seed", "train_seed"}
            and isinstance(value, (int, float))
            and math.isfinite(float(value))
        }
    )
    summary = {
        name: macro_micro_summary(
            evaluated_cases,
            name,
            bootstrap_samples=args.bootstrap_samples,
            confidence=0.95,
        )
        for name in metric_names
    }
    metrics = {
        "report_type": "dominance_calibrated_selector_eval",
        "schema_version": 1,
        "k": args.k,
        "alpha": args.alpha,
        "tau": tau,
        "tau_mode": args.tau,
        "score_source": args.score_source,
        "chart_feature_mode": chart_feature_mode,
        "checkpoint_template": args.checkpoint_template,
        "residual_quantile": residual_quantile,
        "calibration_input": str(args.calibration_input),
        "eval_input": str(args.eval_input),
        "data_hash": eval_index.get("content_hash"),
        "split_hash": eval_index.get("split_hash"),
        "calibration_target_content_hash": calibration_index.get("content_hash"),
        "calibration_target_split_hash": calibration_index.get("split_hash"),
        "eval_target_content_hash": eval_index.get("content_hash"),
        "eval_target_split_hash": eval_index.get("split_hash"),
        "num_calibration_rows": len(calibration_cases),
        "num_eval_rows": len(evaluated_cases),
        "calibration_summary": _summary_with_pairwise(calibration_eval_cases, calibration_pairwise),
        "eval_summary": _summary_with_pairwise(evaluated_cases, eval_pairwise),
        "pairwise_causal_calibration": {
            "calibration": _pairwise_calibration_global(calibration_pairwise),
            "eval": _pairwise_calibration_global(eval_pairwise),
        },
        "summary": summary,
        "rows": evaluated_cases,
    }
    (out_dir / "metrics.json").write_text(json.dumps(metrics, indent=2, sort_keys=True) + "\n")
    (out_dir / "metrics_by_task.json").write_text(
        json.dumps(_group_means(evaluated_cases, "task_id", metric_names), indent=2, sort_keys=True)
        + "\n"
    )
    (out_dir / "metrics_by_seed.json").write_text(
        json.dumps(_group_means(evaluated_cases, "seed", metric_names), indent=2, sort_keys=True)
        + "\n"
    )
    (out_dir / "table.tex").write_text(_table(metrics) + "\n")
    report_path = out_dir / "report.md"
    if args.no_markdown_report:
        report_path.unlink(missing_ok=True)
    else:
        report_path.write_text(_report(metrics) + "\n")
    (out_dir / "train.log").write_text(
        "fit conformal residual quantile and tau on calibration measured rows only\n"
        f"calibration_input={args.calibration_input}\n"
        f"residual_quantile={residual_quantile:.6f}\n"
        f"tau={tau:.6f}\n"
    )
    (out_dir / "eval.log").write_text(
        "evaluated calibrated fallback rule on held-out measured rollout rows\n"
        f"eval_input={args.eval_input}\n"
        f"num_eval_rows={len(evaluated_cases)}\n"
    )
    print(json.dumps({"out_dir": str(out_dir), "tau": tau, "rows": len(evaluated_cases)}, indent=2))
    return 0


class _DominanceScorer:
    def __init__(self, checkpoint_template: str, *, score_source: str = "row") -> None:
        if score_source not in {"row", "checkpoint"}:
            raise ValueError("score_source must be 'row' or 'checkpoint'")
        self.checkpoint_template = checkpoint_template
        self.score_source = score_source
        self._models: dict[str, tuple[ChartEncoder, UtilityEnergy, TangentNormalizer, int]] = {}
        self._feature_modes: dict[str, str] = {}
        self._encoded_chart_cache: dict[tuple[str, str], torch.Tensor] = {}
        self._base_score_cache: dict[tuple[str, str], float] = {}

    def chart_feature_mode(self, seed: str) -> str:
        # Row-scored rollouts may still need the checkpoint utility model to
        # compute the base-action score. Load once so chart maps use the same
        # feature dimensionality as that checkpoint (for example base_context_obs).
        self._model(seed)
        return self._feature_modes.get(seed, "base")

    def base_score(self, row: dict[str, Any], chart: Any) -> float:
        if "base_predicted_score" in row:
            return float(row["base_predicted_score"])
        seed = str(row.get("train_seed", "0"))
        cache_key = (seed, str(chart.chart_id))
        if cache_key in self._base_score_cache:
            return self._base_score_cache[cache_key]
        _encoder, utility_energy, normalizer, tangent_dim = self._model(seed)
        z_chart = self._encoded_chart(seed, chart)
        with torch.inference_mode():
            zero = torch.zeros((1, tangent_dim), dtype=torch.float32)
            zero_norm = normalizer.transform(zero)
            score = float(utility_energy(z_chart, zero_norm).squeeze(0).item())
        self._base_score_cache[cache_key] = score
        return score

    def candidate_scores(self, row: dict[str, Any], chart: Any, *, k: int) -> list[float]:
        if self.score_source == "row":
            return [float(value) for value in row.get("predicted_scores", [])[:k]]
        seed = str(row.get("train_seed", "0"))
        encoder, utility_energy, normalizer, tangent_dim = self._model(seed)
        tangents = row.get("generated_tangents", [])[:k]
        if not tangents:
            return []
        z_chart = self._encoded_chart(seed, chart)
        with torch.inference_mode():
            xi = torch.as_tensor(tangents, dtype=torch.float32)
            if xi.ndim != 2:
                xi = xi.reshape(1, -1)
            if xi.shape[1] < tangent_dim:
                pad = torch.zeros((xi.shape[0], tangent_dim - xi.shape[1]), dtype=xi.dtype)
                xi = torch.cat([xi, pad], dim=1)
            elif xi.shape[1] > tangent_dim:
                xi = xi[:, :tangent_dim]
            xi_norm = normalizer.transform(xi)
            z = z_chart.repeat(xi_norm.shape[0], 1)
            return [float(value) for value in utility_energy(z, xi_norm).detach().cpu().tolist()]

    def _encoded_chart(self, seed: str, chart: Any) -> torch.Tensor:
        cache_key = (seed, str(chart.chart_id))
        if cache_key in self._encoded_chart_cache:
            return self._encoded_chart_cache[cache_key]
        encoder, _utility_energy, _normalizer, _tangent_dim = self._model(seed)
        with torch.inference_mode():
            feature = torch.as_tensor(chart.feature, dtype=torch.float32).unsqueeze(0)
            z_chart = encoder(feature)
        self._encoded_chart_cache[cache_key] = z_chart
        return z_chart

    def _model(self, seed: str) -> tuple[ChartEncoder, UtilityEnergy, TangentNormalizer, int]:
        if seed in self._models:
            return self._models[seed]
        checkpoint_path = Path(self.checkpoint_template.format(seed=seed))
        checkpoint = torch.load(checkpoint_path, map_location="cpu", weights_only=False)
        if "config" in checkpoint:
            config = CTTConfig(**checkpoint["config"])
            chart_feature_dim = config.chart_feature_dim
            chart_dim = config.chart_dim
            tangent_dim = config.tangent_dim
        else:
            chart_feature_dim = int(checkpoint["feature_dim"])
            chart_dim = int(checkpoint.get("chart_dim", 64))
            tangent_dim = int(checkpoint["tangent_dim"])
        self._feature_modes[seed] = str(checkpoint.get("chart_feature_mode", "base"))
        encoder = ChartEncoder(chart_feature_dim, output_dim=chart_dim)
        utility_energy = UtilityEnergy(chart_dim=chart_dim, tangent_dim=tangent_dim)
        encoder.load_state_dict(checkpoint["chart_encoder"])
        utility_energy.load_state_dict(checkpoint["utility_energy"])
        normalizer = TangentNormalizer.from_dict(checkpoint["normalizer"])
        encoder.eval()
        utility_energy.eval()
        self._models[seed] = (encoder, utility_energy, normalizer, tangent_dim)
        return self._models[seed]


def _dominance_case(
    row: dict[str, Any],
    charts: dict[str, Any],
    *,
    scorer: _DominanceScorer,
    k: int,
) -> dict[str, Any]:
    generated_utilities = [float(value) for value in row.get("generated_utilities", [])[:k]]
    candidate_success = [float(bool(value)) for value in row.get("candidate_success", [])[:k]]
    chart_id = str(row.get("chart_id", row.get("group_id", "")))
    if chart_id not in charts:
        raise KeyError(f"chart_id {chart_id!r} not found in target index")
    predicted_scores = scorer.candidate_scores(row, charts[chart_id], k=k)
    count = min(len(generated_utilities), len(predicted_scores), len(candidate_success))
    generated_utilities = generated_utilities[:count]
    predicted_scores = predicted_scores[:count]
    candidate_success = candidate_success[:count]
    if not generated_utilities or not predicted_scores:
        raise ValueError("dominance evaluation requires generated utilities and predicted scores")
    candidate_safety_labels = _candidate_safety_labels(row, count=count)
    top_index = max(range(len(predicted_scores)), key=lambda index: predicted_scores[index])
    base_score = scorer.base_score(row, charts[chart_id])
    base_utility = float(row["base_utility"])
    base_success = float(bool(row.get("base_success", False)))
    base_safety_label = _base_safety_label(row)
    selected_generated_utility = generated_utilities[top_index]
    selected_generated_success = candidate_success[top_index]
    top_safety_label = (
        candidate_safety_labels[top_index] if top_index < len(candidate_safety_labels) else None
    )
    proposal_oracle_utility = max(generated_utilities)
    proposal_oracle_success = float(any(candidate_success))
    hidden = [float(value) for value in row.get("hidden_chart_utilities", [])]
    hidden_oracle_utility = max(hidden) if hidden else math.nan
    hidden_oracle_success = float(any(value >= 1.0 for value in hidden)) if hidden else math.nan
    predicted_margin = predicted_scores[top_index] - base_score
    measured_margin = selected_generated_utility - base_utility
    return {
        "chart_id": chart_id,
        "task_id": str(row.get("task_id", "unknown")),
        "seed": str(row.get("seed", "unknown")),
        "train_seed": str(row.get("train_seed", "unknown")),
        "top_index": top_index,
        "base_predicted_score": base_score,
        "top_predicted_score": predicted_scores[top_index],
        "predicted_scores": predicted_scores,
        "predicted_margin": predicted_margin,
        "measured_margin": measured_margin,
        "base_utility": base_utility,
        "base_success": base_success,
        "base_safety_label_known": float(base_safety_label is not None),
        "base_unsafe_known": _label_to_float(base_safety_label),
        "top_generated_utility": selected_generated_utility,
        "top_generated_success": selected_generated_success,
        "top_candidate_safety_label_known": float(top_safety_label is not None),
        "top_candidate_unsafe_known": _label_to_float(top_safety_label),
        "candidate_safety_label_coverage": _safety_label_coverage(candidate_safety_labels),
        "candidate_unsafe_rate_known": _unsafe_rate_known(candidate_safety_labels),
        "generated_utilities": generated_utilities,
        "proposal_oracle_utility": proposal_oracle_utility,
        "proposal_oracle_success": proposal_oracle_success,
        "hidden_chart_oracle_utility": hidden_oracle_utility,
        "hidden_chart_oracle_success": hidden_oracle_success,
        "outcome_ptr": float(any(value > base_utility for value in generated_utilities)),
    }


def _evaluate_case(
    case: dict[str, Any],
    *,
    residual_quantile: float,
    tau: float,
    pairwise_calibration: dict[str, Any] | None = None,
) -> dict[str, Any]:
    lcb = float(case["predicted_margin"]) - float(residual_quantile)
    execute_generated = lcb > float(tau)
    selected_utility = (
        float(case["top_generated_utility"]) if execute_generated else float(case["base_utility"])
    )
    selected_success = (
        float(case["top_generated_success"]) if execute_generated else float(case["base_success"])
    )
    proposal_oracle_utility = float(case["proposal_oracle_utility"])
    proposal_oracle_success = float(case["proposal_oracle_success"])
    hidden_utility = float(case["hidden_chart_oracle_utility"])
    hidden_success = float(case["hidden_chart_oracle_success"])
    selected_safety = (
        _float_to_label(case.get("top_candidate_unsafe_known"))
        if execute_generated
        else _float_to_label(case.get("base_unsafe_known"))
    )
    output = dict(case)
    output.update(
        {
            "lcb_margin": lcb,
            "execute_generated": float(execute_generated),
            "fallback_to_base": float(not execute_generated),
            "coverage": float(execute_generated),
            "fallback_rate": float(not execute_generated),
            "selected_utility": selected_utility,
            "selected_success": selected_success,
            "selected_safety_label_known": float(selected_safety is not None),
            "selected_unsafe_known": _label_to_float(selected_safety),
            "unsafe_execution_label_known": float(selected_safety is not None),
            "unsafe_execution_known": _label_to_float(selected_safety),
            "selected_utility_gain_over_base": selected_utility - float(case["base_utility"]),
            "selected_success_gain_over_base": selected_success - float(case["base_success"]),
            "selector_regret": max(0.0, proposal_oracle_utility - selected_utility),
            "branch_car": max(0.0, proposal_oracle_utility - selected_utility),
            "success_selector_gap": max(0.0, proposal_oracle_success - selected_success),
            "support_gap": max(0.0, hidden_utility - proposal_oracle_utility)
            if math.isfinite(hidden_utility)
            else math.nan,
            "success_support_gap": max(0.0, hidden_success - proposal_oracle_success)
            if math.isfinite(hidden_success)
            else math.nan,
            "success_total_car_to_hidden": max(0.0, hidden_success - selected_success)
            if math.isfinite(hidden_success)
            else math.nan,
        }
    )
    if pairwise_calibration is not None:
        output.update(_pairwise_calibration_scalars(pairwise_calibration))
    return output


def _choose_tau(cases: list[dict[str, Any]], *, residual_quantile: float) -> float:
    candidates = sorted({float(case["predicted_margin"]) - float(residual_quantile) for case in cases})
    thresholds = [min(candidates, default=0.0) - 1.0, *candidates, max(candidates, default=0.0) + 1.0]
    best_tau = thresholds[0]
    best_key: tuple[float, float, float] | None = None
    for tau in thresholds:
        evaluated = [_evaluate_case(case, residual_quantile=residual_quantile, tau=tau) for case in cases]
        summary = _simple_summary(evaluated)
        # Maximize selected success, then selected utility, then coverage.
        key = (
            float(summary.get("selected_success", 0.0) or 0.0),
            float(summary.get("selected_utility", 0.0) or 0.0),
            float(summary.get("coverage", 0.0) or 0.0),
        )
        if best_key is None or key > best_key:
            best_key = key
            best_tau = tau
    return float(best_tau)


def _conformal_quantile(values: list[float], *, alpha: float) -> float:
    clean = sorted(float(value) for value in values if math.isfinite(float(value)))
    if not clean:
        raise ValueError("cannot calibrate dominance without residuals")
    index = min(len(clean) - 1, max(0, math.ceil((1.0 - alpha) * (len(clean) + 1)) - 1))
    return clean[index]


def _chart_map(index_path: Path, *, chart_feature_mode: str = "base") -> tuple[dict[str, Any], dict[str, Any]]:
    charts, index = load_chart_items(
        index_path,
        max_charts=None,
        require_positive=True,
        include_hidden=True,
        include_metadata=True,
        chart_feature_mode=chart_feature_mode,
    )
    return {chart.chart_id: chart for chart in charts}, index


def _first_train_seed(rows: list[dict[str, Any]]) -> str:
    for row in rows:
        if row.get("train_seed") is not None:
            return str(row.get("train_seed"))
    return "0"


def _rows(payload: Any) -> list[dict[str, Any]]:
    rows = payload.get("rows", payload) if isinstance(payload, dict) else payload
    if not isinstance(rows, list):
        raise SystemExit("input must be a JSON list or object with rows")
    return rows


def _simple_summary(rows: list[dict[str, Any]]) -> dict[str, float | None]:
    keys = [
        "base_success",
        "selected_success",
        "proposal_oracle_success",
        "hidden_chart_oracle_success",
        "selected_success_gain_over_base",
        "coverage",
        "fallback_rate",
        "base_safety_label_known",
        "base_unsafe_known",
        "candidate_safety_label_coverage",
        "candidate_unsafe_rate_known",
        "top_candidate_safety_label_known",
        "top_candidate_unsafe_known",
        "selected_safety_label_known",
        "selected_unsafe_known",
        "unsafe_execution_label_known",
        "unsafe_execution_known",
        "outcome_ptr",
        "success_support_gap",
        "success_selector_gap",
        "base_utility",
        "selected_utility",
        "proposal_oracle_utility",
        "hidden_chart_oracle_utility",
        "support_gap",
        "selector_regret",
    ]
    return {key: _mean([row.get(key) for row in rows]) for key in keys}


def _pairwise_calibration_summary(cases: list[dict[str, Any]], *, n_bins: int = 10) -> dict[str, Any]:
    bins = [
        {
            "count": 0,
            "accuracy_sum": 0.0,
            "confidence_sum": 0.0,
            "lower": index / n_bins,
            "upper": (index + 1) / n_bins,
        }
        for index in range(n_bins)
    ]
    rows: dict[int, dict[str, Any]] = {}
    total_pairs = 0
    correct_sum = 0.0
    confidence_sum = 0.0
    for index, case in enumerate(cases):
        row_metrics = pairwise_causal_dominance_ece(
            case.get("predicted_scores", []),
            case.get("generated_utilities", []),
            n_bins=n_bins,
        )
        rows[index] = row_metrics
        row_pairs = int(row_metrics.get("num_pairs") or 0)
        if row_pairs <= 0:
            continue
        total_pairs += row_pairs
        correct_sum += float(row_metrics.get("accuracy") or 0.0) * row_pairs
        confidence_sum += float(row_metrics.get("mean_confidence") or 0.0) * row_pairs
        for bin_index, row_bin in enumerate(row_metrics.get("bins", [])):
            if bin_index >= len(bins):
                break
            count = int(row_bin.get("count") or 0)
            bins[bin_index]["count"] += count
            bins[bin_index]["accuracy_sum"] += float(row_bin.get("accuracy") or 0.0) * count
            bins[bin_index]["confidence_sum"] += float(row_bin.get("confidence") or 0.0) * count

    ece = 0.0
    rendered_bins: list[dict[str, float | int]] = []
    for bucket in bins:
        count = int(bucket["count"])
        accuracy = bucket["accuracy_sum"] / count if count else 0.0
        confidence = bucket["confidence_sum"] / count if count else 0.0
        if total_pairs:
            ece += (count / total_pairs) * abs(accuracy - confidence)
        rendered_bins.append(
            {
                "lower": float(bucket["lower"]),
                "upper": float(bucket["upper"]),
                "count": count,
                "accuracy": accuracy,
                "confidence": confidence,
                "abs_gap": abs(accuracy - confidence),
            }
        )
    return {
        "n_bins": int(n_bins),
        "num_rows": len(cases),
        "ece": ece if total_pairs else math.nan,
        "num_pairs": int(total_pairs),
        "accuracy": correct_sum / total_pairs if total_pairs else math.nan,
        "mean_confidence": confidence_sum / total_pairs if total_pairs else math.nan,
        "bins": rendered_bins,
        "rows": rows,
    }


def _pairwise_calibration_scalars(calibration: dict[str, Any]) -> dict[str, float]:
    return {
        "pairwise_causal_calibration_ece": _finite_or_nan(calibration.get("ece")),
        "pairwise_causal_calibration_pairs": float(calibration.get("num_pairs") or 0),
        "pairwise_causal_calibration_accuracy": _finite_or_nan(calibration.get("accuracy")),
        "pairwise_causal_calibration_confidence": _finite_or_nan(
            calibration.get("mean_confidence")
        ),
    }


def _pairwise_calibration_global(calibration: dict[str, Any]) -> dict[str, Any]:
    return {key: value for key, value in calibration.items() if key != "rows"}


def _summary_with_pairwise(
    rows: list[dict[str, Any]],
    pairwise_calibration: dict[str, Any],
) -> dict[str, float | None]:
    summary = _simple_summary(rows)
    summary.update(_pairwise_calibration_scalars(pairwise_calibration))
    return summary


def _group_means(
    rows: list[dict[str, Any]],
    key: str,
    metric_names: list[str],
) -> dict[str, dict[str, float]]:
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in rows:
        grouped[str(row.get(key, "unknown"))].append(row)
    output: dict[str, dict[str, float]] = {}
    for group, group_rows in sorted(grouped.items()):
        payload: dict[str, float] = {}
        for metric in metric_names:
            value = _mean([row.get(metric) for row in group_rows])
            if value is not None:
                payload[metric] = value
        output[group] = payload
    return output


def _mean(values: list[Any]) -> float | None:
    clean = [
        float(value)
        for value in values
        if isinstance(value, (int, float)) and math.isfinite(float(value))
    ]
    return sum(clean) / len(clean) if clean else None


def _finite_or_nan(value: Any) -> float:
    return float(value) if isinstance(value, (int, float)) and math.isfinite(float(value)) else math.nan


def _base_safety_label(row: dict[str, Any]) -> bool | None:
    label = outcome_safety_violation(row.get("base_outcome"))
    if label is not None:
        return label
    if "base_safety_violation" in row:
        return outcome_safety_violation({"safety_violation": row.get("base_safety_violation")})
    return None


def _candidate_safety_labels(row: dict[str, Any], *, count: int) -> list[bool | None]:
    outcomes = row.get("candidate_outcomes", [])
    flags = row.get("candidate_safety_violation", [])
    labels: list[bool | None] = []
    for index in range(count):
        label = None
        if isinstance(outcomes, list) and index < len(outcomes):
            label = outcome_safety_violation(outcomes[index])
        if label is None and isinstance(flags, list) and index < len(flags):
            label = outcome_safety_violation({"safety_violation": flags[index]})
        labels.append(label)
    return labels


def _safety_label_coverage(labels: list[bool | None]) -> float:
    if not labels:
        return math.nan
    return sum(label is not None for label in labels) / len(labels)


def _unsafe_rate_known(labels: list[bool | None]) -> float:
    known = [label for label in labels if label is not None]
    if not known:
        return math.nan
    return sum(float(label) for label in known) / len(known)


def _label_to_float(label: bool | None) -> float:
    return math.nan if label is None else float(label)


def _float_to_label(value: Any) -> bool | None:
    if not isinstance(value, (int, float)) or not math.isfinite(float(value)):
        return None
    return bool(float(value))


def _table(metrics: dict[str, Any]) -> str:
    summary = metrics["eval_summary"]
    lines = [
        "% Auto-generated by scripts/eval_dominance_selector.py",
        "\\begin{tabular}{lrrrrrrrrrr}",
        "\\toprule",
        "Rows & Coverage & Fallback & Unsafe exec. & Base succ. & Selected succ. & Oracle succ. & OutcomePTR & Succ. support gap & Succ. selector gap & Cal. ECE \\\\",
        "\\midrule",
        f"{metrics['num_eval_rows']} & {_fmt(summary.get('coverage'))} & "
        f"{_fmt(summary.get('fallback_rate'))} & {_fmt(summary.get('unsafe_execution_known'))} & "
        f"{_fmt(summary.get('base_success'))} & "
        f"{_fmt(summary.get('selected_success'))} & {_fmt(summary.get('proposal_oracle_success'))} & "
        f"{_fmt(summary.get('outcome_ptr'))} & {_fmt(summary.get('success_support_gap'))} & "
        f"{_fmt(summary.get('success_selector_gap'))} & "
        f"{_fmt(summary.get('pairwise_causal_calibration_ece'))} \\\\",
        "\\bottomrule",
        "\\end{tabular}",
    ]
    return "\n".join(lines)


def _report(metrics: dict[str, Any]) -> str:
    summary = metrics["eval_summary"]
    calibration = metrics["calibration_summary"]
    lines = [
        "# Dominance-Calibrated CTT Selector",
        "",
        f"Calibration rows: `{metrics['num_calibration_rows']}`",
        f"Eval rows: `{metrics['num_eval_rows']}`",
        f"Alpha: `{metrics['alpha']}`",
        f"Residual quantile: `{metrics['residual_quantile']:.6f}`",
        f"Tau: `{metrics['tau']:.6f}` (`{metrics['tau_mode']}`)",
        "",
        "The threshold is fit on calibration rows only. Eval outcomes are used only for reporting.",
        "",
        "| Split | Coverage | Fallback | Unsafe exec. | Safety label coverage | Base success | Selected success | Proposal oracle | OutcomePTR | Success support gap | Success selector gap | Calibration ECE |",
        "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
        f"| calibration | {_fmt(calibration.get('coverage'))} | {_fmt(calibration.get('fallback_rate'))} | "
        f"{_fmt(calibration.get('unsafe_execution_known'))} | {_fmt(calibration.get('unsafe_execution_label_known'))} | "
        f"{_fmt(calibration.get('base_success'))} | {_fmt(calibration.get('selected_success'))} | "
        f"{_fmt(calibration.get('proposal_oracle_success'))} | {_fmt(calibration.get('outcome_ptr'))} | "
        f"{_fmt(calibration.get('success_support_gap'))} | {_fmt(calibration.get('success_selector_gap'))} | "
        f"{_fmt(calibration.get('pairwise_causal_calibration_ece'))} |",
        f"| eval | {_fmt(summary.get('coverage'))} | {_fmt(summary.get('fallback_rate'))} | "
        f"{_fmt(summary.get('unsafe_execution_known'))} | {_fmt(summary.get('unsafe_execution_label_known'))} | "
        f"{_fmt(summary.get('base_success'))} | {_fmt(summary.get('selected_success'))} | "
        f"{_fmt(summary.get('proposal_oracle_success'))} | {_fmt(summary.get('outcome_ptr'))} | "
        f"{_fmt(summary.get('success_support_gap'))} | {_fmt(summary.get('success_selector_gap'))} | "
        f"{_fmt(summary.get('pairwise_causal_calibration_ece'))} |",
        "",
        "This is a calibrated fallback diagnostic over already measured candidates; unsafe rates use available action-bound safety labels only.",
    ]
    return "\n".join(lines)


def _write_provenance(out_dir: Path, args: argparse.Namespace) -> None:
    (out_dir / "config.yaml").write_text(
        "\n".join(f"{key}: {value}" for key, value in sorted(vars(args).items())) + "\n"
    )
    (out_dir / "command.txt").write_text(
        "python scripts/eval_dominance_selector.py " + " ".join(sys.argv[1:]) + "\n"
    )
    (out_dir / "git_hash.txt").write_text(_run(["git", "rev-parse", "HEAD"]) + "\n")
    hashes = {
        "calibration_input": _sha256(args.calibration_input),
        "calibration_target_index": _sha256(args.calibration_target_index),
        "eval_input": _sha256(args.eval_input),
        "eval_target_index": _sha256(args.eval_target_index),
    }
    (out_dir / "data_hash.txt").write_text(json.dumps(hashes, indent=2, sort_keys=True) + "\n")
    (out_dir / "split_hash.txt").write_text(
        json.dumps(
            {
                "calibration_target_index": _index_hash(args.calibration_target_index),
                "eval_target_index": _index_hash(args.eval_target_index),
            },
            indent=2,
            sort_keys=True,
        )
        + "\n"
    )


def _index_hash(path: Path) -> dict[str, Any]:
    payload = json.loads(path.read_text())
    return {
        "split": payload.get("split"),
        "content_hash": payload.get("content_hash"),
        "split_hash": payload.get("split_hash"),
        "retrieval_index_allowed": payload.get("retrieval_index_allowed"),
    }


def _sha256(path: Path) -> str:
    import hashlib

    h = hashlib.sha256()
    h.update(path.read_bytes())
    return h.hexdigest()


def _fmt(value: Any) -> str:
    if not isinstance(value, (int, float)) or not math.isfinite(float(value)):
        return "n/a"
    return f"{float(value):.4f}"


def _run(command: list[str]) -> str:
    try:
        return subprocess.check_output(command, cwd=PROJECT_ROOT, text=True).strip()
    except (subprocess.CalledProcessError, FileNotFoundError):
        return ""


if __name__ == "__main__":
    raise SystemExit(main())