release/v1/scripts/make_figures.py

#!/usr/bin/env python3
"""Create auditable, self-contained interactive figures from canonical CSVs.

Every generated HTML file embeds Plotly itself and has a sibling CSV containing
the exact derived rows shown in the figure.  A deterministic MANIFEST.json lists
both generated and unavailable panels, so absent evidence cannot be mistaken for
an empty or successful visualization.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import math
import os
import shutil
import statistics
import sys
import tempfile
from collections import Counter, defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence

import plotly.graph_objects as go
import plotly.io as pio
from plotly.subplots import make_subplots


FIGURE_BUILDER_VERSION = "1.2.0"
SUPPORTED_TABLE_SCHEMA = "1.0.0"
REQUIRED_TABLES = (
    "episodes.csv",
    "training_runs.csv",
    "training_metrics.csv",
    "proof_certificates.csv",
    "parameter_audit.csv",
    "summaries.csv",
    "artifacts.csv",
    "selections.csv",
    "environments.csv",
    "claim_summaries.csv",
    "errors.csv",
    "exclusions.csv",
)
MOUNTAINCAR_ENVIRONMENT = "MountainCarContinuous-v0"
PENDULUM_ENVIRONMENT = "DeterministicPendulum-v1"
COLORS = (
    "#0072B2",
    "#D55E00",
    "#009E73",
    "#CC79A7",
    "#E69F00",
    "#56B4E9",
    "#F0E442",
    "#000000",
)

METHOD_LABELS = {
    "analytic_two_phase": "Analytical",
    "ars": "ARS",
    "ppo": "PPO",
    "sac": "SAC",
    "ch3_ars": "CH-3-ARS",
    "ch3_ppo": "CH-3-PPO",
    "ch3_reinforce_adamw_seeded": "CH-3-REINFORCE",
    "ch3_reinforce_adamw_seeded_selected": "CH-3-REINFORCE · selected policy",
    "ch6_ars": "CH-6-ARS",
    "ch6_ars_pendulum": "CH-6-ARS candidates",
    "ch6_ars_pendulum_selected": "CH-6-ARS",
    "ars_baseline_pendulum_released": "ARS (released)",
    "ars_baseline_pendulum_released_corrected_reset": "ARS (corrected reset)",
}


class ValidationError(RuntimeError):
    """Raised when canonical evidence is invalid or internally inconsistent."""


@dataclass(frozen=True)
class FigureResult:
    key: str
    status: str
    reason: str | None = None
    html_path: str | None = None
    data_path: str | None = None
    row_count: int = 0
    details: Mapping[str, Any] | None = None


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def canonical_json(value: Any) -> str:
    return json.dumps(
        value,
        ensure_ascii=False,
        allow_nan=False,
        separators=(",", ":"),
        sort_keys=True,
    )


def _reject_json_constant(value: str) -> None:
    raise ValidationError(f"Non-standard JSON constant is forbidden: {value}")


def _unique_object(pairs: Sequence[tuple[str, Any]]) -> dict[str, Any]:
    output: dict[str, Any] = {}
    for key, value in pairs:
        if key in output:
            raise ValidationError(f"Duplicate JSON object key: {key!r}")
        output[key] = value
    return output


def load_json_object(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(
            path.read_text(encoding="utf-8"),
            object_pairs_hook=_unique_object,
            parse_constant=_reject_json_constant,
        )
    except (OSError, json.JSONDecodeError, ValidationError) as exc:
        raise ValidationError(f"Invalid JSON in {path}: {exc}") from exc
    if not isinstance(value, dict):
        raise ValidationError(f"Expected JSON object in {path}")
    return value


def require_nonempty(value: Any, label: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValidationError(f"{label} must be a non-empty string")
    return value


def parse_finite(value: str, label: str) -> float:
    if value == "":
        raise ValidationError(f"Missing required numeric value: {label}")
    try:
        number = float(value)
    except ValueError as exc:
        raise ValidationError(f"Invalid numeric value for {label}: {value!r}") from exc
    if not math.isfinite(number):
        raise ValidationError(f"Non-finite numeric value for {label}: {value!r}")
    return number


def parse_optional_int(value: str, label: str) -> int | None:
    if value == "":
        return None
    try:
        number = int(value)
    except ValueError as exc:
        raise ValidationError(f"Invalid integer value for {label}: {value!r}") from exc
    return number


def read_verified_tables(
    tables_root: Path,
) -> tuple[dict[str, Any], dict[str, list[dict[str, str]]]]:
    tables_root = tables_root.resolve()
    manifest_path = tables_root / "MANIFEST.json"
    if not manifest_path.is_file() or manifest_path.is_symlink():
        raise ValidationError(f"Missing real canonical manifest: {manifest_path}")
    manifest = load_json_object(manifest_path)
    if manifest.get("schema_version") != SUPPORTED_TABLE_SCHEMA:
        raise ValidationError(
            f"Unsupported canonical schema: {manifest.get('schema_version')!r}"
        )
    if manifest.get("status") != "success":
        raise ValidationError("Canonical table manifest is not successful")
    declared_tables = manifest.get("tables")
    if not isinstance(declared_tables, dict):
        raise ValidationError("Canonical manifest.tables must be an object")
    missing = sorted(set(REQUIRED_TABLES) - set(declared_tables))
    unexpected = sorted(set(declared_tables) - set(REQUIRED_TABLES))
    if missing or unexpected:
        raise ValidationError(
            f"Canonical table inventory mismatch; missing={missing}, unexpected={unexpected}"
        )

    tables: dict[str, list[dict[str, str]]] = {}
    for table_name in REQUIRED_TABLES:
        declaration = declared_tables[table_name]
        if not isinstance(declaration, dict):
            raise ValidationError(f"Invalid declaration for {table_name}")
        if declaration.get("path") != table_name:
            raise ValidationError(f"Non-canonical table path for {table_name}")
        expected_hash = require_nonempty(
            declaration.get("sha256"), f"{table_name}.sha256"
        )
        expected_rows = declaration.get("rows")
        expected_columns = declaration.get("columns")
        if (
            isinstance(expected_rows, bool)
            or not isinstance(expected_rows, int)
            or expected_rows < 0
        ):
            raise ValidationError(f"Invalid row count for {table_name}")
        if not isinstance(expected_columns, list) or not all(
            isinstance(column, str) and column for column in expected_columns
        ):
            raise ValidationError(f"Invalid column declaration for {table_name}")
        if len(expected_columns) != len(set(expected_columns)):
            raise ValidationError(f"Duplicate declared columns for {table_name}")
        path = tables_root / table_name
        if not path.is_file() or path.is_symlink():
            raise ValidationError(f"Missing real canonical table: {path}")
        if sha256_file(path) != expected_hash:
            raise ValidationError(f"Canonical table hash mismatch: {table_name}")
        with path.open("r", encoding="utf-8", newline="") as handle:
            reader = csv.DictReader(handle)
            if reader.fieldnames != expected_columns:
                raise ValidationError(
                    f"Canonical CSV header mismatch for {table_name}: "
                    f"{reader.fieldnames} != {expected_columns}"
                )
            rows = []
            for line_number, row in enumerate(reader, start=2):
                if None in row:
                    raise ValidationError(
                        f"Extra CSV fields in {table_name}:{line_number}"
                    )
                if any(value is None for value in row.values()):
                    raise ValidationError(
                        f"Short CSV row in {table_name}:{line_number}"
                    )
                rows.append(dict(row))
        if len(rows) != expected_rows:
            raise ValidationError(
                f"Canonical row-count mismatch for {table_name}: {len(rows)} != {expected_rows}"
            )
        tables[table_name] = rows
    return manifest, tables


def csv_scalar(value: Any) -> str | int:
    if value is None:
        return ""
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        if not math.isfinite(value):
            raise ValidationError("Cannot write a non-finite figure value")
        return format(value, ".17g")
    if isinstance(value, (dict, list)):
        return canonical_json(value)
    return str(value)


def write_rows(
    path: Path, rows: Sequence[Mapping[str, Any]], columns: Sequence[str]
) -> None:
    if not rows:
        raise ValidationError(
            f"Refusing to create an empty figure dataset: {path.name}"
        )
    unknown = sorted({key for row in rows for key in row} - set(columns))
    if unknown:
        raise ValidationError(f"Figure data has undeclared columns: {unknown}")
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(columns), lineterminator="\n")
        writer.writeheader()
        for row in rows:
            writer.writerow({column: csv_scalar(row.get(column)) for column in columns})


def style_figure(
    fig: go.Figure, title: str, height: int = 650, width: int | None = None
) -> None:
    fig.update_layout(
        template="plotly_white",
        title={
            "text": title,
            "x": 0.01,
            "xanchor": "left",
            "y": 0.985,
            "yanchor": "top",
        },
        font={"family": "Inter, ui-sans-serif, system-ui, sans-serif", "size": 14},
        hoverlabel={"font_size": 13},
        legend={
            "orientation": "h",
            "yanchor": "bottom",
            "y": 1.02,
            "x": 0,
            "font": {"size": 12},
        },
        margin={"l": 75, "r": 35, "t": 155, "b": 70},
        height=height,
        width=width,
    )


def display_method(method: str) -> str:
    """Return a compact stable label while preserving unknown method names."""

    return METHOD_LABELS.get(method, method.replace("_", "-"))


def rgba(hex_color: str, alpha: float) -> str:
    value = hex_color.lstrip("#")
    red, green, blue = (int(value[index : index + 2], 16) for index in (0, 2, 4))
    return f"rgba({red},{green},{blue},{alpha})"


def write_self_contained_html(path: Path, fig: go.Figure, div_id: str) -> None:
    html = pio.to_html(
        fig,
        include_plotlyjs=True,
        full_html=True,
        config={"displaylogo": False, "responsive": True},
        div_id=div_id,
        auto_play=False,
    )
    if "<script src=" in html.lower():
        raise ValidationError(
            f"Generated HTML unexpectedly references external JavaScript: {path}"
        )
    path.write_text(html, encoding="utf-8")


def generated_result(
    staging: Path,
    key: str,
    fig: go.Figure,
    rows: Sequence[Mapping[str, Any]],
    columns: Sequence[str],
    details: Mapping[str, Any],
) -> FigureResult:
    html_name = f"{key}.html"
    data_name = f"{key}.csv"
    write_rows(staging / data_name, rows, columns)
    write_self_contained_html(staging / html_name, fig, f"figure-{key}")
    return FigureResult(
        key=key,
        status="generated",
        html_path=html_name,
        data_path=data_name,
        row_count=len(rows),
        details=dict(details),
    )


def successful_environment_rows(
    episodes: Sequence[Mapping[str, str]], environment: str
) -> list[Mapping[str, str]]:
    return [
        row
        for row in episodes
        if row.get("status") == "success" and row.get("environment") == environment
    ]


def selected_seed_index(
    selections: Sequence[Mapping[str, str]],
    context: str,
    *,
    claim_ids: set[str] | None = None,
) -> dict[tuple[str, str], int]:
    """Return the unique declared selected seed for each batch and method."""

    output: dict[tuple[str, str], int] = {}
    for row in selections:
        if row.get("status") != "success":
            continue
        if claim_ids is not None and row.get("claim_id") not in claim_ids:
            continue
        batch_id = require_nonempty(
            row.get("batch_id"), f"{context}.selection.batch_id"
        )
        method = require_nonempty(row.get("method"), f"{context}.selection.method")
        selected_seed = parse_optional_int(
            row.get("selected_seed", ""), f"{context}.selection.selected_seed"
        )
        if selected_seed is None:
            raise ValidationError(
                f"{context}: successful selection has no selected_seed"
            )
        identity = (batch_id, method)
        if identity in output:
            raise ValidationError(
                f"{context}: multiple successful selection records for {identity}"
            )
        output[identity] = selected_seed
    return output


def build_mountaincar_return_figure(
    staging: Path, tables: Mapping[str, Sequence[Mapping[str, str]]]
) -> FigureResult:
    key = "mountaincar_return_vs_start"
    environment_rows = successful_environment_rows(
        tables["episodes.csv"], MOUNTAINCAR_ENVIRONMENT
    )
    source_rows = [
        row for row in environment_rows if row.get("start_position", "") != ""
    ]
    if not source_rows:
        return FigureResult(
            key,
            "missing",
            f"No successful {MOUNTAINCAR_ENVIRONMENT} fixed-start episode rows",
        )
    selections = selected_seed_index(
        tables["selections.csv"], key, claim_ids={"claim3", "claim4"}
    )
    groups: dict[tuple[str, str, float], list[tuple[float, str, str, int | None]]] = (
        defaultdict(list)
    )
    for row in source_rows:
        start = parse_finite(row.get("start_position", ""), f"{key}.start_position")
        value = parse_finite(row.get("return", ""), f"{key}.return")
        batch_id = require_nonempty(row.get("batch_id"), f"{key}.batch_id")
        method = require_nonempty(row.get("method"), f"{key}.method")
        seed = parse_optional_int(row.get("seed", ""), f"{key}.seed")
        groups[(batch_id, method, start)].append(
            (value, row.get("claim_id", ""), row.get("panel_id", ""), seed)
        )
    derived: list[dict[str, Any]] = []
    for (batch_id, method, start), values in sorted(groups.items()):
        returns = [item[0] for item in values]
        derived.append(
            {
                "batch_id": batch_id,
                "method": method,
                "performance_scope": "all_recorded_fixed_grid",
                "selected_seed": None,
                "start_position": start,
                "episode_count": len(returns),
                "mean_return": statistics.fmean(returns),
                "min_return": min(returns),
                "max_return": max(returns),
                "claim_ids": sorted({item[1] for item in values if item[1]}),
                "panel_ids": sorted({item[2] for item in values if item[2]}),
            }
        )

    selected_trace_status: list[dict[str, Any]] = []
    available_methods = {
        (str(row.get("batch_id")), str(row.get("method"))) for row in source_rows
    }
    for identity, selected_seed in sorted(selections.items()):
        if identity not in available_methods:
            alias_method = {
                "ch3_reinforce_adamw_seeded": "ch3_reinforce_adamw_seeded_selected"
            }.get(identity[1])
            alias_identity = (identity[0], alias_method) if alias_method else None
            if alias_identity in available_methods:
                selected_trace_status.append(
                    {
                        "batch_id": identity[0],
                        "method": identity[1],
                        "selected_policy_method": alias_method,
                        "selected_seed": selected_seed,
                        "status": "represented_by_explicit_selected_policy_method",
                    }
                )
                continue
            selected_trace_status.append(
                {
                    "batch_id": identity[0],
                    "method": identity[1],
                    "selected_seed": selected_seed,
                    "status": "missing",
                    "reason": "selection method has no fixed-start episodes under the same method name",
                }
            )
            continue
        selected_source = [
            row
            for row in source_rows
            if (row.get("batch_id"), row.get("method")) == identity
            and parse_optional_int(row.get("seed", ""), f"{key}.selected.seed")
            == selected_seed
        ]
        if not selected_source:
            raise ValidationError(
                f"{key}: selection {identity} names seed {selected_seed}, but that seed has no "
                "fixed-grid episode rows"
            )
        selected_by_start: dict[float, list[Mapping[str, str]]] = defaultdict(list)
        for row in selected_source:
            selected_by_start[
                parse_finite(row["start_position"], f"{key}.selected.start_position")
            ].append(row)
        expected_starts = {
            start
            for batch_id, method, start in groups
            if (batch_id, method) == identity
        }
        if set(selected_by_start) != expected_starts:
            raise ValidationError(
                f"{key}: selected seed {selected_seed} does not cover the method's fixed-start grid"
            )
        for start, rows in sorted(selected_by_start.items()):
            if len(rows) != 1:
                raise ValidationError(
                    f"{key}: selected seed {selected_seed} has {len(rows)} rows at start {start}; "
                    "expected exactly one"
                )
            row = rows[0]
            value = parse_finite(row["return"], f"{key}.selected.return")
            derived.append(
                {
                    "batch_id": identity[0],
                    "method": identity[1],
                    "performance_scope": "selected_policy_fixed_grid",
                    "selected_seed": selected_seed,
                    "start_position": start,
                    "episode_count": 1,
                    "mean_return": value,
                    "min_return": value,
                    "max_return": value,
                    "claim_ids": [row["claim_id"]] if row.get("claim_id") else [],
                    "panel_ids": [row["panel_id"]] if row.get("panel_id") else [],
                }
            )
        selected_trace_status.append(
            {
                "batch_id": identity[0],
                "method": identity[1],
                "selected_seed": selected_seed,
                "status": "generated",
                "start_count": len(selected_by_start),
            }
        )

    series_keys = sorted(
        {
            (
                str(row["batch_id"]),
                str(row["method"]),
                str(row["performance_scope"]),
                row["selected_seed"],
            )
            for row in derived
        },
        key=lambda item: (
            item[0],
            item[1],
            item[2],
            -1 if item[3] is None else int(item[3]),
        ),
    )
    method_colors = {
        identity: COLORS[index % len(COLORS)]
        for index, identity in enumerate(
            sorted({(str(row["batch_id"]), str(row["method"])) for row in derived})
        )
    }
    base_labels: dict[tuple[str, str, str, Any], str] = {}
    for series_key in series_keys:
        batch_id, method, performance_scope, selected_seed = series_key
        series_rows = [
            row
            for row in derived
            if (
                row["batch_id"],
                row["method"],
                row["performance_scope"],
                row["selected_seed"],
            )
            == series_key
        ]
        if performance_scope == "selected_policy_fixed_grid":
            base_labels[series_key] = (
                f"{display_method(method)} · selected seed {selected_seed}"
            )
        else:
            recorded_count = max(int(row["episode_count"]) for row in series_rows)
            base_labels[series_key] = (
                f"{display_method(method)} · {recorded_count}-seed mean"
                if recorded_count > 1
                else display_method(method)
            )
    label_counts = Counter(base_labels.values())
    series_labels = {
        series_key: (f"{label} · {series_key[0]}" if label_counts[label] > 1 else label)
        for series_key, label in base_labels.items()
    }

    fig = make_subplots(
        rows=2,
        cols=1,
        shared_xaxes=True,
        vertical_spacing=0.12,
        row_heights=[0.48, 0.52],
    )
    for series_key in series_keys:
        batch_id, method, performance_scope, selected_seed = series_key
        rows = sorted(
            [
                row
                for row in derived
                if (
                    row["batch_id"],
                    row["method"],
                    row["performance_scope"],
                    row["selected_seed"],
                )
                == series_key
            ],
            key=lambda row: float(row["start_position"]),
        )
        color = method_colors[(batch_id, method)]
        is_selected = performance_scope == "selected_policy_fixed_grid"
        label = series_labels[series_key]
        x = [row["start_position"] for row in rows]
        custom = [
            [
                row["episode_count"],
                batch_id,
                performance_scope,
                selected_seed,
                row["min_return"],
                row["max_return"],
            ]
            for row in rows
        ]
        if not is_selected:
            fig.add_trace(
                go.Scatter(
                    x=x,
                    y=[row["max_return"] for row in rows],
                    mode="lines",
                    line={"width": 0, "color": color},
                    hoverinfo="skip",
                    showlegend=False,
                    legendgroup=label,
                ),
                row=1,
                col=1,
            )
            fig.add_trace(
                go.Scatter(
                    x=x,
                    y=[row["min_return"] for row in rows],
                    mode="lines",
                    line={"width": 0, "color": color},
                    fill="tonexty",
                    fillcolor=rgba(color, 0.13),
                    hoverinfo="skip",
                    showlegend=False,
                    legendgroup=label,
                ),
                row=1,
                col=1,
            )
        for row_index in (1, 2):
            fig.add_trace(
                go.Scatter(
                    x=x,
                    y=[row["mean_return"] for row in rows],
                    mode="lines+markers",
                    name=label,
                    line={
                        "color": color,
                        "width": 3.5 if is_selected else 2.2,
                        "dash": "dash" if is_selected else "solid",
                    },
                    marker={
                        "size": (7 if is_selected else 5) if row_index == 2 else 3,
                        "symbol": "diamond" if is_selected else "circle",
                    },
                    customdata=custom,
                    hovertemplate=(
                        "start=%{x:.5f}<br>mean return=%{y:.4f}"
                        "<br>min–max=%{customdata[4]:.4f}–%{customdata[5]:.4f}"
                        "<br>episodes=%{customdata[0]}<br>batch=%{customdata[1]}"
                        "<br>scope=%{customdata[2]}<br>selected seed=%{customdata[3]}"
                        "<extra>%{fullData.name}</extra>"
                    ),
                    legendgroup=label,
                    showlegend=row_index == 2,
                ),
                row=row_index,
                col=1,
            )
    mean_values = [float(row["mean_return"]) for row in derived]
    mean_span = max(mean_values) - min(mean_values)
    zoom_pad = max(0.25, 0.06 * mean_span)
    fig.update_xaxes(showgrid=True, row=1, col=1)
    fig.update_xaxes(title="Initial position", showgrid=True, row=2, col=1)
    fig.update_yaxes(title="Return · full range", showgrid=True, row=1, col=1)
    fig.update_yaxes(
        title="Mean return · zoom",
        showgrid=True,
        range=[min(mean_values) - zoom_pad, max(mean_values) + zoom_pad],
        row=2,
        col=1,
    )
    style_figure(
        fig, "Mountain Car performance across deterministic starts", height=860
    )
    fig.update_layout(
        legend={
            "orientation": "h",
            "yanchor": "top",
            "y": -0.10,
            "x": 0,
            "font": {"size": 12},
        },
        margin={"l": 80, "r": 35, "t": 120, "b": 145},
    )
    return generated_result(
        staging,
        key,
        fig,
        derived,
        (
            "batch_id",
            "method",
            "performance_scope",
            "selected_seed",
            "start_position",
            "episode_count",
            "mean_return",
            "min_return",
            "max_return",
            "claim_ids",
            "panel_ids",
        ),
        {
            "aggregation": (
                "all-recorded-policy mean/min/max at each batch-method-start, plus an exact "
                "selected-seed trace declared by selections.csv"
            ),
            "selected_policy_traces": selected_trace_status,
            "source_rows": len(source_rows),
            "excluded_random_start_rows": len(environment_rows) - len(source_rows),
            "exclusion_rule": "rows without start_position are not part of a return-vs-start curve",
            "series": len(series_keys),
        },
    )


def build_seed_distribution_figure(
    staging: Path, tables: Mapping[str, Sequence[Mapping[str, str]]]
) -> FigureResult:
    key = "mountaincar_seed_distribution"
    source_rows = successful_environment_rows(
        tables["episodes.csv"], MOUNTAINCAR_ENVIRONMENT
    )
    seeded = [row for row in source_rows if row.get("seed", "") != ""]
    if not seeded:
        return FigureResult(
            key,
            "missing",
            "No successful Mountain Car episode rows carry an explicit seed",
        )
    selections = selected_seed_index(
        tables["selections.csv"], key, claim_ids={"claim3", "claim4"}
    )
    groups: dict[tuple[str, str, str, str, int], list[float]] = defaultdict(list)
    for row in seeded:
        seed = parse_optional_int(row.get("seed", ""), f"{key}.seed")
        assert seed is not None
        group = (
            require_nonempty(row.get("batch_id"), f"{key}.batch_id"),
            row.get("claim_id", ""),
            row.get("panel_id", ""),
            require_nonempty(row.get("method"), f"{key}.method"),
            seed,
        )
        groups[group].append(parse_finite(row.get("return", ""), f"{key}.return"))
    derived = [
        {
            "batch_id": group[0],
            "claim_id": group[1],
            "panel_id": group[2],
            "method": group[3],
            "seed": group[4],
            "episode_count": len(values),
            "mean_return": statistics.fmean(values),
            "std_population": statistics.pstdev(values),
            "min_return": min(values),
            "max_return": max(values),
        }
        for group, values in sorted(groups.items())
    ]
    for row in derived:
        row["selected"] = (
            selections.get((str(row["batch_id"]), str(row["method"]))) == row["seed"]
        )
    fig = go.Figure()
    series_keys = sorted(
        {
            (row["batch_id"], row["claim_id"], row["panel_id"], row["method"])
            for row in derived
        }
    )
    method_counts = Counter(display_method(series_key[3]) for series_key in series_keys)
    selected_legend_added = False
    for index, series_key in enumerate(series_keys):
        rows = [
            row
            for row in derived
            if (row["batch_id"], row["claim_id"], row["panel_id"], row["method"])
            == series_key
        ]
        batch_id, claim_id, panel_id, method = series_key
        method_label = display_method(method)
        panel_context = panel_id or claim_id
        if "selection" in panel_context or "random" in panel_context:
            context_label = "selection panel"
        elif "primary" in panel_context or "fixed" in panel_context:
            context_label = "fixed grid"
        else:
            context_label = claim_id or panel_context
        label = (
            f"{method_label} · {context_label}"
            if method_counts[method_label] > 1
            else method_label
        )
        color = COLORS[index % len(COLORS)]
        fig.add_trace(
            go.Violin(
                x=[label] * len(rows),
                y=[row["mean_return"] for row in rows],
                name=label,
                customdata=[[row["seed"], row["episode_count"]] for row in rows],
                box_visible=True,
                meanline_visible=True,
                points="all",
                jitter=0.18,
                pointpos=0,
                marker={"color": color, "size": 7},
                line={"color": color},
                showlegend=False,
                hovertemplate=(
                    "seed=%{customdata[0]}<br>grid mean=%{y:.4f}"
                    "<br>episodes=%{customdata[1]}<br>batch="
                    + batch_id
                    + "<br>panel="
                    + panel_context
                    + "<extra>%{fullData.name}</extra>"
                ),
            )
        )
        selected_rows = [row for row in rows if row["selected"]]
        if selected_rows:
            if len(selected_rows) != 1:
                raise ValidationError(
                    f"{key}: {series_key} has {len(selected_rows)} selected seed rows"
                )
            selected_row = selected_rows[0]
            fig.add_trace(
                go.Scatter(
                    x=[label],
                    y=[selected_row["mean_return"]],
                    mode="markers",
                    name="Selected seed",
                    marker={
                        "size": 15,
                        "symbol": "diamond",
                        "color": "#E69F00",
                        "line": {"color": "#000000", "width": 1.5},
                    },
                    customdata=[[selected_row["seed"], selected_row["episode_count"]]],
                    hovertemplate=(
                        "selected seed=%{customdata[0]}<br>grid mean=%{y:.4f}"
                        "<br>episodes=%{customdata[1]}<extra>%{fullData.name}</extra>"
                    ),
                    showlegend=not selected_legend_added,
                )
            )
            selected_legend_added = True
    fig.update_xaxes(title="Method / evaluation panel", tickangle=0)
    fig.update_yaxes(title="Per-seed mean episode return")
    style_figure(
        fig, "Mountain Car robustness across released-protocol seeds", height=700
    )
    return generated_result(
        staging,
        key,
        fig,
        derived,
        (
            "batch_id",
            "claim_id",
            "panel_id",
            "method",
            "seed",
            "episode_count",
            "mean_return",
            "std_population",
            "min_return",
            "max_return",
            "selected",
        ),
        {
            "experimental_unit": "training seed",
            "seed_statistic": "mean return over that seed's recorded evaluation episodes",
            "source_rows": len(seeded),
            "seed_rows": len(derived),
            "selected_seed_rows": sum(bool(row["selected"]) for row in derived),
        },
    )


def build_parameter_performance_figure(
    staging: Path, tables: Mapping[str, Sequence[Mapping[str, str]]]
) -> FigureResult:
    key = "parameter_performance_comparison"
    all_episodes = successful_environment_rows(
        tables["episodes.csv"], MOUNTAINCAR_ENVIRONMENT
    )
    episodes = [row for row in all_episodes if row.get("start_position", "") != ""]
    parameters = [
        row for row in tables["parameter_audit.csv"] if row.get("status") == "success"
    ]
    if not episodes:
        return FigureResult(
            key, "missing", "No successful Mountain Car performance rows"
        )
    if not parameters:
        return FigureResult(key, "missing", "No successful parameter-audit rows")
    selections = selected_seed_index(
        tables["selections.csv"], key, claim_ids={"claim3"}
    )
    performance_rows: dict[tuple[str, str], list[Mapping[str, str]]] = defaultdict(list)
    for row in episodes:
        identity = (
            require_nonempty(row.get("batch_id"), f"{key}.batch_id"),
            require_nonempty(row.get("method"), f"{key}.method"),
        )
        parse_finite(row.get("return", ""), f"{key}.return")
        performance_rows[identity].append(row)

    def scoped_performance(
        identity: tuple[str, str],
    ) -> tuple[list[float], set[str], str, int | None]:
        rows = performance_rows[identity]
        selected_seed = selections.get(identity)
        if selected_seed is not None:
            rows = [
                row
                for row in rows
                if parse_optional_int(row.get("seed", ""), f"{key}.selected_seed")
                == selected_seed
            ]
            if not rows:
                raise ValidationError(
                    f"{key}: selection {identity} names seed {selected_seed}, but no matching "
                    "fixed-grid episodes exist"
                )
            starts = [
                parse_finite(row["start_position"], f"{key}.selected.start_position")
                for row in rows
            ]
            if len(starts) != len(set(starts)):
                raise ValidationError(
                    f"{key}: selected policy {identity}, seed {selected_seed} has duplicate "
                    "fixed-grid start positions"
                )
            performance_scope = "selected_policy_fixed_grid"
        else:
            performance_scope = "method_fixed_grid"
        values = [parse_finite(row["return"], f"{key}.return") for row in rows]
        claim_ids = {row["claim_id"] for row in rows if row.get("claim_id")}
        return values, claim_ids, performance_scope, selected_seed

    derived: list[dict[str, Any]] = []
    for row in parameters:
        batch_id = require_nonempty(row.get("batch_id"), f"{key}.parameter.batch_id")
        parameter_method = require_nonempty(
            row.get("method"), f"{key}.parameter.method"
        )
        if row.get("chebyshev_parameter_count", "") != "":
            # The raw arithmetic audit is intentionally wide because it keeps
            # conflicting paper and implementation counts separate.  Only the
            # Chebyshev count has an exact matching evaluated policy.
            identity = (batch_id, "ch3_ars")
            if identity in performance_rows:
                count = parse_finite(
                    row["chebyshev_parameter_count"],
                    f"{key}.chebyshev_parameter_count",
                )
                if not count.is_integer() or count <= 0:
                    raise ValidationError(
                        f"{key}: chebyshev_parameter_count must be a positive integer"
                    )
                values, claim_ids, performance_scope, selected_seed = (
                    scoped_performance(identity)
                )
                derived.append(
                    {
                        "batch_id": batch_id,
                        "method": "ch3_ars",
                        "parameter_method": parameter_method,
                        "count_source": "degree_3_bivariate_chebyshev_formula",
                        "parameter_count": int(count),
                        "mean_return": statistics.fmean(values),
                        "episode_count": len(values),
                        "performance_scope": performance_scope,
                        "selected_seed": selected_seed,
                        "claim_ids": sorted(claim_ids),
                    }
                )
        if row.get("parameter_count", "") == "":
            continue
        performance_method = row.get("performance_method") or parameter_method
        identity = (batch_id, performance_method)
        if identity not in performance_rows:
            continue
        count = parse_finite(row["parameter_count"], f"{key}.parameter_count")
        if not count.is_integer() or count <= 0:
            raise ValidationError(f"{key}: parameter_count must be a positive integer")
        values, claim_ids, performance_scope, selected_seed = scoped_performance(
            identity
        )
        derived.append(
            {
                "batch_id": batch_id,
                "method": performance_method,
                "parameter_method": parameter_method,
                "count_source": require_nonempty(
                    row.get("count_source"), f"{key}.count_source"
                ),
                "parameter_count": int(count),
                "mean_return": statistics.fmean(values),
                "episode_count": len(values),
                "performance_scope": performance_scope,
                "selected_seed": selected_seed,
                "claim_ids": sorted(claim_ids),
            }
        )
    derived.sort(
        key=lambda row: (
            row["batch_id"],
            row["method"],
            row["parameter_count"],
            row["count_source"],
        )
    )
    if not derived:
        return FigureResult(
            key,
            "missing",
            "Parameter methods do not match any Mountain Car performance method in the same batch",
        )
    fig = go.Figure()
    method_colors = {
        method: COLORS[index % len(COLORS)]
        for index, method in enumerate(sorted({str(row["method"]) for row in derived}))
    }
    for row in derived:
        method_label = display_method(str(row["method"]))
        label = f"{method_label} · {row['count_source']}"
        fig.add_trace(
            go.Scatter(
                x=[row["parameter_count"]],
                y=[row["mean_return"]],
                mode="markers+text",
                text=[f"{method_label}<br>{int(row['parameter_count']):,} parameters"],
                textposition="top center",
                name=label,
                marker={
                    "size": 14,
                    "color": method_colors[str(row["method"])],
                    "line": {"color": "white", "width": 1.5},
                },
                customdata=[
                    [
                        row["count_source"],
                        row["episode_count"],
                        row["batch_id"],
                        row["performance_scope"],
                        row["selected_seed"],
                    ]
                ],
                hovertemplate=(
                    "parameters=%{x:,}<br>mean return=%{y:.4f}"
                    "<br>source=%{customdata[0]}<br>episodes=%{customdata[1]}"
                    "<br>batch=%{customdata[2]}<br>scope=%{customdata[3]}"
                    "<br>selected seed=%{customdata[4]}<extra>%{fullData.name}</extra>"
                ),
                showlegend=False,
            )
        )
    fig.update_xaxes(title="Trainable scalar count (log scale)", type="log")
    performance_values = [float(row["mean_return"]) for row in derived]
    performance_pad = max(
        0.5, 0.08 * (max(performance_values) - min(performance_values))
    )
    fig.update_yaxes(
        title="Mean Mountain Car return",
        range=[
            min(performance_values) - performance_pad,
            max(performance_values) + performance_pad,
        ],
    )
    style_figure(fig, "Parameter efficiency versus reproduced Mountain Car performance")
    fig.update_layout(margin={"l": 75, "r": 35, "t": 100, "b": 70})
    return generated_result(
        staging,
        key,
        fig,
        derived,
        (
            "batch_id",
            "method",
            "parameter_method",
            "count_source",
            "parameter_count",
            "mean_return",
            "episode_count",
            "performance_scope",
            "selected_seed",
            "claim_ids",
        ),
        {
            "join": "exact batch_id + (performance_method if declared, otherwise method)",
            "performance_statistic": (
                "if selections.csv declares a selected seed, mean only that seed's fixed-grid "
                "episodes; otherwise mean the method's fixed-grid episodes"
            ),
            "selection_precedence": (
                "exact batch_id + method selection is authoritative for the headline Pareto "
                "point; all-seed robustness remains in mountaincar_seed_distribution"
            ),
            "performance_filter": "fixed-start rows only; random-start selection episodes are excluded",
            "excluded_random_start_rows": len(all_episodes) - len(episodes),
            "points": len(derived),
        },
    )


def build_pendulum_figure(
    staging: Path, tables: Mapping[str, Sequence[Mapping[str, str]]]
) -> FigureResult:
    key = "pendulum_heatmap_difference"
    source_rows = successful_environment_rows(
        tables["episodes.csv"], PENDULUM_ENVIRONMENT
    )
    if not source_rows:
        return FigureResult(
            key, "missing", f"No successful {PENDULUM_ENVIRONMENT} episode rows"
        )

    unit_rows: dict[tuple[str, str, str, str, int | None], list[Mapping[str, str]]] = (
        defaultdict(list)
    )
    for row in source_rows:
        identity = (
            require_nonempty(row.get("batch_id"), f"{key}.batch_id"),
            row.get("claim_id", ""),
            row.get("panel_id", ""),
            require_nonempty(row.get("method"), f"{key}.method"),
            parse_optional_int(row.get("seed", ""), f"{key}.seed"),
        )
        parse_finite(row.get("initial_angle", ""), f"{key}.initial_angle")
        parse_finite(
            row.get("initial_angular_velocity", ""),
            f"{key}.initial_angular_velocity",
        )
        parse_finite(row.get("return", ""), f"{key}.return")
        unit_rows[identity].append(row)

    selection_rows = [
        row
        for row in tables["selections.csv"]
        if row.get("status") == "success"
        and row.get("claim_id") == "claim5"
        and row.get("method") == "ch6_ars_pendulum"
    ]
    if len(selection_rows) != 1:
        raise ValidationError(
            f"{key}: expected one authoritative Claim 5 selection, found "
            f"{len(selection_rows)}"
        )
    selection = selection_rows[0]
    batch_id = require_nonempty(selection.get("batch_id"), f"{key}.selection.batch_id")
    declared_seed = parse_optional_int(
        selection.get("selected_seed", ""), f"{key}.selection.selected_seed"
    )
    candidate_count = parse_optional_int(
        selection.get("candidate_count", ""), f"{key}.selection.candidate_count"
    )
    if declared_seed is None or candidate_count is None or candidate_count <= 0:
        raise ValidationError(
            f"{key}: selection seed/count must be present and positive"
        )
    if selection.get("selection_and_reporting_grid_reused") != "true":
        raise ValidationError(
            f"{key}: selection must declare reuse of the fixed reporting grid"
        )

    candidate_identities = sorted(
        identity
        for identity in unit_rows
        if identity[0] == batch_id
        and identity[1] == "claim5"
        and identity[2] == "pendulum_ch6_ars_seed_grid_v1"
        and identity[3] == "ch6_ars_pendulum"
        and identity[4] is not None
    )
    candidate_seeds = [int(identity[4]) for identity in candidate_identities]
    if len(candidate_identities) != candidate_count:
        raise ValidationError(
            f"{key}: selection declares {candidate_count} candidates, found "
            f"{len(candidate_identities)} unique candidate grids"
        )
    if len(candidate_seeds) != len(set(candidate_seeds)):
        raise ValidationError(f"{key}: duplicate candidate seed surfaces")

    selected_identity = (
        batch_id,
        "claim5",
        "pendulum_primary_comparison_v1",
        "ch6_ars_pendulum_selected",
        declared_seed,
    )
    author_identity = (
        batch_id,
        "claim5",
        "pendulum_primary_comparison_v1",
        "ars_baseline_pendulum_released",
        0,
    )
    corrected_identity = (
        batch_id,
        "claim5",
        "pendulum_baseline_reset_sensitivity_v1",
        "ars_baseline_pendulum_released_corrected_reset",
        0,
    )
    expected_identities = {
        *candidate_identities,
        selected_identity,
        author_identity,
        corrected_identity,
    }

    def materialize_surface(
        identity: tuple[str, str, str, str, int | None],
        *,
        expected_protocol: str | None = None,
    ) -> tuple[dict[tuple[float, float], float], str, tuple[int, int]]:
        rows = unit_rows.get(identity)
        if not rows:
            raise ValidationError(
                f"{key}: missing required Pendulum surface {identity}"
            )
        values: dict[tuple[float, float], float] = {}
        for row in rows:
            angle = parse_finite(row["initial_angle"], f"{key}.initial_angle")
            velocity = parse_finite(
                row["initial_angular_velocity"], f"{key}.initial_angular_velocity"
            )
            coordinate = (angle, velocity)
            if coordinate in values:
                raise ValidationError(
                    f"{key}: duplicate Pendulum coordinate for surface {identity}"
                )
            values[coordinate] = parse_finite(row["return"], f"{key}.return")
        protocols = {row.get("evaluation_protocol", "") for row in rows}
        if len(protocols) != 1:
            raise ValidationError(
                f"{key}: mixed protocols in surface {identity}: {protocols}"
            )
        protocol = next(iter(protocols))
        if expected_protocol is not None and protocol != expected_protocol:
            raise ValidationError(
                f"{key}: protocol mismatch for {identity}: {protocol!r} != "
                f"{expected_protocol!r}"
            )
        angles = sorted({coordinate[0] for coordinate in values})
        velocities = sorted({coordinate[1] for coordinate in values})
        expected = {(angle, velocity) for angle in angles for velocity in velocities}
        if set(values) != expected:
            missing = sorted(expected - set(values))
            raise ValidationError(
                f"{key}: incomplete rectangular grid for {identity}; "
                f"missing {len(missing)} coordinates"
            )
        return values, protocol, (len(angles), len(velocities))

    candidate_surfaces: dict[int, dict[tuple[float, float], float]] = {}
    candidate_shapes: set[tuple[int, int]] = set()
    for identity in candidate_identities:
        seed = int(identity[4])
        values, _, shape = materialize_surface(identity)
        candidate_surfaces[seed] = values
        candidate_shapes.add(shape)
    if len(candidate_shapes) != 1:
        raise ValidationError(
            f"{key}: candidate grid shapes disagree: {candidate_shapes}"
        )
    recomputed_seed = max(
        candidate_seeds,
        key=lambda seed: (statistics.fmean(candidate_surfaces[seed].values()), -seed),
    )
    if recomputed_seed != declared_seed:
        raise ValidationError(
            f"{key}: declared selected seed {declared_seed} disagrees with complete-grid "
            f"argmax {recomputed_seed}"
        )
    unexpected_identities = sorted(set(unit_rows) - expected_identities, key=repr)
    if unexpected_identities:
        raise ValidationError(
            f"{key}: unexpected Pendulum method/panel/seed surfaces: "
            f"{unexpected_identities}"
        )

    selected_values, selected_protocol, selected_shape = materialize_surface(
        selected_identity,
        expected_protocol="author_exact_polynomial_helper",
    )
    author_values, author_protocol, author_shape = materialize_surface(
        author_identity,
        expected_protocol="author_exact_first_observation_bypasses_wrappers",
    )
    corrected_values, corrected_protocol, corrected_shape = materialize_surface(
        corrected_identity,
        expected_protocol="corrected_first_observation_through_vecnormalize",
    )
    if candidate_surfaces[declared_seed] != selected_values:
        raise ValidationError(
            f"{key}: explicit selected-policy grid differs from candidate seed "
            f"{declared_seed}"
        )
    coordinate_set = set(selected_values)
    if set(author_values) != coordinate_set or set(corrected_values) != coordinate_set:
        raise ValidationError(
            f"{key}: selected, author, and corrected grids do not align"
        )
    if len({selected_shape, author_shape, corrected_shape, *candidate_shapes}) != 1:
        raise ValidationError(f"{key}: displayed and candidate grid shapes disagree")

    selected_mean = statistics.fmean(selected_values.values())
    author_mean = statistics.fmean(author_values.values())
    corrected_mean = statistics.fmean(corrected_values.values())
    corrected_delta = corrected_mean - author_mean
    performance_difference = {
        coordinate: selected_values[coordinate] - author_values[coordinate]
        for coordinate in coordinate_set
    }

    surface_rows: list[dict[str, Any]] = []
    raw_surfaces = (
        (selected_identity, selected_values, selected_protocol),
        (author_identity, author_values, author_protocol),
        (corrected_identity, corrected_values, corrected_protocol),
    )
    for identity, values, protocol in raw_surfaces:
        for (angle, velocity), value in sorted(values.items()):
            surface_rows.append(
                {
                    "panel_kind": "surface",
                    "batch_id": identity[0],
                    "claim_id": identity[1],
                    "panel_id": identity[2],
                    "method": identity[3],
                    "comparison": "",
                    "selected_seed": identity[4],
                    "evaluation_protocol": protocol,
                    "left_method": "",
                    "left_seed": "",
                    "left_protocol": "",
                    "left_value": "",
                    "right_method": "",
                    "right_seed": "",
                    "right_protocol": "",
                    "right_value": "",
                    "initial_angle": angle,
                    "initial_angular_velocity": velocity,
                    "metric": "episode_return",
                    "value": value,
                }
            )

    difference_label = "CH-6-ARS − ARS · author-exact"
    for (angle, velocity), value in sorted(performance_difference.items()):
        surface_rows.append(
            {
                "panel_kind": "difference",
                "batch_id": batch_id,
                "claim_id": "claim5",
                "panel_id": "pendulum_primary_comparison_v1",
                "method": "",
                "comparison": difference_label,
                "selected_seed": declared_seed,
                "evaluation_protocol": "paired_coordinate_difference",
                "left_method": selected_identity[3],
                "left_seed": declared_seed,
                "left_protocol": selected_protocol,
                "left_value": selected_values[(angle, velocity)],
                "right_method": author_identity[3],
                "right_seed": 0,
                "right_protocol": author_protocol,
                "right_value": author_values[(angle, velocity)],
                "initial_angle": angle,
                "initial_angular_velocity": velocity,
                "metric": "return_difference",
                "value": value,
            }
        )

    performance_delta = statistics.fmean(performance_difference.values())
    panels: tuple[tuple[str, dict[tuple[float, float], float], bool], ...] = (
        (
            f"CH-6-ARS · selected seed {declared_seed}"
            f"<br><sup>mean {selected_mean:.2f}</sup>",
            selected_values,
            False,
        ),
        (
            f"ARS · author-exact reset<br><sup>mean {author_mean:.2f}</sup>",
            author_values,
            False,
        ),
        (
            f"{difference_label}<br><sup>mean Δ {performance_delta:+.2f}</sup>",
            performance_difference,
            True,
        ),
        (
            f"ARS · corrected reset<br><sup>mean {corrected_mean:.2f}; "
            f"Δ vs exact {corrected_delta:+.2f}</sup>",
            corrected_values,
            False,
        ),
    )
    fig = make_subplots(
        rows=2,
        cols=2,
        subplot_titles=[panel[0] for panel in panels],
        horizontal_spacing=0.14,
        vertical_spacing=0.16,
    )
    surface_values = [
        value
        for _, values, is_difference in panels
        if not is_difference
        for value in values.values()
    ]
    difference_values = [
        value
        for _, values, is_difference in panels
        if is_difference
        for value in values.values()
    ]
    surface_min, surface_max = min(surface_values), max(surface_values)
    if surface_min == surface_max:
        surface_min -= 1e-12
        surface_max += 1e-12
    surface_bounds = (surface_min, surface_max)
    difference_extent = max(
        max((abs(value) for value in difference_values), default=0.0),
        1e-12,
    )
    for index, (label, values, is_difference) in enumerate(panels):
        row = index // 2 + 1
        column = index % 2 + 1
        angles = sorted({coordinate[0] for coordinate in values})
        velocities = sorted({coordinate[1] for coordinate in values})
        z = [[values[(angle, velocity)] for angle in angles] for velocity in velocities]
        protocol = (
            "paired coordinate difference"
            if is_difference
            else {
                0: selected_protocol,
                1: author_protocol,
                3: corrected_protocol,
            }[index]
        )
        customdata = [[[protocol] for _ in angles] for _ in velocities]
        trace = go.Heatmap(
            x=angles,
            y=velocities,
            z=z,
            coloraxis="coloraxis2" if is_difference else "coloraxis",
            customdata=customdata,
            hovertemplate=(
                "angle=%{x:.4f}<br>angular velocity=%{y:.4f}<br>"
                + ("return difference=%{z:.4f}" if is_difference else "return=%{z:.4f}")
                + "<br>protocol=%{customdata[0]}"
                + f"<extra>{label}</extra>"
            ),
        )
        fig.add_trace(trace, row=row, col=column)
        fig.update_xaxes(title="Initial angle", row=row, col=column)
        fig.update_yaxes(title="Initial angular velocity", row=row, col=column)
    style_figure(
        fig,
        "Pendulum initial-state returns, improvement, and reset sensitivity",
        920,
    )
    fig.update_layout(
        coloraxis={
            "colorscale": "Viridis",
            "cmin": surface_bounds[0],
            "cmax": surface_bounds[1],
            "colorbar": {
                "title": "return",
                "x": 1.02,
                "y": 0.76,
                "len": 0.34,
            },
        },
        coloraxis2={
            "colorscale": "RdBu",
            "cmin": -difference_extent,
            "cmid": 0,
            "cmax": difference_extent,
            "colorbar": {
                "title": "Δ return",
                "x": 1.02,
                "y": 0.24,
                "len": 0.34,
            },
        },
        margin={"l": 80, "r": 110, "t": 155, "b": 75},
    )
    details = {
        "surface_selection": (
            "authoritative selections.csv record; complete-grid argmax "
            "recomputed only as an integrity check"
        ),
        "declared_selected_seed": declared_seed,
        "recomputed_selected_seed": recomputed_seed,
        "candidate_count": candidate_count,
        "candidate_seeds": candidate_seeds,
        "selection_matches_recomputed_argmax": True,
        "selected_copy_matches_candidate_grid": True,
        "selection_and_reporting_grid_reused": True,
        "grid_shape": list(selected_shape),
        "selected_ch6_mean": selected_mean,
        "author_exact_ars_mean": author_mean,
        "ch6_minus_author_exact_ars_mean": performance_delta,
        "corrected_reset_ars_mean": corrected_mean,
        "corrected_minus_author_exact_mean": corrected_delta,
        "protocols": {
            "selected_ch6": selected_protocol,
            "author_exact_ars": author_protocol,
            "corrected_reset_ars": corrected_protocol,
        },
        "surface_count": 3,
        "difference_count": 1,
        "displayed_panels": [panel[0] for panel in panels],
        "candidate_surfaces_excluded_from_render": candidate_count,
        "candidate_surface_reason": (
            "all candidates remain in canonical tables; the selected candidate is "
            "identical to the explicitly named selected-policy grid"
        ),
        "source_rows": len(source_rows),
    }
    surface_rows.sort(
        key=lambda row: (
            row["panel_kind"],
            row["batch_id"],
            row["claim_id"],
            row["panel_id"],
            row["method"],
            row["comparison"],
            float(row["initial_angle"]),
            float(row["initial_angular_velocity"]),
        )
    )
    return generated_result(
        staging,
        key,
        fig,
        surface_rows,
        (
            "panel_kind",
            "batch_id",
            "claim_id",
            "panel_id",
            "method",
            "comparison",
            "selected_seed",
            "evaluation_protocol",
            "left_method",
            "left_seed",
            "left_protocol",
            "left_value",
            "right_method",
            "right_seed",
            "right_protocol",
            "right_value",
            "initial_angle",
            "initial_angular_velocity",
            "metric",
            "value",
        ),
        details,
    )


def publish_directory(staging: Path, destination: Path, replace: bool) -> None:
    if destination.is_symlink():
        raise ValidationError(f"Refusing to publish through a symlink: {destination}")
    destination = destination.resolve()
    if destination.exists() and not destination.is_dir():
        raise ValidationError(
            f"Output path exists and is not a directory: {destination}"
        )
    if destination.exists() and not replace:
        raise ValidationError(
            f"Output directory already exists: {destination}; use --replace for derived outputs"
        )
    destination.parent.mkdir(parents=True, exist_ok=True)
    if not destination.exists():
        os.replace(staging, destination)
        return
    backup = destination.parent / f".{destination.name}.old-{os.getpid()}"
    if backup.exists():
        raise ValidationError(f"Refusing to overwrite stale backup directory: {backup}")
    os.replace(destination, backup)
    try:
        os.replace(staging, destination)
    except BaseException:
        os.replace(backup, destination)
        raise
    shutil.rmtree(backup)


def result_manifest_entry(staging: Path, result: FigureResult) -> dict[str, Any]:
    entry: dict[str, Any] = {
        "status": result.status,
        "reason": result.reason,
        "details": dict(result.details or {}),
    }
    if result.status == "generated":
        assert result.html_path is not None and result.data_path is not None
        html_path = staging / result.html_path
        data_path = staging / result.data_path
        entry.update(
            {
                "html": {
                    "path": result.html_path,
                    "sha256": sha256_file(html_path),
                    "bytes": html_path.stat().st_size,
                    "plotly_javascript": "embedded",
                },
                "data": {
                    "path": result.data_path,
                    "sha256": sha256_file(data_path),
                    "bytes": data_path.stat().st_size,
                    "rows": result.row_count,
                },
            }
        )
    return entry


def make_figures(tables_root: Path, output_root: Path, replace: bool) -> dict[str, Any]:
    table_manifest, tables = read_verified_tables(tables_root)
    tables_resolved = tables_root.resolve()
    output_resolved = output_root.resolve()
    if output_resolved == tables_resolved or tables_resolved in output_resolved.parents:
        raise ValidationError(
            "Figure output must not overwrite or be nested in canonical tables"
        )
    output_root.parent.mkdir(parents=True, exist_ok=True)
    staging = Path(
        tempfile.mkdtemp(
            prefix=f".{output_root.name}.tmp-", dir=output_root.parent.resolve()
        )
    )
    builders: tuple[
        Callable[[Path, Mapping[str, Sequence[Mapping[str, str]]]], FigureResult], ...
    ] = (
        build_mountaincar_return_figure,
        build_seed_distribution_figure,
        build_parameter_performance_figure,
        build_pendulum_figure,
    )
    try:
        results = [builder(staging, tables) for builder in builders]
        manifest = {
            "schema_version": SUPPORTED_TABLE_SCHEMA,
            "builder": "make_figures.py",
            "builder_version": FIGURE_BUILDER_VERSION,
            "builder_sha256": sha256_file(Path(__file__).resolve()),
            "command": build_command(tables_root, output_root, replace),
            "status": "success",
            "experiment_id": table_manifest["experiment_id"],
            "spec_version": table_manifest["spec_version"],
            "paper_version": table_manifest["paper_version"],
            "openreview_id": table_manifest["openreview_id"],
            "source": {
                "tables_manifest_sha256": sha256_file(
                    tables_root.resolve() / "MANIFEST.json"
                ),
                "table_hashes": {
                    name: table_manifest["tables"][name]["sha256"]
                    for name in REQUIRED_TABLES
                },
            },
            "availability_contract": (
                "A missing figure is recorded explicitly and emits no placeholder HTML or data CSV. "
                "Malformed or incomplete data for a present panel is a build error."
            ),
            "figures": {
                result.key: result_manifest_entry(staging, result) for result in results
            },
        }
        (staging / "MANIFEST.json").write_text(
            json.dumps(
                manifest, ensure_ascii=False, allow_nan=False, indent=2, sort_keys=True
            )
            + "\n",
            encoding="utf-8",
        )
        publish_directory(staging, output_root, replace)
        return manifest
    except BaseException:
        if staging.exists():
            shutil.rmtree(staging)
        raise


def build_command(tables_root: Path, output_root: Path, replace: bool) -> list[str]:
    """Return the canonical, interpreter-independent command for this build."""
    command = [
        str(Path(__file__).resolve()),
        "--tables-root",
        str(tables_root.resolve()),
        "--output-root",
        str(output_root.resolve()),
    ]
    if replace:
        command.append("--replace")
    return command


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--tables-root", type=Path, required=True)
    parser.add_argument("--output-root", type=Path, required=True)
    parser.add_argument(
        "--replace",
        action="store_true",
        help="Atomically replace an existing derived figure directory.",
    )
    return parser.parse_args(argv)


def main(argv: Sequence[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        manifest = make_figures(args.tables_root, args.output_root, args.replace)
    except (OSError, ValidationError, ValueError) as exc:
        print(f"make_figures: ERROR: {exc}", file=sys.stderr)
        return 2
    generated = sum(
        entry["status"] == "generated" for entry in manifest["figures"].values()
    )
    missing = len(manifest["figures"]) - generated
    print(
        f"Generated {generated} interactive figure(s); {missing} explicitly unavailable."
    )
    print(args.output_root.resolve() / "MANIFEST.json")
    return 0


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