v2 scoring-pipeline fixes: repair immutability, structured-canonical, resume fingerprints, export guards, spend ledgers

docs/runbook.md

@@ -228,6 +228,9 @@ The runner skips complete chunks and rewrites per-model merged CSVs on resume.
 Provider transport, timeout, rate-limit, server, authentication, and
 request-configuration errors are infrastructure failures; chunks containing
 those errors remain incomplete and should be retried or rerun.
+Each evaluation CSV also has a `.spend.jsonl` call ledger. It records initial,
+failed, and repair calls separately; the supervisor uses this sidecar for its
+disk spend total and falls back to the legacy CSV total when no ledger exists.
 
 ## 4b. Batch Mode (Anthropic, OpenAI, Gemini)
 
@@ -256,6 +259,10 @@ both deliberate: latency columns are left empty (batch round-trips include
 provider queue time, which is not model latency), and cost columns are
 reconstructed at standard synchronous rates so the leaderboard basis stays
 comparable while actual spend is roughly half.
+The per-model `batches/<model>.spend.jsonl` ledger retains every initial and
+repair result and de-duplicates a resumed batch by its provider batch id and
+custom id. Completed runs mirror it next to the per-model CSV so the combined
+predictions ledger includes batch calls.
 
 ## 5. Retry Broken Full Responses
 

policybench/analysis.py

@@ -1432,7 +1432,7 @@ def model_cost_latency(
     if "elapsed_seconds" in predictions.columns:
         latency_median = (
             predictions.groupby(["model", "scenario_id"])["elapsed_seconds"]
-            .sum()
+            .sum(min_count=1)
             .groupby("model")
             .median()
         )
@@ -2323,7 +2323,7 @@ def build_dashboard_payload(
         item.update(cost_latency.get(str(row["model"]), {}))
         model_stats.append({k: v for k, v in item.items() if v is not None})
     model_stats.sort(
-        key=lambda row: (row.get("within1pct", row["score"]), row["score"]),
+        key=lambda row: (row.get("exact", row["score"]), row["score"]),
         reverse=True,
     )
 
@@ -2552,11 +2552,81 @@ def build_scenario_prompt_map(
     return prompt_map
 
 
+_UNIT_SCORE_COLUMNS = frozenset(
+    {
+        "score",
+        "exact",
+        "within_1pct",
+        "within_5pct",
+        "within_10pct",
+        "threshold_score",
+        "accuracy",
+        "coverage",
+        "bounded_score",
+        "amount_accuracy",
+        "participation_accuracy",
+        "equal_score",
+        "aggregate_score",
+    }
+)
+
+
+def _is_unit_score_column(column: str) -> bool:
+    if column in _UNIT_SCORE_COLUMNS:
+        return True
+    for prefix in ("mean_", "weighted_"):
+        if (
+            column.startswith(prefix)
+            and column.removeprefix(prefix) in _UNIT_SCORE_COLUMNS
+        ):
+            return True
+    return column.startswith(
+        ("score_run_", "within10pct_run_", "coverage_run_", "accuracy_run_")
+    )
+
+
+def _validate_analysis_export(analysis: dict[str, pd.DataFrame]) -> None:
+    for table_name in (
+        "model_summary",
+        "bounded_summary",
+        "usage_summary",
+        "run_stability",
+    ):
+        frame = analysis.get(table_name)
+        if not isinstance(frame, pd.DataFrame) or "model" not in frame.columns:
+            continue
+        duplicates = sorted(
+            frame.loc[frame["model"].duplicated(keep=False), "model"]
+            .astype(str)
+            .unique()
+        )
+        if duplicates:
+            raise ValueError(f"{table_name} has duplicate model entries: {duplicates}")
+
+    for table_name, frame in analysis.items():
+        if not isinstance(frame, pd.DataFrame):
+            continue
+        for column in frame.columns:
+            if not _is_unit_score_column(str(column)):
+                continue
+            numeric = pd.to_numeric(frame[column], errors="coerce")
+            invalid = frame[column].notna() & (
+                ~np.isfinite(numeric) | numeric.lt(0) | numeric.gt(1)
+            )
+            if invalid.any():
+                values = frame.loc[invalid, column].head(5).tolist()
+                raise ValueError(
+                    f"{table_name}.{column} has score-like values outside "
+                    f"[0, 1]: {values}"
+                )
+
+
 def export_analysis(
     analysis: dict[str, pd.DataFrame],
     output_dir: str | Path,
 ) -> dict[str, Path]:
     """Write production analysis artifacts to disk."""
+    _validate_analysis_export(analysis)
     output_path = Path(output_dir)
     output_path.mkdir(parents=True, exist_ok=True)