gpu/single-gpu/home_benjamin_hdbasin/code/hdbasin/reporting.py

from __future__ import annotations

import json
from pathlib import Path
import re

import numpy as np
import pandas as pd


def _safe_markdown_table(df: pd.DataFrame) -> str:
    try:
        return df.to_markdown(index=False)
    except ImportError:
        return "```\n" + df.to_csv(index=False) + "```"


def summarize_metrics(df: pd.DataFrame) -> pd.DataFrame:
    final = (
        df.sort_values("cumulative_cost")
        .groupby(["experiment", "method", "ablation", "seed"], as_index=False)
        .tail(1)
        .copy()
    )
    rows = []
    for keys, group in final.groupby(["experiment", "method", "ablation"]):
        experiment, method, ablation = keys
        best = pd.to_numeric(group["best_loss_so_far"], errors="coerce")
        regret = pd.to_numeric(group["regret"], errors="coerce")
        rows.append(
            {
                "experiment": experiment,
                "method": method,
                "ablation": ablation,
                "runs": int(len(group)),
                "median_final_best_loss": float(best.median()),
                "iqr_final_best_loss": float(best.quantile(0.75) - best.quantile(0.25)),
                "median_final_regret": float(regret.median()) if regret.notna().any() else np.nan,
                "median_cost": float(pd.to_numeric(group["cumulative_cost"], errors="coerce").median()),
                "median_walltime": float(pd.to_numeric(group["walltime"], errors="coerce").median()),
                "median_active_dim": float(pd.to_numeric(group["active_dim"], errors="coerce").median()) if "active_dim" in group else np.nan,
                "median_tail_energy": float(pd.to_numeric(group["tail_energy"], errors="coerce").median()) if "tail_energy" in group else np.nan,
                "accepted_basin_rate": float(pd.Series(group["accepted_basin"]).astype(str).str.lower().isin(["true", "1"]).mean()),
            }
        )
    metrics = pd.DataFrame(rows)
    target_rows = []
    for exp, exp_df in df.groupby("experiment"):
        finals = exp_df.groupby(["method", "ablation", "seed"])["best_loss_so_far"].last()
        target = float(finals.quantile(0.25))
        for keys, group in exp_df.groupby(["method", "ablation", "seed"]):
            method, ablation, seed = keys
            hit = group[pd.to_numeric(group["best_loss_so_far"], errors="coerce") <= target]
            target_rows.append(
                {
                    "experiment": exp,
                    "method": method,
                    "ablation": ablation,
                    "seed": seed,
                    "target_loss": target,
                    "time_to_target": float(hit["cumulative_cost"].iloc[0]) if not hit.empty else np.inf,
                }
            )
    ttt = pd.DataFrame(target_rows)
    if not ttt.empty:
        ttt_summary = (
            ttt.replace([np.inf], np.nan)
            .groupby(["experiment", "method", "ablation"], as_index=False)["time_to_target"]
            .median()
            .rename(columns={"time_to_target": "median_time_to_target"})
        )
        metrics = metrics.merge(ttt_summary, on=["experiment", "method", "ablation"], how="left")
    return metrics


def _binary_auc(y_true: np.ndarray, score: np.ndarray) -> float:
    y_true = np.asarray(y_true, dtype=bool)
    score = np.asarray(score, dtype=float)
    pos = score[y_true]
    neg = score[~y_true]
    if len(pos) == 0 or len(neg) == 0:
        return np.nan
    wins = 0.0
    total = 0
    for p in pos:
        wins += float(np.sum(p > neg)) + 0.5 * float(np.sum(p == neg))
        total += len(neg)
    return wins / total if total else np.nan


def _early_stopping_tables(df: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    kill_columns = [
        "experiment",
        "method",
        "ablation",
        "good_threshold_loss_top_10pct",
        "all_killed_runs",
        "all_bad_runs",
        "all_good_runs",
        "kill_precision",
        "kill_recall",
        "good_run_false_kill_rate",
    ]
    summary_columns = [
        "experiment",
        "method",
        "ablation",
        "all_killed_runs",
        "all_good_runs",
        "good_run_false_kill_rate",
        "false_kill_pass_5pct",
        "false_kill_pass_10pct",
        "commercial_false_kill_status",
    ]
    kill_rows = []
    if "killed_early" not in df:
        return pd.DataFrame(columns=kill_columns), pd.DataFrame(columns=summary_columns)
    for (exp, method, ablation), g in df.groupby(["experiment", "method", "ablation"]):
        killed_text = g["killed_early"].astype(str).str.lower()
        explicit_decision = killed_text.isin(["true", "false", "1", "0"]).any()
        if g.empty or not explicit_decision:
            continue
        losses = pd.to_numeric(g["loss"], errors="coerce")
        if losses.dropna().empty:
            continue
        threshold = float(losses.quantile(0.10))
        good = losses <= threshold
        killed = killed_text.isin(["true", "1"])
        bad = ~good
        bad_killed = bad & killed
        good_killed = good & killed
        kill_rows.append(
            {
                "experiment": exp,
                "method": method,
                "ablation": ablation,
                "good_threshold_loss_top_10pct": threshold,
                "all_killed_runs": int(killed.sum()),
                "all_bad_runs": int(bad.sum()),
                "all_good_runs": int(good.sum()),
                "kill_precision": float(bad_killed.sum() / killed.sum()) if killed.sum() else np.nan,
                "kill_recall": float(bad_killed.sum() / bad.sum()) if bad.sum() else np.nan,
                "good_run_false_kill_rate": float(good_killed.sum() / good.sum()) if good.sum() else np.nan,
            }
        )
    kill_df = pd.DataFrame(kill_rows, columns=kill_columns)
    summary_rows = []
    for row in kill_df.itertuples(index=False):
        false_kill = float(row.good_run_false_kill_rate) if pd.notna(row.good_run_false_kill_rate) else np.nan
        killed = int(row.all_killed_runs)
        good_runs = int(row.all_good_runs)
        if killed == 0:
            status = "no_kills_to_evaluate"
        elif not np.isfinite(false_kill):
            status = "insufficient_good_runs"
        elif false_kill <= 0.05:
            status = "pass_5pct"
        elif false_kill <= 0.10:
            status = "pass_10pct"
        else:
            status = "fail_gt_10pct"
        summary_rows.append(
            {
                "experiment": row.experiment,
                "method": row.method,
                "ablation": row.ablation,
                "all_killed_runs": killed,
                "all_good_runs": good_runs,
                "good_run_false_kill_rate": false_kill,
                "false_kill_pass_5pct": bool(np.isfinite(false_kill) and false_kill <= 0.05),
                "false_kill_pass_10pct": bool(np.isfinite(false_kill) and false_kill <= 0.10),
                "commercial_false_kill_status": status,
            }
        )
    summary_df = pd.DataFrame(summary_rows, columns=summary_columns)
    return kill_df, summary_df


def _ablation_interpretation(metrics: pd.DataFrame) -> pd.DataFrame:
    columns = [
        "experiment",
        "full_median_final_best_loss",
        "best_ablation",
        "best_ablation_median_final_best_loss",
        "best_ablation_relative_delta_vs_full",
        "full_rank_by_loss",
        "ablations_worse_than_full",
        "ablations_better_than_full",
        "flow_certificate_pieces_matter",
        "interpretation",
    ]
    hd = metrics[metrics["method"].eq("HD-BasinFlow")].copy()
    if hd.empty:
        return pd.DataFrame(columns=columns)
    rows = []
    for exp, exp_df in hd.groupby("experiment"):
        full = exp_df[exp_df["ablation"].eq("full")]
        if full.empty:
            continue
        exp_df = exp_df.copy()
        exp_df["median_final_best_loss"] = pd.to_numeric(exp_df["median_final_best_loss"], errors="coerce")
        exp_df = exp_df.dropna(subset=["median_final_best_loss"])
        if exp_df.empty:
            continue
        full_loss = float(full["median_final_best_loss"].iloc[0])
        order = exp_df.sort_values("median_final_best_loss").reset_index(drop=True)
        best = order.iloc[0]
        full_rank = int(order.index[order["ablation"].eq("full")][0] + 1) if order["ablation"].eq("full").any() else np.nan
        non_full = exp_df[~exp_df["ablation"].eq("full")]
        worse = int((non_full["median_final_best_loss"] > full_loss).sum())
        better = int((non_full["median_final_best_loss"] < full_loss).sum())
        denom = abs(full_loss) if full_loss != 0 else 1.0
        rel_delta = (float(best["median_final_best_loss"]) - full_loss) / denom
        pieces_matter = worse > 0
        if str(best["ablation"]) == "full":
            interpretation = "full_best"
        elif pieces_matter:
            interpretation = "mixed_ablation_result"
        else:
            interpretation = "ablations_not_worse_than_full"
        rows.append(
            {
                "experiment": exp,
                "full_median_final_best_loss": full_loss,
                "best_ablation": best["ablation"],
                "best_ablation_median_final_best_loss": float(best["median_final_best_loss"]),
                "best_ablation_relative_delta_vs_full": rel_delta,
                "full_rank_by_loss": full_rank,
                "ablations_worse_than_full": worse,
                "ablations_better_than_full": better,
                "flow_certificate_pieces_matter": pieces_matter,
                "interpretation": interpretation,
            }
        )
    return pd.DataFrame(rows, columns=columns)


def make_plots(
    df: pd.DataFrame,
    metrics: pd.DataFrame,
    figures_dir: Path,
    plot_data_dir: Path,
    pairwise: pd.DataFrame | None = None,
    certificates: pd.DataFrame | None = None,
) -> None:
    import matplotlib

    matplotlib.use("Agg")
    import matplotlib.pyplot as plt

    def save_current_figure(name: str) -> None:
        plt.savefig(figures_dir / f"{name}.png", dpi=150)
        plt.savefig(figures_dir / f"{name}.pdf")

    trend = (
        df.groupby(["experiment", "method", "ablation", "cumulative_cost"], as_index=False)["best_loss_so_far"]
        .median()
        .sort_values("cumulative_cost")
    )
    trend.to_csv(plot_data_dir / "best_loss_vs_evaluations.csv", index=False)
    for exp, exp_df in trend.groupby("experiment"):
        plt.figure(figsize=(9, 5))
        for (method, ablation), g in exp_df.groupby(["method", "ablation"]):
            label = method if ablation in ("baseline", "full") else f"{method}-{ablation}"
            if label not in [
                "Random Search",
                "Sobol Quasi-Random",
                "HD-BasinFlow",
                "HD-BasinFlow-NoFlow",
                "HD-BasinFlow-NoShell",
                "IsoBasinFlow",
                "IsoBasinFlow-NoIsoReentry",
                "IsoBasinFlow-IsoRandomDirectionsOnly",
            ]:
                continue
            plt.plot(g["cumulative_cost"], g["best_loss_so_far"], label=label, linewidth=1.8)
        plt.xlabel("evaluations")
        plt.ylabel("median best loss")
        plt.title(f"Best loss vs evaluations: {exp}")
        plt.legend(fontsize=8)
        plt.tight_layout()
        save_current_figure(f"best_loss_vs_evaluations_{exp}")
        plt.close()

    wall_df = df.copy()
    wall_df["walltime"] = pd.to_numeric(wall_df.get("walltime", 0.0), errors="coerce").fillna(0.0)
    wall_df = wall_df.sort_values(["experiment", "method", "ablation", "seed", "iteration"])
    wall_df["cumulative_walltime"] = wall_df.groupby(["experiment", "method", "ablation", "seed"])["walltime"].cumsum()
    wall_trend = (
        wall_df.groupby(["experiment", "method", "ablation", "cumulative_walltime"], as_index=False)["best_loss_so_far"]
        .median()
        .sort_values("cumulative_walltime")
    )
    wall_trend.to_csv(plot_data_dir / "best_loss_vs_walltime.csv", index=False)
    for exp, exp_df in wall_trend.groupby("experiment"):
        plt.figure(figsize=(9, 5))
        for (method, ablation), g in exp_df.groupby(["method", "ablation"]):
            label = method if ablation in ("baseline", "full") else f"{method}-{ablation}"
            if label not in [
                "Random Search",
                "Sobol Quasi-Random",
                "HD-BasinFlow",
                "HD-BasinFlow-NoFlow",
                "HD-BasinFlow-NoShell",
                "IsoBasinFlow",
                "IsoBasinFlow-NoIsoReentry",
                "IsoBasinFlow-IsoRandomDirectionsOnly",
            ]:
                continue
            plt.plot(g["cumulative_walltime"], g["best_loss_so_far"], label=label, linewidth=1.8)
        plt.xlabel("wall time (seconds)")
        plt.ylabel("median best loss")
        plt.title(f"Best loss vs wall time: {exp}")
        plt.legend(fontsize=8)
        plt.tight_layout()
        save_current_figure(f"best_loss_vs_walltime_{exp}")
        plt.close()

    regret = trend.merge(df[["experiment", "cumulative_cost", "regret"]], on=["experiment", "cumulative_cost"], how="left")
    regret.to_csv(plot_data_dir / "synthetic_regret_vs_evaluations.csv", index=False)

    ab = metrics[metrics["method"].eq("HD-BasinFlow")].copy()
    if not ab.empty:
        ab.to_csv(plot_data_dir / "ablation_summary.csv", index=False)
        _ablation_interpretation(metrics).to_csv(plot_data_dir / "ablation_interpretation.csv", index=False)
        for exp, exp_df in ab.groupby("experiment"):
            plt.figure(figsize=(10, 5))
            order = exp_df.sort_values("median_final_best_loss")
            plt.bar(order["ablation"], order["median_final_best_loss"])
            plt.xticks(rotation=45, ha="right")
            plt.ylabel("median final best loss")
            plt.title(f"Ablation comparison: {exp}")
            plt.tight_layout()
            save_current_figure(f"ablation_best_loss_{exp}")
            plt.close()
            if "median_time_to_target" in exp_df:
                ttt = exp_df.dropna(subset=["median_time_to_target"]).copy()
                if not ttt.empty:
                    plt.figure(figsize=(10, 5))
                    order = ttt.sort_values("median_time_to_target")
                    plt.bar(order["ablation"], order["median_time_to_target"])
                    plt.xticks(rotation=45, ha="right")
                    plt.ylabel("median evaluations to target")
                    plt.title(f"Ablation time-to-target: {exp}")
                    plt.tight_layout()
                    save_current_figure(f"ablation_time_to_target_{exp}")
                    plt.close()

    speed = []
    for exp, exp_df in metrics.groupby("experiment"):
        base = exp_df[(exp_df["method"] == "Random Search") & (exp_df["ablation"] == "baseline")]
        full = exp_df[(exp_df["method"] == "HD-BasinFlow") & (exp_df["ablation"] == "full")]
        if base.empty or full.empty:
            continue
        b = float(base["median_time_to_target"].iloc[0])
        h = float(full["median_time_to_target"].iloc[0])
        speed.append({"experiment": exp, "speedup_to_target": b / h if h and np.isfinite(h) else np.nan})
    speed_df = pd.DataFrame(speed)
    speed_df.to_csv(plot_data_dir / "speedup_to_target.csv", index=False)
    if not speed_df.empty:
        plt.figure(figsize=(8, 4))
        plt.bar(speed_df["experiment"], speed_df["speedup_to_target"])
        plt.axhline(1.0, color="black", linewidth=1)
        plt.xticks(rotation=45, ha="right")
        plt.ylabel("Random / HD-BasinFlow time-to-target")
        plt.title("Speedup to target")
        plt.tight_layout()
        save_current_figure("speedup_to_target")
        plt.close()

    iso_cols = {
        "experiment",
        "method",
        "ablation",
        "seed",
        "iso_rays_tested",
        "iso_reentries_found",
        "iso_candidates_descended",
        "iso_no_worse_minima_found",
        "iso_deeper_minima_found",
        "iso_hit_rate",
    }
    if iso_cols.issubset(df.columns):
        iso_final = (
            df[df["method"].eq("IsoBasinFlow")]
            .sort_values("iteration")
            .groupby(["experiment", "method", "ablation", "seed"], as_index=False)
            .tail(1)
            .copy()
        )
        if not iso_final.empty:
            for col in [
                "iso_rays_tested",
                "iso_reentries_found",
                "iso_candidates_descended",
                "iso_no_worse_minima_found",
                "iso_deeper_minima_found",
                "iso_hit_rate",
            ]:
                iso_final[col] = pd.to_numeric(iso_final[col], errors="coerce")
            iso_summary = (
                iso_final.groupby(["experiment", "ablation"], as_index=False)
                .agg(
                    iso_rays_tested=("iso_rays_tested", "sum"),
                    iso_reentries_found=("iso_reentries_found", "sum"),
                    iso_candidates_descended=("iso_candidates_descended", "sum"),
                    iso_no_worse_minima_found=("iso_no_worse_minima_found", "sum"),
                    iso_deeper_minima_found=("iso_deeper_minima_found", "sum"),
                    median_iso_hit_rate=("iso_hit_rate", "median"),
                )
            )
            iso_summary.to_csv(plot_data_dir / "iso_reentry_summary.csv", index=False)
            plot = iso_summary[iso_summary["ablation"].isin(["full", "NoIsoReentry", "IsoRandomDirectionsOnly"])].copy()
            if not plot.empty:
                plt.figure(figsize=(10, 5))
                labels = plot["experiment"].astype(str) + "\n" + plot["ablation"].astype(str)
                plt.bar(labels, plot["median_iso_hit_rate"].fillna(0.0))
                plt.ylabel("median no-worse rate")
                plt.title("Iso re-entry no-worse hit rate")
                plt.xticks(rotation=45, ha="right")
                plt.tight_layout()
                save_current_figure("iso_reentry_hit_rate")
                plt.close()
            diag_cols = {
                "experiment",
                "ablation",
                "source",
                "iso_candidate_cert_loss",
                "iso_candidate_rejected_reason",
                "iso_candidate_rank",
            }
            if diag_cols.issubset(df.columns):
                iso_diag = df[
                    df["method"].eq("IsoBasinFlow")
                    & df["source"].eq("iso_reentry")
                ].copy()
                if not iso_diag.empty:
                    iso_diag["iso_candidate_cert_loss"] = pd.to_numeric(
                        iso_diag["iso_candidate_cert_loss"], errors="coerce"
                    )
                    iso_diag["iso_candidate_rank"] = pd.to_numeric(iso_diag["iso_candidate_rank"], errors="coerce")
                    iso_diag.to_csv(plot_data_dir / "iso_candidate_diagnostics.csv", index=False)
                    rejection = (
                        iso_diag.assign(
                            iso_candidate_rejected_reason=iso_diag["iso_candidate_rejected_reason"]
                            .fillna("")
                            .replace("", "accepted_or_pending")
                        )
                        .groupby(["experiment", "ablation", "iso_candidate_rejected_reason"], as_index=False)
                        .size()
                    )
                    rejection.to_csv(plot_data_dir / "iso_candidate_rejection_summary.csv", index=False)
                    if not rejection.empty:
                        pivot = rejection.pivot_table(
                            index=["experiment", "ablation"],
                            columns="iso_candidate_rejected_reason",
                            values="size",
                            aggfunc="sum",
                            fill_value=0,
                        )
                        pivot.plot(kind="bar", stacked=True, figsize=(10, 5))
                        plt.ylabel("evaluated iso candidates")
                        plt.title("Iso candidate acceptance/rejection diagnostics")
                        plt.xticks(rotation=45, ha="right")
                        plt.tight_layout()
                        save_current_figure("iso_candidate_rejections")
                        plt.close()

    noise_rows = []
    noisy = metrics[metrics["experiment"].astype(str).str.startswith("noisy_")].copy()
    if not noisy.empty:
        for _, noisy_row in noisy.iterrows():
            noisy_exp = str(noisy_row["experiment"])
            base_exp = re.sub(r"^noisy_", "", noisy_exp)
            base_exp = re.sub(r"_sigma_[0-9.eE+-]+$", "", base_exp)
            clean = metrics[
                metrics["experiment"].astype(str).eq(base_exp)
                & metrics["method"].eq(noisy_row["method"])
                & metrics["ablation"].eq(noisy_row["ablation"])
            ]
            clean_loss = float(clean["median_final_best_loss"].iloc[0]) if not clean.empty else np.nan
            clean_iqr = float(clean["iqr_final_best_loss"].iloc[0]) if not clean.empty else np.nan
            noisy_loss = float(noisy_row["median_final_best_loss"])
            noisy_iqr = float(noisy_row["iqr_final_best_loss"])
            noise_rows.append(
                {
                    "noisy_experiment": noisy_exp,
                    "clean_experiment": base_exp,
                    "method": noisy_row["method"],
                    "ablation": noisy_row["ablation"],
                    "runs": noisy_row["runs"],
                    "clean_median_final_best_loss": clean_loss,
                    "noisy_median_final_best_loss": noisy_loss,
                    "absolute_noise_penalty": noisy_loss - clean_loss if np.isfinite(clean_loss) else np.nan,
                    "relative_noise_penalty": noisy_loss / clean_loss if np.isfinite(clean_loss) and clean_loss != 0 else np.nan,
                    "clean_iqr_final_best_loss": clean_iqr,
                    "noisy_iqr_final_best_loss": noisy_iqr,
                    "iqr_increase": noisy_iqr - clean_iqr if np.isfinite(clean_iqr) else np.nan,
                }
            )
    noise_columns = [
        "noisy_experiment",
        "clean_experiment",
        "method",
        "ablation",
        "runs",
        "clean_median_final_best_loss",
        "noisy_median_final_best_loss",
        "absolute_noise_penalty",
        "relative_noise_penalty",
        "clean_iqr_final_best_loss",
        "noisy_iqr_final_best_loss",
        "iqr_increase",
    ]
    noise_df = pd.DataFrame(noise_rows, columns=noise_columns)
    noise_df.to_csv(plot_data_dir / "noise_robustness.csv", index=False)
    if not noise_df.empty:
        plot_df = noise_df[
            (noise_df["ablation"].eq("baseline"))
            | ((noise_df["method"].eq("HD-BasinFlow")) & (noise_df["ablation"].eq("full")))
        ].copy()
        if plot_df.empty:
            plot_df = noise_df.copy()
        labels = [f"{r.method}\n{r.ablation}" for r in plot_df.itertuples()]
        plt.figure(figsize=(10, 4))
        plt.bar(labels, plot_df["absolute_noise_penalty"])
        plt.axhline(0.0, color="black", linewidth=1)
        plt.xticks(rotation=45, ha="right", fontsize=8)
        plt.ylabel("noisy - clean median final best loss")
        plt.title("Noise robustness")
        plt.tight_layout()
        save_current_figure("noise_robustness")
        plt.close()

    active_cols = ["experiment", "method", "ablation", "iteration", "active_dim", "tail_energy"]
    if "subspace_error" in df:
        active_cols.append("subspace_error")
    active = df[active_cols].copy()
    active["active_dim"] = pd.to_numeric(active["active_dim"], errors="coerce")
    active["tail_energy"] = pd.to_numeric(active["tail_energy"], errors="coerce")
    if "subspace_error" in active:
        active["subspace_error"] = pd.to_numeric(active["subspace_error"], errors="coerce")
    active.dropna(subset=["active_dim", "tail_energy"]).to_csv(plot_data_dir / "active_subspace_recovery.csv", index=False)
    active_plot = active[
        active["method"].eq("HD-BasinFlow")
        & active["ablation"].eq("full")
        & active["active_dim"].notna()
        & active["tail_energy"].notna()
    ].copy()
    if not active_plot.empty:
        summary_cols = ["active_dim", "tail_energy"]
        if "subspace_error" in active_plot and active_plot["subspace_error"].notna().any():
            summary_cols.append("subspace_error")
        active_summary = (
            active_plot.groupby(["experiment", "iteration"], as_index=False)[summary_cols]
            .median()
            .sort_values(["experiment", "iteration"])
        )
        experiments = list(active_summary["experiment"].drop_duplicates())[:8]
        if experiments:
            fig, axes = plt.subplots(len(experiments), 1, figsize=(9, max(3, 2.2 * len(experiments))), squeeze=False)
            for ax, exp in zip(axes[:, 0], experiments):
                exp_df = active_summary[active_summary["experiment"].eq(exp)]
                ax.plot(exp_df["iteration"], exp_df["active_dim"], label="estimated h", linewidth=1.5)
                ax.plot(exp_df["iteration"], exp_df["tail_energy"], label="tail energy", linewidth=1.5)
                if "subspace_error" in exp_df and exp_df["subspace_error"].notna().any():
                    ax.plot(exp_df["iteration"], exp_df["subspace_error"], label="subspace error", linewidth=1.5)
                ax.set_title(exp)
                ax.set_xlabel("iteration")
                ax.set_ylabel("median value")
                ax.legend(fontsize=7)
            plt.tight_layout()
            save_current_figure("active_subspace_recovery")
            plt.close()

    allocation_columns = ["experiment", "method", "ablation", "allocation_decision", "source", "repair_type", "count", "fraction"]
    allocation_rows = []
    if "allocation_decision" in df:
        alloc = df.copy()
        alloc["allocation_decision"] = alloc["allocation_decision"].fillna("").astype(str)
        alloc = alloc[alloc["allocation_decision"].ne("")]
        if not alloc.empty:
            for keys, g in alloc.groupby(["experiment", "method", "ablation", "allocation_decision", "source", "repair_type"], dropna=False):
                experiment, method, ablation, decision, source, repair_type = keys
                denom = len(alloc[(alloc["experiment"] == experiment) & (alloc["method"] == method) & (alloc["ablation"] == ablation)])
                allocation_rows.append(
                    {
                        "experiment": experiment,
                        "method": method,
                        "ablation": ablation,
                        "allocation_decision": decision,
                        "source": "" if pd.isna(source) else source,
                        "repair_type": "" if pd.isna(repair_type) else repair_type,
                        "count": int(len(g)),
                        "fraction": float(len(g) / denom) if denom else np.nan,
                    }
                )
    allocation_df = pd.DataFrame(allocation_rows, columns=allocation_columns)
    allocation_df.to_csv(plot_data_dir / "basin_allocation_summary.csv", index=False)
    if not allocation_df.empty:
        plot_df = allocation_df[(allocation_df["method"] == "HD-BasinFlow") & (allocation_df["ablation"] == "full")].copy()
        if not plot_df.empty:
            pivot = (
                plot_df.pivot_table(
                    index="experiment",
                    columns="allocation_decision",
                    values="fraction",
                    aggfunc="sum",
                    fill_value=0.0,
                )
                .sort_index()
            )
            ax = pivot.plot(kind="bar", stacked=True, figsize=(10, 4))
            ax.set_ylabel("fraction of HD-BasinFlow evaluations")
            ax.set_title("HD-BasinFlow basin allocation mix")
            ax.legend(fontsize=8, loc="upper left", bbox_to_anchor=(1.02, 1.0))
            plt.xticks(rotation=45, ha="right")
            plt.tight_layout()
            save_current_figure("basin_allocation_summary")
            plt.close()

    allocation_time_columns = [
        "experiment",
        "method",
        "ablation",
        "seed",
        "iteration",
        "cumulative_cost",
        "allocation_decision",
        "cumulative_count",
        "cumulative_fraction",
    ]
    allocation_time_rows = []
    if "allocation_decision" in df:
        alloc_time = df.copy()
        alloc_time["allocation_decision"] = alloc_time["allocation_decision"].fillna("").astype(str)
        alloc_time = alloc_time[alloc_time["allocation_decision"].ne("")]
        if not alloc_time.empty:
            alloc_time = alloc_time.sort_values(["experiment", "method", "ablation", "seed", "iteration"])
            for keys, g in alloc_time.groupby(["experiment", "method", "ablation", "seed"]):
                experiment, method, ablation, seed = keys
                decisions = sorted(g["allocation_decision"].unique())
                counts = {decision: 0 for decision in decisions}
                total = 0
                for row in g.itertuples(index=False):
                    decision = getattr(row, "allocation_decision")
                    counts[decision] += 1
                    total += 1
                    for tracked in decisions:
                        allocation_time_rows.append(
                            {
                                "experiment": experiment,
                                "method": method,
                                "ablation": ablation,
                                "seed": seed,
                                "iteration": getattr(row, "iteration", np.nan),
                                "cumulative_cost": getattr(row, "cumulative_cost", np.nan),
                                "allocation_decision": tracked,
                                "cumulative_count": counts[tracked],
                                "cumulative_fraction": counts[tracked] / total if total else np.nan,
                            }
                        )
    allocation_time_df = pd.DataFrame(allocation_time_rows, columns=allocation_time_columns)
    allocation_time_df.to_csv(plot_data_dir / "basin_allocation_over_time.csv", index=False)
    if not allocation_time_df.empty:
        plot_df = allocation_time_df[(allocation_time_df["method"] == "HD-BasinFlow") & (allocation_time_df["ablation"] == "full")].copy()
        if not plot_df.empty:
            plot_df["iteration"] = pd.to_numeric(plot_df["iteration"], errors="coerce")
            summary = (
                plot_df.groupby(["iteration", "allocation_decision"], as_index=False)["cumulative_fraction"]
                .median()
                .dropna(subset=["iteration"])
            )
            if not summary.empty:
                pivot = summary.pivot_table(
                    index="iteration",
                    columns="allocation_decision",
                    values="cumulative_fraction",
                    aggfunc="median",
                    fill_value=0.0,
                ).sort_index()
                plt.figure(figsize=(9, 4))
                for decision in pivot.columns:
                    plt.plot(pivot.index, pivot[decision], label=decision)
                plt.xlabel("iteration")
                plt.ylabel("median cumulative allocation fraction")
                plt.title("HD-BasinFlow basin allocation over time")
                plt.legend(fontsize=8)
                plt.tight_layout()
                save_current_figure("basin_allocation_over_time")
                plt.close()

    curve_columns = [
        "experiment",
        "method",
        "ablation",
        "seed",
        "iteration",
        "curve_type",
        "curve_step",
        "curve_loss",
        "cumulative_cost",
    ]
    curve_rows = []
    for curve_type, column in [("train", "train_curve_json"), ("validation", "val_curve_json")]:
        if column not in df:
            continue
        for row in df[df[column].fillna("").astype(str).ne("")].itertuples(index=False):
            raw = getattr(row, column)
            try:
                points = json.loads(raw)
            except Exception:
                continue
            if not isinstance(points, list):
                continue
            for idx, point in enumerate(points):
                if not isinstance(point, dict) or "loss" not in point:
                    continue
                curve_rows.append(
                    {
                        "experiment": getattr(row, "experiment"),
                        "method": getattr(row, "method"),
                        "ablation": getattr(row, "ablation"),
                        "seed": getattr(row, "seed"),
                        "iteration": getattr(row, "iteration"),
                        "curve_type": curve_type,
                        "curve_step": point.get("epoch", point.get("step", idx + 1)),
                        "curve_loss": point.get("loss"),
                        "cumulative_cost": getattr(row, "cumulative_cost", np.nan),
                    }
                )
    curve_df = pd.DataFrame(curve_rows, columns=curve_columns)
    curve_df["curve_step"] = pd.to_numeric(curve_df["curve_step"], errors="coerce") if not curve_df.empty else []
    curve_df["curve_loss"] = pd.to_numeric(curve_df["curve_loss"], errors="coerce") if not curve_df.empty else []
    curve_df.to_csv(plot_data_dir / "validation_curves.csv", index=False)
    if not curve_df.empty:
        val = curve_df[curve_df["curve_type"].eq("validation")].dropna(subset=["curve_step", "curve_loss"])
        keep_methods = {"Random Search", "Sobol Quasi-Random", "Optuna TPE", "ASHA Successive Halving", "HD-BasinFlow"}
        val = val[val["method"].isin(keep_methods)]
        if not val.empty:
            summary = (
                val.groupby(["experiment", "method", "ablation", "curve_step"], as_index=False)["curve_loss"]
                .median()
                .sort_values("curve_step")
            )
            experiments = list(summary["experiment"].drop_duplicates())[:6]
            if experiments:
                fig, axes = plt.subplots(len(experiments), 1, figsize=(9, max(3, 2.5 * len(experiments))), squeeze=False)
                for ax, exp in zip(axes[:, 0], experiments):
                    exp_df = summary[summary["experiment"].eq(exp)]
                    for (method, ablation), g in exp_df.groupby(["method", "ablation"]):
                        if method == "HD-BasinFlow" and ablation != "full":
                            continue
                        label = method if ablation in ("baseline", "full") else f"{method}-{ablation}"
                        ax.plot(g["curve_step"], g["curve_loss"], marker="o", linewidth=1.5, label=label)
                    ax.set_title(exp)
                    ax.set_xlabel("curve step")
                    ax.set_ylabel("median validation loss")
                    ax.legend(fontsize=7)
                plt.tight_layout()
                save_current_figure("validation_curves")
                plt.close()

    reduction = []
    for exp, exp_df in metrics.groupby("experiment"):
        full = exp_df[(exp_df["method"] == "HD-BasinFlow") & (exp_df["ablation"] == "full")]
        if full.empty:
            continue
        full_t = float(full["median_time_to_target"].iloc[0]) if "median_time_to_target" in full else np.nan
        for _, row in exp_df[exp_df["ablation"].eq("baseline")].iterrows():
            base_t = float(row.get("median_time_to_target", np.nan))
            reduction.append(
                {
                    "experiment": exp,
                    "baseline": row["method"],
                    "cost_reduction": 1.0 - full_t / base_t if np.isfinite(full_t) and np.isfinite(base_t) and base_t > 0 else np.nan,
                }
            )
    reduction_df = pd.DataFrame(reduction)
    reduction_df.to_csv(plot_data_dir / "cost_reduction.csv", index=False)
    if not reduction_df.empty:
        plt.figure(figsize=(10, 4))
        labels = [f"{r.experiment}\nvs {r.baseline}" for r in reduction_df.itertuples()]
        plt.bar(labels, reduction_df["cost_reduction"])
        plt.axhline(0.0, color="black", linewidth=1)
        plt.xticks(rotation=45, ha="right", fontsize=8)
        plt.ylabel("cost reduction to target")
        plt.title("Cost reduction versus baselines")
        plt.tight_layout()
        save_current_figure("cost_reduction")
        plt.close()

    gpu_rows = []
    if "walltime" in df:
        work = df.copy()
        work["walltime"] = pd.to_numeric(work["walltime"], errors="coerce").fillna(0.0)
        work["best_loss_so_far"] = pd.to_numeric(work["best_loss_so_far"], errors="coerce")
        work = work.sort_values(["experiment", "method", "ablation", "seed", "iteration"])
        work["cumulative_walltime"] = work.groupby(["experiment", "method", "ablation", "seed"])["walltime"].cumsum()
        for exp, exp_df in work.groupby("experiment"):
            finals = exp_df.groupby(["method", "ablation", "seed"])["best_loss_so_far"].last()
            if finals.dropna().empty:
                continue
            target = float(finals.quantile(0.25))
            time_rows = []
            for keys, group in exp_df.groupby(["method", "ablation", "seed"]):
                method, ablation, seed = keys
                hit = group[group["best_loss_so_far"] <= target]
                if hit.empty:
                    continue
                time_rows.append(
                    {
                        "experiment": exp,
                        "method": method,
                        "ablation": ablation,
                        "seed": seed,
                        "target_loss": target,
                        "walltime_to_target": float(hit["cumulative_walltime"].iloc[0]),
                    }
                )
            if not time_rows:
                continue
            time_df = pd.DataFrame(time_rows)
            med = time_df.groupby(["experiment", "method", "ablation"], as_index=False)["walltime_to_target"].median()
            full = med[(med["method"] == "HD-BasinFlow") & (med["ablation"] == "full")]
            if full.empty:
                continue
            full_t = float(full["walltime_to_target"].iloc[0])
            for _, row in med[med["ablation"].eq("baseline")].iterrows():
                baseline_t = float(row["walltime_to_target"])
                gpu_rows.append(
                    {
                        "experiment": exp,
                        "baseline": row["method"],
                        "target_loss": target,
                        "baseline_walltime_to_target": baseline_t,
                        "hdbasinflow_walltime_to_target": full_t,
                        "gpu_time_saved_fraction": 1.0 - full_t / baseline_t if baseline_t > 0 else np.nan,
                    }
                )
    gpu_columns = [
        "experiment",
        "baseline",
        "target_loss",
        "baseline_walltime_to_target",
        "hdbasinflow_walltime_to_target",
        "gpu_time_saved_fraction",
    ]
    gpu_df = pd.DataFrame(gpu_rows, columns=gpu_columns)
    gpu_df.to_csv(plot_data_dir / "gpu_time_savings.csv", index=False)
    if not gpu_df.empty:
        plt.figure(figsize=(10, 4))
        labels = [f"{r.experiment}\nvs {r.baseline}" for r in gpu_df.itertuples()]
        plt.bar(labels, gpu_df["gpu_time_saved_fraction"])
        plt.axhline(0.0, color="black", linewidth=1)
        plt.xticks(rotation=45, ha="right", fontsize=8)
        plt.ylabel("wall-clock time saved to target")
        plt.title("GPU-time savings proxy")
        plt.tight_layout()
        save_current_figure("gpu_time_savings")
        plt.close()

    if pairwise is not None and not pairwise.empty:
        rows = []
        for exp, g in pairwise.groupby("experiment"):
            score = pd.to_numeric(g["edge_score"], errors="coerce").to_numpy()
            y = g["segment_improved"].astype(bool).to_numpy()
            order = np.argsort(score)[::-1]
            top_k = max(1, int(np.ceil(0.1 * len(order))))
            rows.append(
                {
                    "experiment": exp,
                    "pairs": len(g),
                    "auc": _binary_auc(y, score),
                    "precision_at_top_10pct": float(np.mean(y[order[:top_k]])),
                }
            )
        pairwise_summary = pd.DataFrame(rows)
        pairwise_summary.to_csv(plot_data_dir / "pairwise_score_auc_precision.csv", index=False)
        plt.figure(figsize=(8, 4))
        plt.bar(pairwise_summary["experiment"], pairwise_summary["auc"])
        plt.axhline(0.5, color="black", linewidth=1)
        plt.xticks(rotation=45, ha="right")
        plt.ylabel("AUC")
        plt.title("Pairwise score predicts midpoint improvement")
        plt.tight_layout()
        save_current_figure("pairwise_score_auc")
        plt.close()

    if certificates is not None and not certificates.empty:
        cert = certificates.copy()
        cert["accepted"] = cert["accepted"].astype(bool)
        cert["local_improved"] = cert["local_improved"].astype(bool)
        rows = []
        for exp, g in cert.groupby("experiment"):
            accepted = g["accepted"]
            improved = g["local_improved"]
            tp = int((accepted & improved).sum())
            fp = int((accepted & ~improved).sum())
            fn = int((~accepted & improved).sum())
            rows.append(
                {
                    "experiment": exp,
                    "basins": len(g),
                    "accepted": int(accepted.sum()),
                    "precision": tp / (tp + fp) if (tp + fp) else np.nan,
                    "recall": tp / (tp + fn) if (tp + fn) else np.nan,
                    "saddle_false_acceptance": int(g["saddle_false_acceptance"].sum()),
                }
            )
        cert_summary = pd.DataFrame(rows)
        cert_summary.to_csv(plot_data_dir / "certificate_precision_recall.csv", index=False)
        plt.figure(figsize=(8, 4))
        x = np.arange(len(cert_summary))
        plt.bar(x - 0.18, cert_summary["precision"], width=0.36, label="precision")
        plt.bar(x + 0.18, cert_summary["recall"], width=0.36, label="recall")
        plt.xticks(x, cert_summary["experiment"], rotation=45, ha="right")
        plt.ylim(0, 1)
        plt.ylabel("score")
        plt.title("Certificate precision and recall")
        plt.legend()
        plt.tight_layout()
        save_current_figure("certificate_precision_recall")
        plt.close()

    kill_df, false_kill_summary = _early_stopping_tables(df)
    kill_df.to_csv(plot_data_dir / "early_stopping_kill_metrics.csv", index=False)
    false_kill_summary.to_csv(plot_data_dir / "false_kill_threshold_summary.csv", index=False)
    if not kill_df.empty:
        plot_df = kill_df[kill_df["method"].isin(["ASHA Successive Halving", "HD-BasinFlow"])].copy()
        if plot_df.empty:
            plot_df = kill_df.copy()
        plot_df = plot_df[
            (plot_df["method"].eq("ASHA Successive Halving") & plot_df["ablation"].eq("baseline"))
            | (plot_df["method"].eq("HD-BasinFlow") & plot_df["ablation"].eq("full"))
        ]
        if plot_df.empty:
            plot_df = kill_df.copy()
        plt.figure(figsize=(10, 4))
        labels = [f"{r.experiment}\n{r.method}" for r in plot_df.itertuples()]
        x = np.arange(len(plot_df))
        plt.bar(x - 0.18, plot_df["kill_precision"], width=0.36, label="precision")
        plt.bar(x + 0.18, plot_df["good_run_false_kill_rate"], width=0.36, label="false kill")
        plt.xticks(x, labels, rotation=45, ha="right", fontsize=8)
        plt.ylim(0, 1)
        plt.ylabel("rate")
        plt.title("Early stopping kill metrics")
        plt.legend()
        plt.tight_layout()
        save_current_figure("early_stopping_kill_metrics")
        plt.close()
    if not false_kill_summary.empty:
        plot_df = false_kill_summary[
            (false_kill_summary["method"].eq("ASHA Successive Halving") & false_kill_summary["ablation"].eq("baseline"))
            | (false_kill_summary["method"].eq("HD-BasinFlow") & false_kill_summary["ablation"].eq("full"))
        ].copy()
        if plot_df.empty:
            plot_df = false_kill_summary.copy()
        plot_df = plot_df.dropna(subset=["good_run_false_kill_rate"])
        if not plot_df.empty:
            plt.figure(figsize=(10, 4))
            labels = [f"{r.experiment}\n{r.method}" for r in plot_df.itertuples()]
            plt.bar(labels, plot_df["good_run_false_kill_rate"])
            plt.axhline(0.05, color="green", linewidth=1, linestyle="--", label="5% pass")
            plt.axhline(0.10, color="orange", linewidth=1, linestyle="--", label="10% pass")
            plt.xticks(rotation=45, ha="right", fontsize=8)
            plt.ylim(0, max(0.15, float(plot_df["good_run_false_kill_rate"].max()) * 1.15))
            plt.ylabel("good-run false-kill rate")
            plt.title("False-kill threshold summary")
            plt.legend()
            plt.tight_layout()
            save_current_figure("false_kill_threshold_summary")
            plt.close()

    scaling = metrics[metrics["experiment"].astype(str).str.startswith("dimension_scaling_")].copy()
    if not scaling.empty:
        parsed = scaling["experiment"].str.extract(r"dimension_scaling_d(?P<ambient_dim>\d+)_h(?P<true_active_dim>\d+)")
        scaling["ambient_dim"] = pd.to_numeric(parsed["ambient_dim"], errors="coerce")
        scaling["true_active_dim"] = pd.to_numeric(parsed["true_active_dim"], errors="coerce")
        scaling.to_csv(plot_data_dir / "dimension_scaling_summary.csv", index=False)
        if "subspace_error" in df:
            err = df[df["experiment"].astype(str).str.startswith("dimension_scaling_")].copy()
            err["subspace_error"] = pd.to_numeric(err["subspace_error"], errors="coerce") if "subspace_error" in err else np.nan
            err = err.dropna(subset=["subspace_error"])
            if not err.empty:
                err_summary = (
                    err.groupby(["experiment", "method", "ablation"], as_index=False)["subspace_error"]
                    .median()
                    .rename(columns={"subspace_error": "median_subspace_error"})
                )
                scaling = scaling.merge(err_summary, on=["experiment", "method", "ablation"], how="left")
                scaling.to_csv(plot_data_dir / "dimension_scaling_summary.csv", index=False)
        full = scaling[(scaling["method"] == "HD-BasinFlow") & (scaling["ablation"] == "full")]
        if not full.empty:
            plt.figure(figsize=(8, 4))
            for h, g in full.groupby("true_active_dim"):
                g = g.sort_values("ambient_dim")
                plt.plot(g["ambient_dim"], g["median_time_to_target"], marker="o", label=f"h={int(h)}")
            plt.xlabel("ambient dimension d")
            plt.ylabel("median evaluations to target")
            plt.title("Dimension scaling")
            plt.legend()
            plt.tight_layout()
            save_current_figure("dimension_scaling_evaluations_to_target")
            plt.close()


def build_report(
    path: Path,
    evaluations: pd.DataFrame | None,
    metrics: pd.DataFrame,
    real_status: pd.DataFrame | None,
    pairwise: pd.DataFrame | None = None,
    certificates: pd.DataFrame | None = None,
) -> None:
    lines = [
        "# HD-BasinFlow Experiment Report",
        "",
        "This report is generated from actual run artifacts. Missing or unavailable workloads are reported explicitly.",
        "",
    ]
    if evaluations is not None and not evaluations.empty:
        lines.extend(
            [
                "## What Ran",
                "",
                f"- Evaluation rows: {len(evaluations):,}",
                f"- Experiments: {', '.join(sorted(evaluations['experiment'].unique()))}",
                f"- Methods: {', '.join(sorted(evaluations['method'].unique()))}",
                "",
                "## Summary Metrics",
                "",
                _safe_markdown_table(metrics.sort_values(["experiment", "method", "ablation"])),
                "",
                "## What Won",
                "",
            ]
        )
        for exp, exp_df in metrics.groupby("experiment"):
            order = exp_df.sort_values("median_final_best_loss")
            winner = order.iloc[0]
            lines.append(
                f"- `{exp}`: best median final loss was `{winner['median_final_best_loss']:.6g}` from `{winner['method']}` / `{winner['ablation']}`."
            )
        sft_metrics = metrics[metrics["experiment"].astype(str).str.startswith("llm_lora_sft_")].copy()
        if not sft_metrics.empty:
            lines.extend(["", "## LLM LoRA SFT Verdict", ""])
            for exp, exp_df in sft_metrics.groupby("experiment"):
                order = exp_df.sort_values("median_final_best_loss")
                winner = order.iloc[0]
                full_iso = exp_df[(exp_df["method"].eq("IsoBasinFlow")) & (exp_df["ablation"].eq("full"))]
                no_iso = exp_df[(exp_df["method"].eq("IsoBasinFlow")) & (exp_df["ablation"].eq("NoIsoReentry"))]
                hd = exp_df[(exp_df["method"].eq("HD-BasinFlow")) & (exp_df["ablation"].eq("full"))]
                notes = [
                    f"winner `{winner['method']}` / `{winner['ablation']}` at `{float(winner['median_final_best_loss']):.6g}`"
                ]
                if not full_iso.empty and not hd.empty:
                    delta = float(full_iso["median_final_best_loss"].iloc[0]) - float(hd["median_final_best_loss"].iloc[0])
                    notes.append(
                        f"full IsoBasinFlow {'beat' if delta < 0 else 'trailed'} HD-BasinFlow by `{abs(delta):.6g}` loss"
                    )
                if not no_iso.empty and not full_iso.empty:
                    delta = float(no_iso["median_final_best_loss"].iloc[0]) - float(full_iso["median_final_best_loss"].iloc[0])
                    notes.append(
                        f"NoIsoReentry {'beat' if delta < 0 else 'trailed'} full IsoBasinFlow by `{abs(delta):.6g}` loss"
                    )
                lines.append(f"- `{exp}`: " + "; ".join(notes) + ".")
            lines.append(
                "- Interpretation: SFT is now a real benchmark in this suite; current evidence supports Iso-family competitiveness, but not a blanket claim that full IsoBasinFlow wins every SFT workload."
            )
        iso_metrics = metrics[metrics["method"].eq("IsoBasinFlow")].copy()
        if not iso_metrics.empty:
            lines.extend(["", "## IsoBasinFlow Re-Entry Verdict", ""])
            for exp, exp_df in iso_metrics.groupby("experiment"):
                full = exp_df[exp_df["ablation"].eq("full")]
                no_iso = exp_df[exp_df["ablation"].eq("NoIsoReentry")]
                random_dirs = exp_df[exp_df["ablation"].eq("IsoRandomDirectionsOnly")]
                if full.empty:
                    continue
                full_loss = float(full["median_final_best_loss"].iloc[0])
                notes = [f"`full` median final loss `{full_loss:.6g}`"]
                if not no_iso.empty:
                    no_iso_loss = float(no_iso["median_final_best_loss"].iloc[0])
                    notes.append(
                        f"NoIsoReentry `{no_iso_loss:.6g}` ({'iso helped' if full_loss < no_iso_loss else 'iso did not improve loss'})"
                    )
                if not random_dirs.empty:
                    random_loss = float(random_dirs["median_final_best_loss"].iloc[0])
                    notes.append(
                        f"IsoRandomDirectionsOnly `{random_loss:.6g}` ({'active rays helped' if full_loss < random_loss else 'active rays not proven'})"
                    )
                lines.append(f"- `{exp}`: " + "; ".join(notes) + ".")
            if "iso_candidate_accepted" in evaluations.columns:
                accepted = evaluations[evaluations["method"].eq("IsoBasinFlow")]["iso_candidate_accepted"].astype(str).str.lower().isin(["true", "1"]).sum()
                reentries = evaluations[evaluations["method"].eq("IsoBasinFlow")]["iso_reentry_found"].astype(str).str.lower().isin(["true", "1"]).sum()
                lines.append(f"- Iso rows with re-entry candidates: `{int(reentries)}`; accepted no-worse candidates: `{int(accepted)}`.")
        lines.extend(["", "## HD-BasinFlow Compute Savings Verdict", ""])
        real_prefixes = ("ag_news", "fashion_mnist", "cifar10", "tabular_credit", "sst2", "imdb")
        savings_rows = []
        loss_rows = []
        for exp, exp_df in metrics.groupby("experiment"):
            full = exp_df[(exp_df["method"] == "HD-BasinFlow") & (exp_df["ablation"] == "full")]
            if full.empty:
                continue
            h_loss = float(full["median_final_best_loss"].iloc[0])
            h_ttt = float(full["median_time_to_target"].iloc[0]) if "median_time_to_target" in full else np.nan
            h_wall = float(full["median_walltime"].iloc[0]) if "median_walltime" in full else np.nan
            baseline_df = exp_df[exp_df["ablation"].eq("baseline")].copy()
            for _, base in baseline_df.iterrows():
                b_ttt = float(base.get("median_time_to_target", np.nan))
                b_wall = float(base.get("median_walltime", np.nan))
                b_loss = float(base.get("median_final_best_loss", np.nan))
                eval_saved = 1.0 - h_ttt / b_ttt if np.isfinite(h_ttt) and np.isfinite(b_ttt) and b_ttt > 0 else np.nan
                wall_saved = 1.0 - h_wall / b_wall if np.isfinite(h_wall) and np.isfinite(b_wall) and b_wall > 0 else np.nan
                same_or_better = bool(np.isfinite(h_loss) and np.isfinite(b_loss) and h_loss <= b_loss)
                savings_rows.append(
                    {
                        "experiment": exp,
                        "baseline": base["method"],
                        "eval_saved": eval_saved,
                        "wall_saved": wall_saved,
                        "same_or_better_loss": same_or_better,
                        "is_real": str(exp).startswith(real_prefixes),
                    }
                )
            best = exp_df.sort_values("median_final_best_loss").iloc[0]
            if best["method"] != "HD-BasinFlow" or best["ablation"] != "full":
                loss_rows.append((exp, best["method"], best["ablation"], float(best["median_final_best_loss"]), h_loss))
        if savings_rows:
            real_wins = [
                row
                for row in savings_rows
                if row["is_real"]
                and row["same_or_better_loss"]
                and (
                    (np.isfinite(row["eval_saved"]) and row["eval_saved"] >= 0.20)
                    or (np.isfinite(row["wall_saved"]) and row["wall_saved"] >= 0.20)
                )
            ]
            synthetic_wins = [
                row
                for row in savings_rows
                if not row["is_real"]
                and row["same_or_better_loss"]
                and np.isfinite(row["eval_saved"])
                and row["eval_saved"] > 0
            ]
            if real_wins:
                lines.append("- Current evidence meets the 20 percent compute-savings target on at least one real workload/baseline comparison:")
                for row in real_wins[:10]:
                    lines.append(
                        f"  - `{row['experiment']}` vs `{row['baseline']}`: eval saving `{row['eval_saved']:.1%}`; wall-time proxy saving `{row['wall_saved']:.1%}`; HD full loss was no worse than the baseline median."
                    )
            else:
                lines.append("- Current combined evidence does **not** prove the PDF's 20 percent real-workload compute-savings target for HD-BasinFlow full. Some real workloads are won by baselines or use too small a budget for a strong savings claim.")
            if synthetic_wins:
                lines.append("- Synthetic or dimension-scaling comparisons where HD-BasinFlow full saved evaluations while matching/beating a baseline include:")
                for row in synthetic_wins[:10]:
                    lines.append(f"  - `{row['experiment']}` vs `{row['baseline']}`: eval saving `{row['eval_saved']:.1%}`.")
            else:
                lines.append("- No synthetic evaluation-saving win for HD-BasinFlow full was proven by the median time-to-target metric in this run.")
        else:
            lines.append("- No HD-BasinFlow full versus baseline comparison was available in this run.")
        _, false_kill_summary = _early_stopping_tables(evaluations)
        lines.extend(["", "## False-Kill Threshold Check", ""])
        if not false_kill_summary.empty:
            tracked = false_kill_summary[
                (false_kill_summary["method"].eq("ASHA Successive Halving") & false_kill_summary["ablation"].eq("baseline"))
                | (false_kill_summary["method"].eq("HD-BasinFlow") & false_kill_summary["ablation"].eq("full"))
            ].copy()
            if tracked.empty:
                tracked = false_kill_summary.copy()
            pass_5 = int(tracked["false_kill_pass_5pct"].sum())
            pass_10 = int(tracked["false_kill_pass_10pct"].sum())
            failed = tracked[tracked["commercial_false_kill_status"].eq("fail_gt_10pct")].copy()
            lines.append(
                f"- Against the PDF's 5-10 percent good-run false-kill condition, `{pass_5}` tracked method/workload rows pass at 5 percent and `{pass_10}` pass at 10 percent."
            )
            if not failed.empty:
                lines.append("- Rows failing the 10 percent false-kill threshold:")
                for row in failed.sort_values("good_run_false_kill_rate", ascending=False).itertuples(index=False):
                    lines.append(
                        f"  - `{row.experiment}` / `{row.method}` / `{row.ablation}`: false-kill rate `{row.good_run_false_kill_rate:.1%}` over `{row.all_good_runs}` good runs."
                    )
            else:
                lines.append("- No tracked method/workload row exceeded the 10 percent false-kill threshold.")
            lines.append("- See `plot_data/false_kill_threshold_summary.csv` and `figures/false_kill_threshold_summary.png/.pdf`.")
        else:
            lines.append("- No explicit `killed_early` decisions were available in this run, so the 5-10 percent false-kill condition is not evaluated here.")
        ablation_interpretation = _ablation_interpretation(metrics)
        lines.extend(["", "## Ablation Interpretation", ""])
        if not ablation_interpretation.empty:
            full_best = ablation_interpretation[ablation_interpretation["best_ablation"].eq("full")]
            mixed = ablation_interpretation[~ablation_interpretation["best_ablation"].eq("full")]
            pieces = ablation_interpretation[ablation_interpretation["flow_certificate_pieces_matter"].astype(bool)]
            lines.append(
                f"- HD-BasinFlow `full` is the best HD ablation by median final loss on `{len(full_best)}` of `{len(ablation_interpretation)}` experiments."
            )
            lines.append(
                f"- At least one ablation is worse than `full` on `{len(pieces)}` experiments, which is evidence that some flow/certificate/randomization pieces matter there."
            )
            if not mixed.empty:
                lines.append("- The ablation evidence is mixed; these experiments are won by an ablation rather than `full`:")
                for row in mixed.sort_values("experiment").itertuples(index=False):
                    lines.append(
                        f"  - `{row.experiment}`: best HD ablation `{row.best_ablation}` at `{row.best_ablation_median_final_best_loss:.6g}` versus full `{row.full_median_final_best_loss:.6g}`."
                    )
            else:
                lines.append("- No HD ablation beats `full` by median final loss in this run.")
            lines.append("- See `plot_data/ablation_interpretation.csv`, `plot_data/ablation_summary.csv`, and the ablation best-loss/time-to-target figures.")
        else:
            lines.append("- No HD-BasinFlow ablation rows were available in this run.")
        lines.extend(["", "## What Failed Or Remains Unproven", ""])
        if loss_rows:
            lines.append("- HD-BasinFlow full was not the best median-final-loss method on these experiments:")
            for exp, method, ablation, best_loss, h_loss in loss_rows[:20]:
                lines.append(
                    f"  - `{exp}`: winner `{method}` / `{ablation}` at `{best_loss:.6g}` versus HD full `{h_loss:.6g}`."
                )
            if len(loss_rows) > 20:
                lines.append(f"  - ...and {len(loss_rows) - 20} more experiments.")
        else:
            lines.append("- HD-BasinFlow full matched the best median-final-loss method on every experiment in this run.")
        lines.extend(["", "## Caveats", ""])
        lines.append("- Synthetic smoke runs are cheap validation, not proof of commercial GPU savings.")
        if "Optuna TPE" not in set(evaluations["method"]):
            lines.append("- Optuna TPE is named in the PDF but was not present in this run. Install `optuna` and rerun to include it.")
        if "ASHA Successive Halving" not in set(evaluations["method"]):
            lines.append("- ASHA/HyperBand is named in the PDF but was not present in this run.")
        experiments = {str(exp) for exp in evaluations["experiment"].unique()}
        if (
            any(exp.startswith("ag_news") for exp in experiments)
            or any(exp.startswith("fashion_mnist") for exp in experiments)
            or any(exp.startswith("cifar10") for exp in experiments)
            or any(exp.startswith("tabular_credit") for exp in experiments)
            or any(exp.startswith("sst2") for exp in experiments)
            or any(exp.startswith("imdb") for exp in experiments)
        ):
            included = []
            if any(exp.startswith("ag_news") for exp in experiments):
                included.append("AG News TF-IDF/logistic-regression")
            if any(exp.startswith("fashion_mnist") for exp in experiments):
                included.append("Fashion-MNIST tiny-CNN")
            if any(exp.startswith("cifar10") for exp in experiments):
                included.append("CIFAR-10 tiny-CNN")
            if any(exp.startswith("tabular_credit") for exp in experiments):
                included.append("tabular credit HistGradientBoosting")
            if any(exp.startswith("sst2") for exp in experiments):
                included.append("SST-2 DistilBERT")
            if any(exp.startswith("imdb") for exp in experiments):
                included.append("IMDB DistilBERT")
            lines.append(f"- This run includes real-data workload(s): {', '.join(included)}. Remaining larger/full-budget workloads still require separate runs.")
        else:
            lines.append("- Real ML training is not executed by the synthetic command. Use the real-data status output to confirm dataset/GPU availability before launching expensive training.")
        lines.append("")
        if pairwise is not None and not pairwise.empty:
            lines.extend(["## Pairwise Score Validation", ""])
            lines.append(f"- Pairwise edge rows: {len(pairwise):,}")
            lines.append("- See `processed/pairwise_score_validation.csv` and `plot_data/pairwise_score_auc_precision.csv`.")
            lines.append("")
        if certificates is not None and not certificates.empty:
            lines.extend(["## Certificate Validation", ""])
            lines.append(f"- Basin certificate rows: {len(certificates):,}")
            saddle_false = int(certificates["saddle_false_acceptance"].sum()) if "saddle_false_acceptance" in certificates else 0
            lines.append(f"- Saddle false acceptances recorded: {saddle_false}")
            lines.append("- See `processed/certificate_validation.csv` and `plot_data/certificate_precision_recall.csv`.")
            lines.append("")
    else:
        lines.extend(["## What Ran", "", "- No synthetic evaluations were requested in this run.", ""])
    if real_status is not None:
        lines.extend(["## Real Dataset Status", "", _safe_markdown_table(real_status), ""])
    lines.extend(
        [
            "## Deliverables",
            "",
            "- Raw logs: `raw/evaluations.csv` when synthetic runs are requested.",
            "- Processed metrics: `processed/metrics.csv`.",
            "- Figures: `figures/*.png` and `figures/*.pdf`.",
            "- Plot data: `plot_data/*.csv`.",
            "",
        ]
    )
    path.write_text("\n".join(lines), encoding="utf-8")