eval/make_tables.py

#!/usr/bin/env python3
"""Emit the manuscript's result tables as CSV from `reproduced_headline.json`.

    python3 eval/make_tables.py [--headline results/reproduced_headline.json]

Writes, under results/:
    experiment1_causal.csv         causal handles per state model
    experiment1_accuracy.csv       per-family accuracy per model
    experiment2_stageA.csv         Stage-A gate
    experiment2_confirmation.csv   sealed-split confirmation and its verdict
    decode_probe.csv               probe power control and durable-binding result
    long_delay.csv                 accuracy vs horizon and vs evidence delay

Every number in the paper's main result tables is one row of one of these.
"""
import argparse
import csv
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "src"))

from pns.common import eval_root  # noqa: E402

FAMS = ["SEM_LATEST", "SEM_2HOP", "IMMEDIATE_CMP", "EXACT_DELAYED", "SELF_REF",
        "HANDLE_REF", "DEADLINE", "GOAL_TOP", "all"]
MECHANISM = {
    "PNSR_K4_fin": "additive write + norm clamp tau=16",
    "PNSR_K4_s2": "additive write + norm clamp tau=16",
    "PNSR_K4_s3": "additive write + norm clamp tau=16",
    "G_TAU512": "additive write + norm clamp tau=512",
    "G_CONVEX_FREE": "convex gated update, no clamp",
    "G_CONVEX_RETAIN": "convex gated update, retention-biased",
    "RMT_s1": "memory tokens + full self-attention",
    "PNSR_K1_fin": "additive write + norm clamp tau=16 (K=1)",
    "TX768_fin": "none (768-token transcript window)",
    "TXE_fin": "none (current event only)",
}
SEED = {"PNSR_K4_fin": 1, "PNSR_K4_s2": 2, "PNSR_K4_s3": 3}


def w(path, header, rows):
    with open(path, "w", newline="") as f:
        c = csv.writer(f)
        c.writerow(header)
        c.writerows(rows)
    print(f"  {path.name:32s} {len(rows)} rows")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--headline", default=str(ROOT / "results" / "reproduced_headline.json"))
    ap.add_argument("--outdir", default=str(ROOT / "results"))
    args = ap.parse_args()
    H = json.loads(Path(args.headline).read_text())
    out = Path(args.outdir)
    out.mkdir(parents=True, exist_ok=True)
    e1 = H["experiment1_generic_recurrent_state"]["models"]

    # -------------------------------------------------- experiment1_causal.csv
    rows = []
    for run, m in e1.items():
        c = m.get("learned_state_causal", {})
        les = m.get("exact_store_lesions", {})
        st = m.get("state_statistics", {})
        rows.append([run, SEED.get(run, 1), MECHANISM.get(run, ""),
                     c.get("memhard_warm_val_1k"),
                     c.get("delta_cross_lifetime_swap"), c.get("delta_reset64"),
                     c.get("delta_reset32"), c.get("delta_state_zero"),
                     les.get("J_self_lesion_SELF_REF"),
                     les.get("J_obs_lesion_pointer_families"),
                     st.get("mean_cross_lifetime_cosine"), st.get("max_slot_norm")])
    w(out / "experiment1_causal.csv",
      ["run", "seed", "state_mechanism", "memhard_acc_warm_L1024",
       "delta_cross_lifetime_swap", "delta_reset64", "delta_reset32",
       "delta_state_zero", "delta_J_self_lesion_SELF_REF",
       "delta_J_obs_lesion_pointer_families", "mean_cross_lifetime_cosine",
       "max_slot_norm"], rows)

    # ------------------------------------------------ experiment1_accuracy.csv
    rows = []
    for run, m in e1.items():
        a = m["semantic_accuracy"]
        for fam in FAMS:
            if a.get(fam) is not None:
                rows.append([run, MECHANISM.get(run, ""), "val", 256, fam, a[fam]])
    w(out / "experiment1_accuracy.csv",
      ["run", "state_mechanism", "split", "lifetime_length", "family", "accuracy"], rows)

    # ----------------------------------------------------------- long_delay.csv
    rows = []
    for run, m in e1.items():
        for horizon, d in (m.get("horizons") or {}).items():
            for k, v in d.items():
                if v is not None:
                    rows.append([run, "experiment1", horizon, k, v])
    e2 = H.get("experiment2_protected_bindings") or {}
    for arm in ("bind", "capacity_matched_unbound"):
        for seed, d in (e2.get(arm) or {}).items():
            for k in ("accuracy_delay_gt16", "accuracy_delay_gt128", "overall"):
                if d.get(k) is not None:
                    rows.append([f"E3B_{arm}_{seed}", "experiment2", e2["split"], k, d[k]])
    w(out / "long_delay.csv",
      ["run", "experiment", "horizon_or_split", "metric", "value"], rows)

    # -------------------------------------------------------- decode_probe.csv
    rows = []
    for label, d in (H.get("experiment1_decode_probe") or {}).items():
        for k, v in d.items():
            rows.append([label, d.get("split_train"), d.get("split_test"), k, v])
    w(out / "decode_probe.csv",
      ["label", "decoder_train_split", "decoder_test_split", "metric", "value"], rows)

    # ------------------------------------------------ experiment2_stageA.csv
    p = eval_root() / "STAGEA_GATE.json"
    rows = []
    if p.exists():
        j = json.loads(p.read_text())
        if "bind" in j and isinstance(j["bind"], dict) and "1" in j["bind"]:
            # the schema the study's own Stage-A reporter wrote; kept readable
            # so results/published/stage_a_gate_output.json still parses
            for mode in ("bind", "unbound"):
                for s, r in sorted(j[mode].items()):
                    rows.append([f"E3A_{mode}_s{s}", mode, int(s), r["acc"],
                                 r["gt16"], r["gt128"], None, r["swap"],
                                 r["zero"], r["meets_all"]])
        else:
            for run, r in j.items():
                if run == "verdict":
                    continue
                rows.append([run, r["mode"], r["seed"], round(r["none"]["acc"], 4),
                             r["none"].get("acc_gt16"), r["none"].get("acc_gt128"),
                             r["none"].get("n_gt128"), r["delta_payload_swap"],
                             r["delta_payload_zero"], r["meets_all"]])
            rows.sort(key=lambda x: (x[1], x[2]))
    w(out / "experiment2_stageA.csv",
      ["run", "arm", "seed", "accuracy", "accuracy_delay_gt16",
       "accuracy_delay_gt128", "n_delay_gt128", "delta_payload_swap",
       "delta_payload_zero", "meets_gate"], rows)

    # ------------------------------------------ experiment2_confirmation.csv
    rows = []
    if e2:
        t = e2["TXE_memory_free_control"]
        rows.append(["TXE_ON_E3", "memory_free_control", "-", t["latest"],
                     t["2hop"], t["pooled"], "", "", "", "", "", ""])
        for arm, label in (("bind", "binding_local_writes"),
                           ("capacity_matched_unbound", "capacity_matched_unstructured")):
            for seed, d in (e2.get(arm) or {}).items():
                rows.append([f"E3B_{'bind' if arm == 'bind' else 'unbound'}_{seed}",
                             label, seed, d["SEM_LATEST"], d["SEM_2HOP"], d["overall"],
                             d["accuracy_delay_gt128"], d["delta_payload_swap"],
                             d["delta_reset64"], d["delta_payload_zero"],
                             d["accuracy_under_payload_zero"],
                             e2["verdict"].get(seed, {}).get("all", "")])
    w(out / "experiment2_confirmation.csv",
      ["run", "arm", "seed", "SEM_LATEST", "SEM_2HOP", "overall",
       "accuracy_delay_gt128", "delta_payload_swap", "delta_reset64",
       "delta_payload_zero", "accuracy_under_payload_zero",
       "conjunction_met_this_seed"], rows)

    if e2:
        v = e2["verdict"]
        print(f"\n  CONJUNCTION_MET = {v['CONJUNCTION_MET']} "
              f"(pooled gain {v['pooled_gain']}, "
              f"SEM_2HOP endpoint met = {e2['preregistered_endpoint_SEM_2HOP']['met']})")


if __name__ == "__main__":
    main()