experiment: complete xiiotid three-alpha beta study

scripts/analyze_cross_alpha.py

@@ -0,0 +1,230 @@
+#!/usr/bin/env python
+"""Read-only cross-alpha comparison for the X-IIoTID beta study.
+
+Aggregates the audited per-condition multiseed artifacts of several Dirichlet
+alpha levels into a single machine-readable comparison:
+
+- ``*__cross_alpha_beta_table.csv`` — alpha x beta pooled validation mean/std,
+  pooled recall, selected flag, and the realized heterogeneity summary fields;
+- ``*__cross_alpha_displacement.csv`` — per-class direct prototype displacement
+  ``||Cbar_j(beta)-Cbar_j(0)||`` by alpha;
+- ``*__cross_alpha_client_summary.csv`` — client improve/degrade/unchanged counts
+  by alpha/seed/beta;
+- ``*__cross_alpha_summary.json`` — condition-level pooled selections and ranges.
+
+Reads only existing analysis artifacts; no training, no test-based selection.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "src"))
+
+from protean_research.aggregation import BETA_CANDIDATES  # noqa: E402
+
+
+def read_csv(path: Path) -> list[dict]:
+    with path.open("r", encoding="utf-8", newline="") as handle:
+        return list(csv.DictReader(handle))
+
+
+def read_json(path: Path):
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def write_csv(path: Path, rows: list[dict]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if not rows:
+        path.write_text("", encoding="utf-8")
+        return
+    with path.open("w", encoding="utf-8", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(rows[0].keys()))
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def parse_args(argv=None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dataset", default="xiiotid")
+    parser.add_argument("--alphas", type=float, nargs="+", default=[0.25, 0.5, 0.75])
+    parser.add_argument("--seeds", type=int, nargs="+", default=[0, 1, 2])
+    parser.add_argument("--analysis-dir", default="artifacts/analysis")
+    parser.add_argument("--tag", default=None)
+    return parser.parse_args(argv)
+
+
+def _resolve(path: str) -> Path:
+    p = Path(path).expanduser()
+    return p if p.is_absolute() else REPO_ROOT / p
+
+
+def main(argv=None) -> int:
+    args = parse_args(argv)
+    analysis = _resolve(args.analysis_dir)
+    tag = args.tag or f"{args.dataset}__cross_alpha"
+
+    het_tag = f"{args.dataset}__cross_alpha__partition_heterogeneity"
+    het_summary = {
+        float(row["alpha"]): row
+        for row in read_csv(analysis / f"{het_tag}_summary.csv")
+    }
+
+    beta_table: list[dict] = []
+    displacement_rows: list[dict] = []
+    client_rows: list[dict] = []
+    summary = {"dataset": args.dataset, "alphas": {}, "betas": list(BETA_CANDIDATES)}
+
+    for alpha in args.alphas:
+        stem = f"{args.dataset}__alpha{alpha:g}__multiseed"
+        pooled_path = analysis / f"{stem}__validation_pooled.csv"
+        selection_path = analysis / f"{stem}__selection_summary.json"
+        effect_path = analysis / f"{stem}__prototype_effect_pooled.csv"
+        client_path = analysis / f"{stem}__client_validation_summary.csv"
+        test_path = analysis / f"{stem}__test_descriptive.csv"
+
+        pooled = read_csv(pooled_path)
+        selection = read_json(selection_path)
+        selected_beta = selection["primary_selection"]["selected_beta"]
+        het = het_summary.get(float(alpha), {})
+
+        effect = read_csv(effect_path) if effect_path.exists() else []
+        for row in effect:
+            displacement_rows.append({
+                "dataset": args.dataset,
+                "alpha": alpha,
+                "class_id": int(row["class_id"]),
+                "class_name": row["class_name"],
+                "beta": float(row["beta"]),
+                "n_seeds": int(row["n_seeds"]),
+                "mean_displacement": float(row["mean_displacement"]),
+                "std_sample_displacement": (
+                    float(row["std_sample_displacement"])
+                    if row["std_sample_displacement"] not in ("", "None")
+                    else None
+                ),
+                "mean_relative_displacement": (
+                    float(row["mean_relative_displacement"])
+                    if row["mean_relative_displacement"] not in ("", "None")
+                    else None
+                ),
+            })
+
+        if client_path.exists():
+            for row in read_csv(client_path):
+                client_rows.append({
+                    "dataset": args.dataset,
+                    "alpha": alpha,
+                    "seed": int(row["seed"]),
+                    "beta": float(row["beta"]),
+                    "n_eligible": int(row["n_eligible"]),
+                    "n_improve": int(row["n_improve_vs_beta0"]),
+                    "n_degrade": int(row["n_degrade_vs_beta0"]),
+                    "n_unchanged": int(row["n_unchanged"]),
+                    "mean_delta": (
+                        float(row["mean_delta_vs_beta0"])
+                        if row["mean_delta_vs_beta0"] not in ("", "None")
+                        else None
+                    ),
+                    "min_delta": (
+                        float(row["min_delta_vs_beta0"])
+                        if row["min_delta_vs_beta0"] not in ("", "None")
+                        else None
+                    ),
+                    "max_delta": (
+                        float(row["max_delta_vs_beta0"])
+                        if row["max_delta_vs_beta0"] not in ("", "None")
+                        else None
+                    ),
+                })
+
+        disp_beta1 = [
+            r for r in displacement_rows
+            if r["alpha"] == alpha and abs(r["beta"] - 1.0) <= 1e-9
+        ]
+        mean_disp_beta1 = (
+            sum(r["mean_displacement"] for r in disp_beta1) / len(disp_beta1)
+            if disp_beta1 else None
+        )
+
+        for row in pooled:
+            beta = float(row["beta"])
+            beta_table.append({
+                "dataset": args.dataset,
+                "alpha": alpha,
+                "beta": beta,
+                "n_seeds": int(row["n_seeds"]),
+                "pooled_val_mean": float(
+                    row["pooled_val_rare_macro_f1_per_client_mean"]
+                ),
+                "pooled_val_std_sample": float(row["std_sample_val_rare_macro_f1"]),
+                "pooled_val_std_pop": float(row["std_pop_val_rare_macro_f1"]),
+                "pooled_val_min": float(row["min_val_rare_macro_f1"]),
+                "pooled_val_max": float(row["max_val_rare_macro_f1"]),
+                "pooled_val_recall": float(
+                    row["pooled_val_rare_recall_per_client_mean"]
+                ),
+                "selected": abs(beta - selected_beta) <= 1e-12,
+                "cells_sj_lt_M_total": int(het.get("cells_sj_lt_M_total", -1)),
+                "cells_total": int(het.get("cells_total", -1)),
+                "missing_client_class_cells_total": int(
+                    het.get("missing_client_class_cells_total", -1)
+                ),
+                "client_size_ratio_mean": float(het.get("client_size_ratio_mean", -1)),
+                "support_cv_mean": float(het.get("support_cv_mean", -1)),
+                "union_unseen_mean": float(het.get("union_unseen_mean", -1)),
+                "mean_displacement_beta1": mean_disp_beta1,
+            })
+
+        summary["alphas"][f"{alpha:g}"] = {
+            "selected_beta": selected_beta,
+            "tie_break_used": selection["primary_selection"]["tie_break_used"],
+            "selection_rule": selection["primary_selection"]["rule"],
+            "test_used_for_selection": selection["primary_selection"][
+                "test_used_for_selection"
+            ],
+            "pooled_val_mean_range": [
+                min(float(r["pooled_val_mean"]) for r in beta_table
+                    if r["alpha"] == alpha),
+                max(float(r["pooled_val_mean"]) for r in beta_table
+                    if r["alpha"] == alpha),
+            ],
+            "pooled_val_std_sample_range": [
+                min(float(r["pooled_val_std_sample"]) for r in beta_table
+                    if r["alpha"] == alpha),
+                max(float(r["pooled_val_std_sample"]) for r in beta_table
+                    if r["alpha"] == alpha),
+            ],
+            "cells_sj_lt_M_total": int(het.get("cells_sj_lt_M_total", -1)),
+            "cells_total": int(het.get("cells_total", -1)),
+            "missing_client_class_cells_total": int(
+                het.get("missing_client_class_cells_total", -1)
+            ),
+            "client_size_ratio_mean": float(het.get("client_size_ratio_mean", -1)),
+            "support_cv_mean": float(het.get("support_cv_mean", -1)),
+            "union_unseen_mean": float(het.get("union_unseen_mean", -1)),
+            "mean_displacement_beta1": mean_disp_beta1,
+            "test_descriptive_present": test_path.exists(),
+        }
+
+    write_csv(analysis / f"{tag}__cross_alpha_beta_table.csv", beta_table)
+    write_csv(analysis / f"{tag}__cross_alpha_displacement.csv", displacement_rows)
+    write_csv(analysis / f"{tag}__cross_alpha_client_summary.csv", client_rows)
+    with (analysis / f"{tag}__cross_alpha_summary.json").open(
+        "w", encoding="utf-8"
+    ) as handle:
+        json.dump(summary, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+
+    print(json.dumps(summary, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/analyze_partition_heterogeneity.py

@@ -0,0 +1,213 @@
+#!/usr/bin/env python
+"""Read-only realized non-IID characterization from persisted partition manifests.
+
+For each ``(dataset, alpha, seed)`` manifest this script reconstructs the training
+class-support matrix ``N_ij`` and emits machine-readable heterogeneity statistics:
+
+- ``*__partition_heterogeneity_per_class.csv`` — one row per ``(alpha, seed, class)``
+  with the protocol-required ``min/max/mean/std/CV/max-min`` inequality measures;
+- ``*__partition_heterogeneity_per_seed.csv`` — one row per ``(alpha, seed)`` with
+  ``|S_j| < M`` cell count, missing client-class cells, client-size spread,
+  rare/unseen counts and class-CV summaries;
+- ``*__partition_heterogeneity_summary.csv`` — per-alpha aggregation across seeds.
+
+This reads existing manifests only. It never regenerates a partition, never reads
+test data and never trains.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import statistics
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "src"))
+
+from protean_research.data import manifest as manifest_mod  # noqa: E402
+from protean_research.data.characterization import characterize_support  # noqa: E402
+from protean_research.data.rare_unseen import (  # noqa: E402
+    rare_class_ids_for_client,
+    unseen_class_ids_for_client,
+)
+
+
+def write_csv(path: Path, rows: list[dict]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if not rows:
+        path.write_text("", encoding="utf-8")
+        return
+    with path.open("w", encoding="utf-8", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(rows[0].keys()))
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def parse_args(argv=None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dataset", default="xiiotid")
+    parser.add_argument("--alphas", type=float, nargs="+", default=[0.75, 0.5, 0.25])
+    parser.add_argument("--seeds", type=int, nargs="+", default=[0, 1, 2])
+    parser.add_argument("--partitions-dir", default="artifacts/partitions")
+    parser.add_argument("--analysis-dir", default="artifacts/analysis")
+    parser.add_argument(
+        "--tag", default=None,
+        help="output stem (default: <dataset>__cross_alpha__partition_heterogeneity)",
+    )
+    return parser.parse_args(argv)
+
+
+def _resolve(path: str) -> Path:
+    p = Path(path).expanduser()
+    return p if p.is_absolute() else REPO_ROOT / p
+
+
+def main(argv=None) -> int:
+    import numpy as np
+
+    args = parse_args(argv)
+    partitions_dir = _resolve(args.partitions_dir)
+    analysis_dir = _resolve(args.analysis_dir)
+    tag = args.tag or f"{args.dataset}__cross_alpha__partition_heterogeneity"
+
+    per_class_rows: list[dict] = []
+    per_seed_rows: list[dict] = []
+
+    for alpha in args.alphas:
+        for seed in args.seeds:
+            path = manifest_mod.manifest_path(partitions_dir, args.dataset, alpha, seed)
+            manifest = manifest_mod.load_manifest(path)
+            support = np.asarray(manifest["support"], dtype=np.int64)
+            class_names = list(manifest["class_names"])
+            benign_index = class_names.index(manifest["benign_class"])
+            num_clients = int(manifest["num_clients"])
+            num_classes = int(manifest["num_classes"])
+            assert support.shape == (num_clients, num_classes)
+
+            class_stats = characterize_support(support, class_names)
+            for row in class_stats:
+                per_class_rows.append({
+                    "dataset": args.dataset,
+                    "alpha": alpha,
+                    "seed": seed,
+                    **row,
+                })
+
+            sj = (support > 0).sum(axis=0)
+            cells_lt_m = int((sj < num_clients).sum())
+            missing_cells = int((support == 0).sum())
+
+            totals = support.sum(axis=1).astype(int)
+            client_min = int(totals.min())
+            client_max = int(totals.max())
+            client_ratio = float(client_max / client_min) if client_min > 0 else None
+
+            rare_lists = [
+                rare_class_ids_for_client(support[c], benign_index)
+                for c in range(num_clients)
+            ]
+            unseen_lists = [
+                unseen_class_ids_for_client(support[c], benign_index)
+                for c in range(num_clients)
+            ]
+            union_rare = sorted({c for row in rare_lists for c in row})
+            union_unseen = sorted({c for row in unseen_lists for c in row})
+
+            cvs = [row["cv"] for row in class_stats if row["total"] > 0]
+            max_min_values = [
+                row["max_min_ratio"] for row in class_stats if row["max_min_ratio"] > 0
+            ]
+
+            per_seed_rows.append({
+                "dataset": args.dataset,
+                "alpha": alpha,
+                "seed": seed,
+                "num_clients": num_clients,
+                "num_classes": num_classes,
+                "num_train": int(manifest["num_train"]),
+                "nij_sum": int(support.sum()),
+                "n_class_condition_cells": num_classes,
+                "n_cells_sj_lt_M": cells_lt_m,
+                "n_missing_client_class_cells": missing_cells,
+                "client_size_min": client_min,
+                "client_size_max": client_max,
+                "client_size_ratio": client_ratio,
+                "union_rare_count": len(union_rare),
+                "union_unseen_count": len(union_unseen),
+                "max_unseen_on_client": max((len(u) for u in unseen_lists), default=0),
+                "n_clients_2_rare": sum(1 for r in rare_lists if len(r) == 2),
+                "support_cv_mean": statistics.fmean(cvs) if cvs else None,
+                "support_cv_max": max(cvs) if cvs else None,
+                "support_max_min_ratio_max": max(max_min_values) if max_min_values else None,
+                "manifest_content_hash": manifest_mod.manifest_content_hash(manifest),
+                "partition_hash": manifest["partition_hash"],
+                "split_hash": manifest["split"]["hash"],
+                "rng_seed": int(manifest["rng_seed"]),
+            })
+
+    # per-alpha summary across seeds
+    summary_rows: list[dict] = []
+    for alpha in args.alphas:
+        subset = [r for r in per_seed_rows if r["alpha"] == alpha]
+        if not subset:
+            continue
+        summary_rows.append({
+            "dataset": args.dataset,
+            "alpha": alpha,
+            "n_seeds": len(subset),
+            "cells_sj_lt_M_total": sum(r["n_cells_sj_lt_M"] for r in subset),
+            "cells_total": sum(r["n_class_condition_cells"] for r in subset),
+            "missing_client_class_cells_total": sum(
+                r["n_missing_client_class_cells"] for r in subset
+            ),
+            "client_size_ratio_mean": statistics.fmean(
+                r["client_size_ratio"] for r in subset
+            ),
+            "client_size_ratio_max": max(r["client_size_ratio"] for r in subset),
+            "union_unseen_mean": statistics.fmean(r["union_unseen_count"] for r in subset),
+            "max_unseen_on_client": max(r["max_unseen_on_client"] for r in subset),
+            "union_rare_mean": statistics.fmean(r["union_rare_count"] for r in subset),
+            "support_cv_mean": statistics.fmean(r["support_cv_mean"] for r in subset),
+            "support_max_min_ratio_max": max(
+                r["support_max_min_ratio_max"] for r in subset
+            ),
+        })
+
+    write_csv(analysis_dir / f"{tag}_per_class.csv", per_class_rows)
+    write_csv(analysis_dir / f"{tag}_per_seed.csv", per_seed_rows)
+    write_csv(analysis_dir / f"{tag}_summary.csv", summary_rows)
+
+    payload = {
+        "dataset": args.dataset,
+        "alphas": args.alphas,
+        "seeds": args.seeds,
+        "per_class": str(analysis_dir / f"{tag}_per_class.csv"),
+        "per_seed": str(analysis_dir / f"{tag}_per_seed.csv"),
+        "summary": str(analysis_dir / f"{tag}_summary.csv"),
+        "per_seed_rows": per_seed_rows,
+        "summary_rows": summary_rows,
+    }
+    out_json = analysis_dir / f"{tag}.json"
+    with out_json.open("w", encoding="utf-8") as handle:
+        json.dump(payload, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+
+    print(json.dumps({
+        "tag": tag,
+        "n_per_class_rows": len(per_class_rows),
+        "n_per_seed_rows": len(per_seed_rows),
+        "summary": [{"alpha": r["alpha"],
+                     "cells_sj_lt_M": f"{r['cells_sj_lt_M_total']}/{r['cells_total']}",
+                     "missing_cells": r["missing_client_class_cells_total"],
+                     "client_ratio_mean": round(r["client_size_ratio_mean"], 3)}
+                    for r in summary_rows],
+        "json": str(out_json),
+    }, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/verify_beta0_equivalence.py

@@ -0,0 +1,209 @@
+#!/usr/bin/env python
+"""Read-only ``beta = 0`` vs ``original_protean`` equivalence audit.
+
+For each seed of one ``(dataset, alpha)`` condition this compares the frozen
+``original_protean`` run with the ``support_weighted beta=0`` run:
+
+- manifest content hash, round-0 snapshot bytes, support/characterization/
+  rare-unseen artifact bytes (pairing / identity);
+- independent aggregation arithmetic on the shared round-0 snapshot
+  (``original_protean`` equal mean vs ``support_weighted beta=0``);
+- downstream divergence (history and test-metric numeric deltas).
+
+Classification per seed:
+  A — semantic + numerical identity;
+  B — semantic equivalence + floating-point divergence;
+  C — semantic discrepancy.
+
+Read-only: no training, no aggregation source change.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "src"))
+
+from protean_research.aggregation import (  # noqa: E402
+    AggregationMode,
+    aggregate_class_prototypes,
+)
+from protean_research.analysis.direct_prototype import (  # noqa: E402
+    snapshot_contributions,
+)
+
+PAIR_FILES = ("train_support_Nij.csv", "characterization.csv", "rare_unseen.csv")
+
+
+def sha256_bytes(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def read_json(path: Path):
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def numeric_max_abs_diff(a, b) -> float:
+    if isinstance(a, dict) and isinstance(b, dict):
+        diffs = [
+            numeric_max_abs_diff(a[k], b[k])
+            for k in a.keys() & b.keys()
+            if isinstance(a[k], (int, float)) and isinstance(b[k], (int, float))
+        ]
+        return max(diffs) if diffs else 0.0
+    if isinstance(a, list) and isinstance(b, list) and len(a) == len(b):
+        diffs = [numeric_max_abs_diff(x, y) for x, y in zip(a, b)]
+        return max(diffs) if diffs else 0.0
+    if isinstance(a, (int, float)) and isinstance(b, (int, float)):
+        return abs(float(a) - float(b))
+    return 0.0
+
+
+def history_rare_f1(history) -> list[float]:
+    return [
+        float(row["rare_class_macro_f1"])
+        for row in history
+        if row.get("rare_class_macro_f1") is not None
+    ]
+
+
+def max_abs_diff(xs, ys) -> float:
+    return max((abs(x - y) for x, y in zip(xs, ys)), default=0.0)
+
+
+def aggregation_difference(snapshot) -> float:
+    contributions = snapshot_contributions(snapshot)
+    original = aggregate_class_prototypes(
+        contributions, AggregationMode.ORIGINAL_PROTEAN
+    )
+    beta0 = aggregate_class_prototypes(
+        contributions, AggregationMode.SUPPORT_WEIGHTED, beta=0.0
+    )
+    diffs = []
+    for class_id in original:
+        diffs.extend(abs(x - y) for x, y in zip(original[class_id], beta0[class_id]))
+    return max(diffs) if diffs else 0.0
+
+
+def parse_args(argv=None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dataset", required=True)
+    parser.add_argument("--alpha", type=float, required=True)
+    parser.add_argument("--seeds", type=int, nargs="+", default=[0, 1, 2])
+    parser.add_argument("--results-dir", default="results")
+    parser.add_argument("--analysis-dir", default="artifacts/analysis")
+    parser.add_argument("--tag", default=None)
+    return parser.parse_args(argv)
+
+
+def _resolve(path: str) -> Path:
+    p = Path(path).expanduser()
+    return p if p.is_absolute() else REPO_ROOT / p
+
+
+def main(argv=None) -> int:
+    args = parse_args(argv)
+    results = _resolve(args.results_dir)
+    analysis = _resolve(args.analysis_dir)
+    tag = args.tag or f"{args.dataset}__alpha{args.alpha:g}__multiseed"
+
+    rows: list[dict] = []
+    for seed in args.seeds:
+        orig_name = f"{args.dataset}__alpha{args.alpha:g}__seed{seed}__original_protean"
+        b0_name = (
+            f"{args.dataset}__alpha{args.alpha:g}__seed{seed}"
+            "__support_weighted__beta0"
+        )
+        orig_dir = results / orig_name
+        b0_dir = results / b0_name
+
+        orig_meta = read_json(orig_dir / "metadata.json")
+        b0_meta = read_json(b0_dir / "metadata.json")
+
+        orig_snap = analysis / f"{orig_name}__prototype_snapshot.json"
+        b0_snap = analysis / f"{b0_name}__prototype_snapshot.json"
+        orig_snap_hash = sha256_bytes(orig_snap) if orig_snap.exists() else None
+        b0_snap_hash = sha256_bytes(b0_snap) if b0_snap.exists() else None
+
+        pair_hashes = {}
+        pair_identical = True
+        for filename in PAIR_FILES:
+            o = orig_dir / filename
+            b = b0_dir / filename
+            same = o.exists() and b.exists() and sha256_bytes(o) == sha256_bytes(b)
+            pair_hashes[filename] = same
+            pair_identical = pair_identical and same
+
+        history_orig = read_json(orig_dir / "history.json")
+        history_b0 = read_json(b0_dir / "history.json")
+        f1_orig = history_rare_f1(history_orig)
+        f1_b0 = history_rare_f1(history_b0)
+        round0_diff = max_abs_diff(f1_orig[:1], f1_b0[:1])
+        later_diff = max_abs_diff(f1_orig[1:], f1_b0[1:])
+
+        test_diff = numeric_max_abs_diff(
+            read_json(orig_dir / "test_global_metrics.json"),
+            read_json(b0_dir / "test_global_metrics.json"),
+        )
+
+        agg_diff = aggregation_difference(read_json(b0_snap))
+
+        same_manifest = (
+            orig_meta.get("manifest_content_hash") == b0_meta.get("manifest_content_hash")
+        )
+        semantic_ok = (
+            same_manifest
+            and orig_snap_hash is not None
+            and orig_snap_hash == b0_snap_hash
+            and pair_identical
+            and agg_diff <= 1e-12
+        )
+        numerical_identity = (
+            semantic_ok
+            and later_diff == 0.0
+            and test_diff == 0.0
+            and history_orig == history_b0
+        )
+        if numerical_identity:
+            classification = "A"
+        elif semantic_ok:
+            classification = "B"
+        else:
+            classification = "C"
+
+        rows.append({
+            "dataset": args.dataset,
+            "alpha": args.alpha,
+            "seed": seed,
+            "same_manifest_content_hash": same_manifest,
+            "round0_snapshot_sha256_original": orig_snap_hash,
+            "round0_snapshot_sha256_beta0": b0_snap_hash,
+            "round0_snapshot_byte_identical": orig_snap_hash == b0_snap_hash,
+            "support_char_rare_identical": pair_identical,
+            "aggregation_max_abs_diff": agg_diff,
+            "history_rare_f1_round0_diff": round0_diff,
+            "history_rare_f1_max_diff_rounds_ge1": later_diff,
+            "test_metric_max_abs_diff": test_diff,
+            "classification": classification,
+        })
+
+    out = analysis / f"{tag}__beta0_equivalence.csv"
+    out.parent.mkdir(parents=True, exist_ok=True)
+    with out.open("w", encoding="utf-8", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(rows[0].keys()))
+        writer.writeheader()
+        writer.writerows(rows)
+
+    print(json.dumps({"output": str(out), "rows": rows}, indent=2, default=str))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())