Add LCB dominance safety calibration metrics

workspace/scripts/eval_dominance_selector.py

@@ -17,7 +17,11 @@ if str(PROJECT_ROOT) not in sys.path:
 
 import torch  # noqa: E402
 
-from cil.metrics import macro_micro_summary  # 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
 
@@ -103,17 +107,29 @@ def main(argv: list[str] | None = None) -> int:
         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(
-            _dominance_case(row, eval_charts, scorer=calibrator, k=args.k),
+            case,
             residual_quantile=residual_quantile,
             tau=tau,
+            pairwise_calibration=eval_pairwise["rows"].get(index, {}),
         )
-        for row in eval_rows
+        for index, case in enumerate(eval_cases)
     ]
     calibration_eval_cases = [
-        _evaluate_case(case, residual_quantile=residual_quantile, tau=tau)
-        for case in calibration_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(
         {
@@ -155,8 +171,12 @@ def main(argv: list[str] | None = None) -> int:
         "eval_target_split_hash": eval_index.get("split_hash"),
         "num_calibration_rows": len(calibration_cases),
         "num_eval_rows": len(evaluated_cases),
-        "calibration_summary": _simple_summary(calibration_eval_cases),
-        "eval_summary": _simple_summary(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,
     }
@@ -292,14 +312,23 @@ def _dominance_case(
     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", [])]
@@ -315,12 +344,20 @@ def _dominance_case(
         "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,
@@ -329,7 +366,13 @@ def _dominance_case(
     }
 
 
-def _evaluate_case(case: dict[str, Any], *, residual_quantile: float, tau: float) -> dict[str, Any]:
+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 = (
@@ -342,6 +385,11 @@ def _evaluate_case(case: dict[str, Any], *, residual_quantile: float, tau: float
     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(
         {
@@ -352,6 +400,10 @@ def _evaluate_case(case: dict[str, Any], *, residual_quantile: float, tau: float
             "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),
@@ -368,6 +420,8 @@ def _evaluate_case(case: dict[str, Any], *, residual_quantile: float, tau: float
             else math.nan,
         }
     )
+    if pairwise_calibration is not None:
+        output.update(_pairwise_calibration_scalars(pairwise_calibration))
     return output
 
 
@@ -434,6 +488,16 @@ def _simple_summary(rows: list[dict[str, Any]]) -> dict[str, float | None]:
         "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",
@@ -447,6 +511,96 @@ def _simple_summary(rows: list[dict[str, Any]]) -> dict[str, float | None]:
     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,
@@ -475,19 +629,71 @@ def _mean(values: list[Any]) -> float | None:
     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}{lrrrrrrrr}",
+        "\\begin{tabular}{lrrrrrrrrrr}",
         "\\toprule",
-        "Rows & Coverage & Fallback & Base succ. & Selected succ. & Oracle succ. & OutcomePTR & Succ. support gap & Succ. selector gap \\\\",
+        "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('base_success'))} & "
+        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('success_selector_gap'))} & "
+        f"{_fmt(summary.get('pairwise_causal_calibration_ece'))} \\\\",
         "\\bottomrule",
         "\\end{tabular}",
     ]
@@ -508,18 +714,22 @@ def _report(metrics: dict[str, Any]) -> str:
         "",
         "The threshold is fit on calibration rows only. Eval outcomes are used only for reporting.",
         "",
-        "| Split | Coverage | Fallback | Base success | Selected success | Proposal oracle | OutcomePTR | Success support gap | Success selector gap |",
-        "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+        "| 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('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('success_support_gap'))} | {_fmt(summary.get('success_selector_gap'))} | "
+        f"{_fmt(summary.get('pairwise_causal_calibration_ece'))} |",
         "",
-        "This is a calibrated fallback diagnostic. It is not a final safety claim because unsafe-contact labels are not measured yet.",
+        "This is a calibrated fallback diagnostic over already measured candidates; unsafe rates use available action-bound safety labels only.",
     ]
     return "\n".join(lines)