Buckets:

AsianPlayer
/

TriDrive

Files

xet

AsianPlayer/TriDrive / code /eval /report_kin_v2p_arm.py

AsianPlayer

16 days ago

download

raw

21.1 kB

"""cwm3/eval/report_kin_v2p_arm.py — results/tables/kin_v2p_arm.md (+ .json): the kin v2p ADDITIVE INPUT ARM (plan A; owner decision docs/decisions.md row 510;
pre-registration row 2026-09-18T05:34:40Z) on the kpf joint model.  Composes eval/report_joint.py + eval/report_final_face.py blocks (load / render / verdict /
margin table) over results/paired_compare.csv, eval/alert_pair_common.gold_block (paired gold clean-test AUROC / PR-AUC on identical device-cluster resamples),
the alert-timing run holding the v2p / kpf read-out scores, the evaluation-only phone / gaze episode alert shares (eval/phone_episode_alerts.py) and the run
jsons (params, GPU minutes).  Nothing is recomputed except the paired gold bootstrap and the 10-bin ECE on the gold rows; every other number is a script output.
  python eval/report_kin_v2p_arm.py [--seeds 0,1,2] [--timing-dir DIR] [--phone results/kin_v2p_arm/phone_alerts/phone_alerts.json] [--n-boot 2000] [--out results/tables/kin_v2p_arm.md]
ACCEPTANCE (pre-registered, read mechanically): joint model on VALIDATION — no own onset cell (report_joint.OWN) WORSE, no onset cell beyond the −0.005 margin,
no hazard-NLL cell beyond +0.010; AND EITHER >= 1 of the 4 reported driver cells BEATS on validation, OR (gold pooled AUROC and PR-AUC of A_W vs B_W not WORSE by the
seed rule AND the phone / gaze alert share of A_W higher than B_W in the 3-seed mean at BOTH 20 and 40 per h).  Meeting the rule = ADOPTION CANDIDATE only
(the owner's one-shot hold-out decides adoption).  No openpilot DMS field is read.
"""
from __future__ import annotations

import argparse
import json
import sys
import time
from pathlib import Path

import numpy as np
import pandas as pd

CWM3 = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(CWM3 / "eval"))
import report_joint as RJ                                            # noqa: E402
import report_final_face as RF                                       # noqa: E402
import alert_pair_common as C                                        # noqa: E402

A, B = "allminusecankpfv2p", "allminusecankpf"
PREREG = "2026-09-18T05:34:40Z"
PAPER_DRV = [("manual", "drv_onset|hand_leave|3s"), ("manual", "drv_onset|hand_return|3s"), ("manual", "drv_onset|head_turn_onset|3s"), ("engaged", "drv_onset|head_turn_onset|3s")]
VEH4 = ["veh_onset|gas_on|3s", "veh_onset|brake_on|3s", "veh_onset|steer_phase_on|3s", "veh_onset|veh_any_on|3s"]
NLL_VEH = ["hazard_nll|gas_on", "hazard_nll|brake_on", "hazard_nll|steer_phase_on"]
MARGIN_ONSET_CELLS = RF.DRV + VEH4
RUN_A, RUN_B = "I1_all_minus_road2drv_s{seed}_final_ecan_kpf_kcv2p", "I1_all_minus_road2drv_s{seed}_final_ecan_kpf"
COL, COLK = "W_learned_curhgb_t15_w10_n979dwfc", "W_learned_curhgb_t15_w10_n979dwfckp"
SPECS = {"A_W": (f"W_{RUN_A}_dwfckp", COLK), "B_W": (f"W_{RUN_B}_dwfc", COL), "Ap_W": (f"W_{RUN_B}_dwfckp", COLK)}
LABELS = {"A_W": "A_W: v2p joint model, cur+dwell+face+kp (HGB)", "B_W": "B_W: kpf joint model (FINAL), cur+dwell+face (paper recipe)", "Ap_W": "A'_W descr.: kpf joint model, cur+dwell+face+kp (read-out-only effect)"}
PAIRS = [("A_W", "B_W", "A_W − B_W (PRIMARY: v2p model + kp features vs the final model's read-out)", True),
         ("Ap_W", "B_W", "A'_W − B_W (descr.: kp features on the unchanged model)", False), ("A_W", "Ap_W", "A_W − A'_W (descr.: model effect at equal features)", False)]
TIMING_PAIRS = [("v2p_dwfckp", "kpf_dwfc", True), ("kpf_dwfckp", "kpf_dwfc", False), ("v2p_dwfckp", "kpf_dwfckp", False)]
LEVEL_SCORES = ["kpf_dwfc", "v2p_dwfckp", "kpf_dwfckp", "cur_dwfc", "surrogate_att3s", "persistence"]
f3, ci_s = C.f3, C.ci_s


def ece_block(seeds, specs, n_bins=10, log=print):
    """10-bin equal-width ECE of each spec's score on the gold clean-test rows (pooled), per seed — descriptive."""
    lab = C.LB.episode_labels(); lab = lab[lab.label != "unsure"].reset_index(drop=True)
    packs = {(r, s): C.WG.load_pack(r, s) for r in C.REGIMES for s in ("test", "val")}
    rows = []
    for seed in seeds:
        G = C.CR.gold_for_seed(lab, packs, seed); yg = G.y.to_numpy().astype(int); m0 = C.strata(G)["pooled"]
        for key, (pat, col) in specs.items():
            if not C.layouts_present(pat, seed):
                continue
            s = C.at_gold(G, C.load_col(pat.format(seed=seed), col, packs)); fin = m0 & np.isfinite(s); y, p = yg[fin], np.clip(s[fin], 0, 1)
            edges = np.linspace(0, 1, n_bins + 1); ece = 0.0
            for lo, hi in zip(edges[:-1], edges[1:]):
                mb = (p >= lo) & (p < hi if hi < 1 else p <= hi)
                if mb.any():
                    ece += mb.mean() * abs(p[mb].mean() - y[mb].mean())
            rows.append({"key": key, "seed": seed, "ece": float(ece), "n": int(fin.sum()), "mean_score": float(p.mean()), "base_rate": float(y.mean())})
            log(f"  ECE s{seed} {key}: {ece:.4f} (n {fin.sum()})")
    return pd.DataFrame(rows)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--seeds", default="0,1,2")
    ap.add_argument("--timing-dir", default="")
    ap.add_argument("--phone", default=str(CWM3 / "results" / "kin_v2p_arm" / "phone_alerts" / "phone_alerts.json"))
    ap.add_argument("--n-boot", type=int, default=2000)
    ap.add_argument("--out", default=str(CWM3 / "results" / "tables" / "kin_v2p_arm.md"))
    ap.add_argument("--skip-gold", action="store_true")
    a = ap.parse_args()
    seeds = [int(s) for s in a.seeds.split(",")]; stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()); t0 = time.time()
    outd = CWM3 / "results" / "kin_v2p_arm" / time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()); outd.mkdir(parents=True, exist_ok=True)
    log = lambda m: print(f"[{time.strftime('%H:%M:%S')}] {m}", flush=True)     # noqa: E731
    dec = {"report": stamp, "prereg": PREREG, "seeds": seeds, "A": RUN_A, "B": RUN_B}
    L = [f"# kin v2p ARM — kinematics v2 + 3 keypoint input proxies (head_pitch_kp / wrist_face_kp / hands_lap_kp) on the kpf joint model (pre-registration docs/decisions.md {PREREG}; report {stamp})", "",
         f"A = `{RUN_A}` (U base `U0_kin_can_noK_s<k>_kcv2p_clip8_ln`, 14 input kinematics; targets / states / gold unchanged) vs B = the FROZEN kpf joint model `{RUN_B}`.  Pairings eval/run_joint_evals_v2p.sh "
         "(pc_joint_locked / pressure_eval, n_boot 1000, identical rows, v2 causal kin targets = the identical populations); verdict = kill_checks.beats; margins onset −0.005 / NLL +0.010.  Read-out: paper recipe "
         "(train_warning_head.py --demand P --feature-sets current --extra-features dwell,face[,kp]); paired gold = alert_pair_common.gold_block on identical device-cluster resamples.  Every number is a script output; the manuscript is not edited here.", ""]
    # ------------------------------------------------------------------ 1. pack
    bs = CWM3 / "results" / "targets_kin_causal_routes_v2p" / "build_stats_v2p_latest.json"
    if bs.exists():
        S = json.loads(bs.read_text())
        L += ["### 1. Pack results/targets_kin_causal_routes_v2p (data/build_kin_causal_v2p.py)", "",
              f"shared arrays bit-identical to v2: **{S['shared_arrays_identical_to_v2']}** ({S['n_shared_arrays_checked']} arrays checked); routes with pose {S['n_routes_with_pose']}/{S['n_routes']}; "
              f"frame coverage: pitch defined {S['frame_coverage']['pitch_defined'] / max(S['frame_coverage']['frames'], 1):.4f}, wrist_face {S['frame_coverage']['wf_defined'] / max(S['frame_coverage']['frames'], 1):.4f}, "
              f"lap {S['frame_coverage']['lap_defined'] / max(S['frame_coverage']['frames'], 1):.4f}; head-down flag share (defined frames) {S['frame_coverage']['headdown'] / max(S['frame_coverage']['pitch_defined'], 1):.4f}", "",
              "| split | rows ok | pitch defined (valid anchors) | head-down share | dwell >= 1 s share | mean pitch dev | mean wrist_face | mean hands_lap |", "|---|---|---|---|---|---|---|---|"]
        for k, st in S["splits"].items():
            L.append(f"| {k} | {st['n_row_ok']} | {f3(st['proxy_defined_given_valid']['head_pitch_kp'], 4)} | {f3(st['headdown_share_given_defined'], 4)} | {f3(st['pitch_dwell_ge1s_share'], 4)} | "
                     f"{f3(st['proxy_mean_given_valid']['head_pitch_kp'], 4)} | {f3(st['proxy_mean_given_valid']['wrist_face_kp'], 3)} | {f3(st['proxy_mean_given_valid']['hands_lap_kp'], 3)} |")
        L.append(""); dec["pack"] = {"identical_to_v2": S["shared_arrays_identical_to_v2"], "n_checked": S["n_shared_arrays_checked"]}
    # ------------------------------------------------------------------ 2. joint model
    L += ["### 2. Joint model — v2p arm vs the frozen kpf joint model", ""]
    dfv, body, verd_v, cl_v = RF.block("finalcv", A, B, "B2", "(a-val) v2p vs FROZEN kpf joint model — VALIDATION split (acceptance decision)", seeds); L += body + [""]
    joint_ok = None
    if len(dfv):
        mt, checks_v = RF.margin_table(dfv, verd_v, "(a-val) driver / vehicle cells and hazard-NLL costs with the acceptable-degradation margins (VALIDATION)", MARGIN_ONSET_CELLS, RF.NLL_DRV + NLL_VEH); L += mt
        worse, broke, _, ht, insuff = RF.own_status(verd_v, checks_v)
        paper_beats = [f"{r}|{c}" for r, c in PAPER_DRV if verd_v.get((r, c, "pr_auc"), {}).get("verdict") == "BEATS"]
        paper_worse = [f"{r}|{c}" for r, c in PAPER_DRV if verd_v.get((r, c, "pr_auc"), {}).get("verdict") == "WORSE"]
        nll_broke = [k for k, c in checks_v.items() if k[2] == "nll" and not c["ok"]]
        joint_ok = (not worse) and (not broke) and (not nll_broke) and not insuff
        L += [f"**(a-val) joint gate:** own onset WORSE: {RF.fmt(worse)}; own onset cells beyond −0.005: {RF.fmt(broke)}; NLL cells beyond +0.010: {RF.fmt(nll_broke)}; insufficient: {insuff}; "
              f"paper driver cells BEATS: {paper_beats or 'none'}; WORSE: {paper_worse or 'none'}; head_turn verdicts {ht} -> **gate {'PASSED' if joint_ok else 'FAILED'}**", ""]
        dec["validation"] = {"gate_passed": joint_ok, "own_worse": RF.fmt(worse), "own_beyond_margin": RF.fmt(broke), "nll_beyond_margin": RF.fmt(nll_broke), "insufficient": insuff, "paper_beats": paper_beats, "paper_worse": paper_worse,
                             "cells": {f"{r}|{c}|{m}": v["verdict"] for (r, c, m), v in verd_v.items()}, "margin_checks": {f"{r}|{c}|{m}": c_ for (r, c, m), c_ in checks_v.items()}}
    else:
        dec["validation"] = {"status": "NO ROWS"}; paper_beats = []
    dft, body, verd_t, cl_t = RF.block("finalc", A, B, "B2", "(a) v2p vs FROZEN kpf joint model — TEST split (confirmation)", seeds); L += body + [""]
    if len(dft):
        mt, checks_t = RF.margin_table(dft, verd_t, "(a) driver / vehicle cells and hazard-NLL costs with the margins (TEST)", MARGIN_ONSET_CELLS, RF.NLL_DRV + NLL_VEH); L += mt
        worse_t, broke_t, _, ht_t, _ = RF.own_status(verd_t, checks_t)
        dec["test"] = {"own_worse": RF.fmt(worse_t), "own_beyond_margin": RF.fmt(broke_t), "head_turn": ht_t, "B2_style": cl_t[0], "cells": {f"{r}|{c}|{m}": v["verdict"] for (r, c, m), v in verd_t.items()},
                       "paper_cells": {f"{r}|{c}": verd_t.get((r, c, "pr_auc"), {}).get("verdict", "NO ROWS") for r, c in PAPER_DRV},
                       "hazard_nll": {f"{r}|{c}": verd_t.get((r, c, "nll"), {}).get("verdict", "NO ROWS") for r in ("manual", "engaged") for c in RF.NLL_DRV + NLL_VEH}}
        L += [f"**(a) TEST:** own onset WORSE: {RF.fmt(worse_t)}; beyond −0.005: {RF.fmt(broke_t)}; head_turn {ht_t}; B2-style claim: {cl_t[0]}.", ""]
    else:
        dec["test"] = {"status": "NO ROWS"}
    dfm, body, verd_m, _ = RF.block("finalcm", A, B, "B2", "masked protocol (human-verified driver-box mask) — TEST", seeds); L += body + [""]
    dec["masked"] = {f"{r}|{c}|{m}": v["verdict"] for (r, c, m), v in verd_m.items()}
    dfk = RJ.load("finalc", A, "k0", seeds)
    if len(dfk):
        L += ["### B1 on the variant — driver onset PR-AUC@3s vs the k0 heuristic (TEST)", "", "| regime | event | model per seed | k0 per seed | Δ per seed [CI] | verdict |", "|---|---|---|---|---|---|"]
        vk = {}
        for reg in ("manual", "engaged"):
            for ev in RF.EV:
                cell = f"drv_onset|{ev}|3s"; g = dfk[(dfk.regime == reg) & (dfk.cell == cell) & (dfk.metric == "pr_auc")].sort_values("seed")
                if len(g):
                    res = RJ.verdict(g); vk[f"{reg}|{cell}"] = res["verdict"]
                    L.append(f"| {reg} | {ev} | {RF.ms(g.value_a)} | {RF.ms(g.value_b)} | " + " / ".join(f"{float(x.delta):+.3f} [{float(x.delta_ci_lo):+.3f},{float(x.delta_ci_hi):+.3f}]" for x in g.itertuples()) + f" | {res['verdict']} |")
        L.append(""); dec["b1_vs_k0"] = vk
    dfp, body, verd_p, _ = RF.block("finalP", A, B, "B2x", "P cells (eval/pressure_eval.py) — TEST", seeds); L += body + [""]
    dec["P"] = {f"{r}|{c}|{m}": v["verdict"] for (r, c, m), v in verd_p.items()}
    # ------------------------------------------------------------------ 3. read-out gold
    L += ["### 3. Warning read-out — gold clean-test AUROC / PR-AUC (paired on identical device-cluster resamples, n_boot {})".format(a.n_boot), ""]
    gold_ok = None
    if not a.skip_gold:
        Lv, P, ident = C.gold_block(seeds, SPECS, PAIRS, a.n_boot, log); Sm = C.pair_summary(P)
        Lv.to_csv(outd / "gold_levels.csv", index=False); P.to_csv(outd / "gold_pairs.csv", index=False); Sm.to_csv(outd / "gold_summary.csv", index=False)
        L += C.md_gold_table(Lv, Sm, list(SPECS), LABELS) + [""]
        prim = Sm[(Sm.a == "A_W") & (Sm.b == "B_W") & (Sm.stratum == "pooled")] if len(Sm) else Sm
        if len(prim) == 2:
            rules = {r.metric: r.rule for r in prim.itertuples()}
            gold_ok = not any(str(v).startswith("SIGNIFICANTLY") for v in rules.values())
            L += [f"**Gold pooled A_W − B_W:** AUROC rule `{rules.get('auroc')}`, PR-AUC rule `{rules.get('pr_auc')}` -> not WORSE: **{gold_ok}**", ""]
            dec["gold"] = {"pooled_rules": rules, "not_worse": gold_ok, "summary": Sm.to_dict("records")}
        else:
            dec["gold"] = {"status": "NO LAYOUTS"}
        E = ece_block(seeds, SPECS, log=log)
        if len(E):
            E.to_csv(outd / "ece.csv", index=False)
            L += ["ECE (10 equal-width bins, gold clean-test pooled rows; descriptive):", "", "| arm | ECE s0 / s1 / s2 | mean score s0 | base rate |", "|---|---|---|---|"]
            for key in SPECS:
                g = E[E.key == key].sort_values("seed")
                if len(g):
                    L.append(f"| {LABELS[key]} | {' / '.join(f3(x, 4) for x in g.ece)} | {f3(g.mean_score.iloc[0], 3)} | {f3(g.base_rate.iloc[0], 3)} |")
            L.append(""); dec["ece"] = E.to_dict("records")
    # ------------------------------------------------------------------ 4. timing
    L += ["### 4. Event-level alert timing (eval/alert_timing_eval.py; Recall_{>= δ s early} at 10 / 20 / 40 per h; VAL-fitted thresholds primary)", ""]
    tdir = C.find_timing_dir("v2p_dwfckp=", a.timing_dir)
    if tdir is not None and (tdir / "paired.csv").exists():
        Td, TS, TL = C.timing_block(tdir, TIMING_PAIRS, log)
        Td.to_csv(outd / "timing_pairs.csv", index=False); TS.to_csv(outd / "timing_summary.csv", index=False); TL.to_csv(outd / "timing_levels.csv", index=False)
        L += [f"run `results/alert_timing/{tdir.name}`", ""] + C.md_timing_tables(TS, TL, "v2p_dwfckp − kpf_dwfc", LEVEL_SCORES) + [""]
        dec["timing"] = {"dir": tdir.name, "summary": TS.to_dict("records")}
    else:
        L += ["timing run not available", ""]; dec["timing"] = {"status": "MISSING"}
    # ------------------------------------------------------------------ 5. phone / gaze episodes
    L += ["### 5. Phone / object / gaze-away episodes (TEST routes, EVALUATION ONLY): alert share at the VAL-fitted 20 / 40 per h thresholds, window [t − 5, t + 20] s, 60 s cooldown", ""]
    phone_ok = None
    pp = Path(a.phone)
    if pp.exists():
        PH = json.loads(pp.read_text())
        L += [f"{PH['n_episodes']} episodes (results/tables/pitch_wristface_diag.json); thresholds from `{Path(PH['timing_dir']).name}`", "",
              "| score | K/h | alerted s0 / s1 / s2 | mean share | manual / engaged (s0) | by tag (s0) |", "|---|---|---|---|---|---|"]
        for name in PH["scores"]:
            for B_ in (20.0, 40.0):
                rs = [r for r in PH["rows"] if r["score"] == name and r["budget"] == B_]
                if rs:
                    r0 = rs[0]
                    L.append(f"| {name} | {B_:g} | {' / '.join(str(r['n_alerted']) for r in rs)} / {r0['n_episodes']} | {np.mean([r['share'] for r in rs]):.3f} | {r0['share_by_regime']['manual']:.3f} / {r0['share_by_regime']['engaged']:.3f} | "
                             + ", ".join(f"{tg} {v:.2f} (n {r0['n_by_tag'][tg]})" for tg, v in r0["share_by_tag"].items()) + " |")
        L += ["", "| pair (a − b) | K/h | Δ share s0 / s1 / s2 | mean Δ | a-only / b-only episodes per seed | a higher in every seed |", "|---|---|---|---|---|---|"]
        for pr in PH["pairs"]:
            L.append(f"| {pr['a']} − {pr['b']} | {pr['budget']:g} | {' / '.join(f'{d['delta']:+.3f}' for d in pr['per_seed'])} | {pr['mean_delta']:+.3f} | "
                     + "; ".join(f"{d['a_only']}/{d['b_only']}" for d in pr["per_seed"]) + f" | {pr['a_higher_all_seeds']} |")
        prim = {pr["budget"]: pr for pr in PH["pairs"] if pr["a"] == "v2p_dwfckp" and pr["b"] == "kpf_dwfc"}
        if 20.0 in prim and 40.0 in prim:
            phone_ok = prim[20.0]["mean_delta"] > 0 and prim[40.0]["mean_delta"] > 0
            L += ["", f"**A_W − B_W mean Δ share: {prim[20.0]['mean_delta']:+.3f} @20/h, {prim[40.0]['mean_delta']:+.3f} @40/h -> higher at both: {phone_ok}**"]
        L.append(""); dec["phone"] = {"summary": PH["summary"], "pairs": PH["pairs"], "higher_both": phone_ok}
    else:
        L += ["phone-episode evaluation not available", ""]; dec["phone"] = {"status": "MISSING"}
    # ------------------------------------------------------------------ 6. params + GPU minutes
    L += ["### 6. Parameters and GPU minutes (run jsons)", "", "| run | params total | U params | epochs | best | wall s |", "|---|---|---|---|---|---|"]
    gpu_s = 0.0; params = {}
    for fam, pat in (("U0", "U0_kin_can_noK_s{seed}_kcv2p_clip8_ln"), ("I1", RUN_A)):
        for s in seeds:
            rd = CWM3 / "results" / "runs" / pat.format(seed=s); rj = sorted(rd.glob("run_train_*_2*.json"))
            if not rj:
                continue
            R = json.loads(rj[-1].read_text()); P_ = R.get("params", {}); wall = R.get("wall_s"); gpu_s += float(wall or 0)
            hist = (rd / "history.jsonl").read_text().splitlines() if (rd / "history.jsonl").exists() else []
            tot = P_.get("total"); up = (P_.get("U", {}) or {}).get("total") if isinstance(P_.get("U"), dict) else P_.get("total")
            L.append(f"| {rd.name} | {tot} | {up} | {len(hist)} | {R.get('best_epoch')} | {wall} |"); params[rd.name] = {"total": tot, "wall_s": wall, "best_epoch": R.get("best_epoch"), "epochs": len(hist)}
    L += ["", f"GPU wall total (U0 + joint, 3 seeds): {gpu_s / 60:.1f} min", ""]; dec["params"] = params; dec["gpu_minutes"] = gpu_s / 60
    # ------------------------------------------------------------------ verdict
    alt = (gold_ok is True) and (phone_ok is True)
    if joint_ok is None:
        status = "INCOMPLETE (joint pairings missing)"
    elif not joint_ok:
        status = "REJECTED (joint validation gate failed)"
    elif len(paper_beats) >= 1:
        status = "ADOPTION CANDIDATE (joint gate passed; reported driver cell BEATS on validation)"
    elif alt:
        status = "ADOPTION CANDIDATE (joint gate passed; gold not worse AND phone/gaze alert share higher at 20 and 40 per h)"
    elif gold_ok is None or phone_ok is None:
        status = "INCOMPLETE (joint gate passed; read-out / phone endpoints missing)"
    else:
        status = "NOT ACCEPTED (joint gate passed; no reported driver cell BEATS; read-out alternative not met)"
    L += ["### VERDICT (pre-registered rule, one pass)", "", f"**{status}.** joint gate {joint_ok}; paper driver cells BEATS (val) {paper_beats or 'none'}; gold pooled not worse {gold_ok}; phone/gaze share higher at 20 & 40 per h {phone_ok}.  "
          "Adoption, if any, requires the owner's one-shot hold-out; the manuscript is not edited by this lane.", ""]
    dec["verdict"] = {"status": status, "joint_gate": joint_ok, "paper_beats": paper_beats, "gold_not_worse": gold_ok, "phone_higher_both": phone_ok, "sec": round(time.time() - t0, 1)}
    out = Path(a.out); md = "\n".join(L) + "\n"
    if out.exists():
        with open(out, "a") as f:
            f.write("\n---\n\n" + md)
    else:
        out.write_text(md)
    jp = out.with_suffix(".json"); arr = json.loads(jp.read_text()) if jp.exists() else []; arr.append(dec); jp.write_text(json.dumps(arr, indent=1, default=str))
    (outd / "decision.json").write_text(json.dumps(dec, indent=1, default=str))
    print(f"VERDICT: {status}"); print(f"wrote {out} + {jp} (+ {outd})")


if __name__ == "__main__":
    main()

Xet Storage Details

Size:: 21.1 kB
Xet hash:: 091b16bab1fd27878874101fe4184e5d8bbeaabfaa64b214d4a20bec14be9d26

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""cwm3/eval/report_kin_v2p_arm.py — results/tables/kin_v2p_arm.md (+ .json): the kin v2p ADDITIVE INPUT ARM (plan A; owner decision docs/decisions.md row 510;
	pre-registration row 2026-09-18T05:34:40Z) on the kpf joint model. Composes eval/report_joint.py + eval/report_final_face.py blocks (load / render / verdict /
	margin table) over results/paired_compare.csv, eval/alert_pair_common.gold_block (paired gold clean-test AUROC / PR-AUC on identical device-cluster resamples),
	the alert-timing run holding the v2p / kpf read-out scores, the evaluation-only phone / gaze episode alert shares (eval/phone_episode_alerts.py) and the run
	jsons (params, GPU minutes). Nothing is recomputed except the paired gold bootstrap and the 10-bin ECE on the gold rows; every other number is a script output.

	python eval/report_kin_v2p_arm.py [--seeds 0,1,2] [--timing-dir DIR] [--phone results/kin_v2p_arm/phone_alerts/phone_alerts.json] [--n-boot 2000] [--out results/tables/kin_v2p_arm.md]

	ACCEPTANCE (pre-registered, read mechanically): joint model on VALIDATION — no own onset cell (report_joint.OWN) WORSE, no onset cell beyond the −0.005 margin,
	no hazard-NLL cell beyond +0.010; AND EITHER >= 1 of the 4 reported driver cells BEATS on validation, OR (gold pooled AUROC and PR-AUC of A_W vs B_W not WORSE by the
	seed rule AND the phone / gaze alert share of A_W higher than B_W in the 3-seed mean at BOTH 20 and 40 per h). Meeting the rule = ADOPTION CANDIDATE only
	(the owner's one-shot hold-out decides adoption). No openpilot DMS field is read.
	"""
	from __future__ import annotations

	import argparse
	import json
	import sys
	import time
	from pathlib import Path

	import numpy as np
	import pandas as pd

	CWM3 = Path(__file__).resolve().parents[1]
	sys.path.insert(0, str(CWM3 / "eval"))
	import report_joint as RJ # noqa: E402
	import report_final_face as RF # noqa: E402
	import alert_pair_common as C # noqa: E402

	A, B = "allminusecankpfv2p", "allminusecankpf"
	PREREG = "2026-09-18T05:34:40Z"
	PAPER_DRV = [("manual", "drv_onset\|hand_leave\|3s"), ("manual", "drv_onset\|hand_return\|3s"), ("manual", "drv_onset\|head_turn_onset\|3s"), ("engaged", "drv_onset\|head_turn_onset\|3s")]
	VEH4 = ["veh_onset\|gas_on\|3s", "veh_onset\|brake_on\|3s", "veh_onset\|steer_phase_on\|3s", "veh_onset\|veh_any_on\|3s"]
	NLL_VEH = ["hazard_nll\|gas_on", "hazard_nll\|brake_on", "hazard_nll\|steer_phase_on"]
	MARGIN_ONSET_CELLS = RF.DRV + VEH4
	RUN_A, RUN_B = "I1_all_minus_road2drv_s{seed}_final_ecan_kpf_kcv2p", "I1_all_minus_road2drv_s{seed}_final_ecan_kpf"
	COL, COLK = "W_learned_curhgb_t15_w10_n979dwfc", "W_learned_curhgb_t15_w10_n979dwfckp"
	SPECS = {"A_W": (f"W_{RUN_A}_dwfckp", COLK), "B_W": (f"W_{RUN_B}_dwfc", COL), "Ap_W": (f"W_{RUN_B}_dwfckp", COLK)}
	LABELS = {"A_W": "A_W: v2p joint model, cur+dwell+face+kp (HGB)", "B_W": "B_W: kpf joint model (FINAL), cur+dwell+face (paper recipe)", "Ap_W": "A'_W descr.: kpf joint model, cur+dwell+face+kp (read-out-only effect)"}
	PAIRS = [("A_W", "B_W", "A_W − B_W (PRIMARY: v2p model + kp features vs the final model's read-out)", True),
	("Ap_W", "B_W", "A'_W − B_W (descr.: kp features on the unchanged model)", False), ("A_W", "Ap_W", "A_W − A'_W (descr.: model effect at equal features)", False)]
	TIMING_PAIRS = [("v2p_dwfckp", "kpf_dwfc", True), ("kpf_dwfckp", "kpf_dwfc", False), ("v2p_dwfckp", "kpf_dwfckp", False)]
	LEVEL_SCORES = ["kpf_dwfc", "v2p_dwfckp", "kpf_dwfckp", "cur_dwfc", "surrogate_att3s", "persistence"]
	f3, ci_s = C.f3, C.ci_s


	def ece_block(seeds, specs, n_bins=10, log=print):
	"""10-bin equal-width ECE of each spec's score on the gold clean-test rows (pooled), per seed — descriptive."""
	lab = C.LB.episode_labels(); lab = lab[lab.label != "unsure"].reset_index(drop=True)
	packs = {(r, s): C.WG.load_pack(r, s) for r in C.REGIMES for s in ("test", "val")}
	rows = []
	for seed in seeds:
	G = C.CR.gold_for_seed(lab, packs, seed); yg = G.y.to_numpy().astype(int); m0 = C.strata(G)["pooled"]
	for key, (pat, col) in specs.items():
	if not C.layouts_present(pat, seed):
	continue
	s = C.at_gold(G, C.load_col(pat.format(seed=seed), col, packs)); fin = m0 & np.isfinite(s); y, p = yg[fin], np.clip(s[fin], 0, 1)
	edges = np.linspace(0, 1, n_bins + 1); ece = 0.0
	for lo, hi in zip(edges[:-1], edges[1:]):
	mb = (p >= lo) & (p < hi if hi < 1 else p <= hi)
	if mb.any():
	ece += mb.mean() * abs(p[mb].mean() - y[mb].mean())
	rows.append({"key": key, "seed": seed, "ece": float(ece), "n": int(fin.sum()), "mean_score": float(p.mean()), "base_rate": float(y.mean())})
	log(f" ECE s{seed} {key}: {ece:.4f} (n {fin.sum()})")
	return pd.DataFrame(rows)


	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("--seeds", default="0,1,2")
	ap.add_argument("--timing-dir", default="")
	ap.add_argument("--phone", default=str(CWM3 / "results" / "kin_v2p_arm" / "phone_alerts" / "phone_alerts.json"))
	ap.add_argument("--n-boot", type=int, default=2000)
	ap.add_argument("--out", default=str(CWM3 / "results" / "tables" / "kin_v2p_arm.md"))
	ap.add_argument("--skip-gold", action="store_true")
	a = ap.parse_args()
	seeds = [int(s) for s in a.seeds.split(",")]; stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()); t0 = time.time()
	outd = CWM3 / "results" / "kin_v2p_arm" / time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()); outd.mkdir(parents=True, exist_ok=True)
	log = lambda m: print(f"[{time.strftime('%H:%M:%S')}] {m}", flush=True) # noqa: E731
	dec = {"report": stamp, "prereg": PREREG, "seeds": seeds, "A": RUN_A, "B": RUN_B}
	L = [f"# kin v2p ARM — kinematics v2 + 3 keypoint input proxies (head_pitch_kp / wrist_face_kp / hands_lap_kp) on the kpf joint model (pre-registration docs/decisions.md {PREREG}; report {stamp})", "",
	f"A = `{RUN_A}` (U base `U0_kin_can_noK_s<k>_kcv2p_clip8_ln`, 14 input kinematics; targets / states / gold unchanged) vs B = the FROZEN kpf joint model `{RUN_B}`. Pairings eval/run_joint_evals_v2p.sh "
	"(pc_joint_locked / pressure_eval, n_boot 1000, identical rows, v2 causal kin targets = the identical populations); verdict = kill_checks.beats; margins onset −0.005 / NLL +0.010. Read-out: paper recipe "
	"(train_warning_head.py --demand P --feature-sets current --extra-features dwell,face[,kp]); paired gold = alert_pair_common.gold_block on identical device-cluster resamples. Every number is a script output; the manuscript is not edited here.", ""]
	# ------------------------------------------------------------------ 1. pack
	bs = CWM3 / "results" / "targets_kin_causal_routes_v2p" / "build_stats_v2p_latest.json"
	if bs.exists():
	S = json.loads(bs.read_text())
	L += ["### 1. Pack results/targets_kin_causal_routes_v2p (data/build_kin_causal_v2p.py)", "",
	f"shared arrays bit-identical to v2: {S['shared_arrays_identical_to_v2']} ({S['n_shared_arrays_checked']} arrays checked); routes with pose {S['n_routes_with_pose']}/{S['n_routes']}; "
	f"frame coverage: pitch defined {S['frame_coverage']['pitch_defined'] / max(S['frame_coverage']['frames'], 1):.4f}, wrist_face {S['frame_coverage']['wf_defined'] / max(S['frame_coverage']['frames'], 1):.4f}, "
	f"lap {S['frame_coverage']['lap_defined'] / max(S['frame_coverage']['frames'], 1):.4f}; head-down flag share (defined frames) {S['frame_coverage']['headdown'] / max(S['frame_coverage']['pitch_defined'], 1):.4f}", "",
	"\| split \| rows ok \| pitch defined (valid anchors) \| head-down share \| dwell >= 1 s share \| mean pitch dev \| mean wrist_face \| mean hands_lap \|", "\|---\|---\|---\|---\|---\|---\|---\|---\|"]
	for k, st in S["splits"].items():
	L.append(f"\| {k} \| {st['n_row_ok']} \| {f3(st['proxy_defined_given_valid']['head_pitch_kp'], 4)} \| {f3(st['headdown_share_given_defined'], 4)} \| {f3(st['pitch_dwell_ge1s_share'], 4)} \| "
	f"{f3(st['proxy_mean_given_valid']['head_pitch_kp'], 4)} \| {f3(st['proxy_mean_given_valid']['wrist_face_kp'], 3)} \| {f3(st['proxy_mean_given_valid']['hands_lap_kp'], 3)} \|")
	L.append(""); dec["pack"] = {"identical_to_v2": S["shared_arrays_identical_to_v2"], "n_checked": S["n_shared_arrays_checked"]}
	# ------------------------------------------------------------------ 2. joint model
	L += ["### 2. Joint model — v2p arm vs the frozen kpf joint model", ""]
	dfv, body, verd_v, cl_v = RF.block("finalcv", A, B, "B2", "(a-val) v2p vs FROZEN kpf joint model — VALIDATION split (acceptance decision)", seeds); L += body + [""]
	joint_ok = None
	if len(dfv):
	mt, checks_v = RF.margin_table(dfv, verd_v, "(a-val) driver / vehicle cells and hazard-NLL costs with the acceptable-degradation margins (VALIDATION)", MARGIN_ONSET_CELLS, RF.NLL_DRV + NLL_VEH); L += mt
	worse, broke, _, ht, insuff = RF.own_status(verd_v, checks_v)
	paper_beats = [f"{r}\|{c}" for r, c in PAPER_DRV if verd_v.get((r, c, "pr_auc"), {}).get("verdict") == "BEATS"]
	paper_worse = [f"{r}\|{c}" for r, c in PAPER_DRV if verd_v.get((r, c, "pr_auc"), {}).get("verdict") == "WORSE"]
	nll_broke = [k for k, c in checks_v.items() if k[2] == "nll" and not c["ok"]]
	joint_ok = (not worse) and (not broke) and (not nll_broke) and not insuff
	L += [f"(a-val) joint gate: own onset WORSE: {RF.fmt(worse)}; own onset cells beyond −0.005: {RF.fmt(broke)}; NLL cells beyond +0.010: {RF.fmt(nll_broke)}; insufficient: {insuff}; "
	f"paper driver cells BEATS: {paper_beats or 'none'}; WORSE: {paper_worse or 'none'}; head_turn verdicts {ht} -> gate {'PASSED' if joint_ok else 'FAILED'}", ""]
	dec["validation"] = {"gate_passed": joint_ok, "own_worse": RF.fmt(worse), "own_beyond_margin": RF.fmt(broke), "nll_beyond_margin": RF.fmt(nll_broke), "insufficient": insuff, "paper_beats": paper_beats, "paper_worse": paper_worse,
	"cells": {f"{r}\|{c}\|{m}": v["verdict"] for (r, c, m), v in verd_v.items()}, "margin_checks": {f"{r}\|{c}\|{m}": c_ for (r, c, m), c_ in checks_v.items()}}
	else:
	dec["validation"] = {"status": "NO ROWS"}; paper_beats = []
	dft, body, verd_t, cl_t = RF.block("finalc", A, B, "B2", "(a) v2p vs FROZEN kpf joint model — TEST split (confirmation)", seeds); L += body + [""]
	if len(dft):
	mt, checks_t = RF.margin_table(dft, verd_t, "(a) driver / vehicle cells and hazard-NLL costs with the margins (TEST)", MARGIN_ONSET_CELLS, RF.NLL_DRV + NLL_VEH); L += mt
	worse_t, broke_t, _, ht_t, _ = RF.own_status(verd_t, checks_t)
	dec["test"] = {"own_worse": RF.fmt(worse_t), "own_beyond_margin": RF.fmt(broke_t), "head_turn": ht_t, "B2_style": cl_t[0], "cells": {f"{r}\|{c}\|{m}": v["verdict"] for (r, c, m), v in verd_t.items()},
	"paper_cells": {f"{r}\|{c}": verd_t.get((r, c, "pr_auc"), {}).get("verdict", "NO ROWS") for r, c in PAPER_DRV},
	"hazard_nll": {f"{r}\|{c}": verd_t.get((r, c, "nll"), {}).get("verdict", "NO ROWS") for r in ("manual", "engaged") for c in RF.NLL_DRV + NLL_VEH}}
	L += [f"(a) TEST: own onset WORSE: {RF.fmt(worse_t)}; beyond −0.005: {RF.fmt(broke_t)}; head_turn {ht_t}; B2-style claim: {cl_t[0]}.", ""]
	else:
	dec["test"] = {"status": "NO ROWS"}
	dfm, body, verd_m, _ = RF.block("finalcm", A, B, "B2", "masked protocol (human-verified driver-box mask) — TEST", seeds); L += body + [""]
	dec["masked"] = {f"{r}\|{c}\|{m}": v["verdict"] for (r, c, m), v in verd_m.items()}
	dfk = RJ.load("finalc", A, "k0", seeds)
	if len(dfk):
	L += ["### B1 on the variant — driver onset PR-AUC@3s vs the k0 heuristic (TEST)", "", "\| regime \| event \| model per seed \| k0 per seed \| Δ per seed [CI] \| verdict \|", "\|---\|---\|---\|---\|---\|---\|"]
	vk = {}
	for reg in ("manual", "engaged"):
	for ev in RF.EV:
	cell = f"drv_onset\|{ev}\|3s"; g = dfk[(dfk.regime == reg) & (dfk.cell == cell) & (dfk.metric == "pr_auc")].sort_values("seed")
	if len(g):
	res = RJ.verdict(g); vk[f"{reg}\|{cell}"] = res["verdict"]
	L.append(f"\| {reg} \| {ev} \| {RF.ms(g.value_a)} \| {RF.ms(g.value_b)} \| " + " / ".join(f"{float(x.delta):+.3f} [{float(x.delta_ci_lo):+.3f},{float(x.delta_ci_hi):+.3f}]" for x in g.itertuples()) + f" \| {res['verdict']} \|")
	L.append(""); dec["b1_vs_k0"] = vk
	dfp, body, verd_p, _ = RF.block("finalP", A, B, "B2x", "P cells (eval/pressure_eval.py) — TEST", seeds); L += body + [""]
	dec["P"] = {f"{r}\|{c}\|{m}": v["verdict"] for (r, c, m), v in verd_p.items()}
	# ------------------------------------------------------------------ 3. read-out gold
	L += ["### 3. Warning read-out — gold clean-test AUROC / PR-AUC (paired on identical device-cluster resamples, n_boot {})".format(a.n_boot), ""]
	gold_ok = None
	if not a.skip_gold:
	Lv, P, ident = C.gold_block(seeds, SPECS, PAIRS, a.n_boot, log); Sm = C.pair_summary(P)
	Lv.to_csv(outd / "gold_levels.csv", index=False); P.to_csv(outd / "gold_pairs.csv", index=False); Sm.to_csv(outd / "gold_summary.csv", index=False)
	L += C.md_gold_table(Lv, Sm, list(SPECS), LABELS) + [""]
	prim = Sm[(Sm.a == "A_W") & (Sm.b == "B_W") & (Sm.stratum == "pooled")] if len(Sm) else Sm
	if len(prim) == 2:
	rules = {r.metric: r.rule for r in prim.itertuples()}
	gold_ok = not any(str(v).startswith("SIGNIFICANTLY") for v in rules.values())
	L += [f"Gold pooled A_W − B_W: AUROC rule `{rules.get('auroc')}`, PR-AUC rule `{rules.get('pr_auc')}` -> not WORSE: {gold_ok}", ""]
	dec["gold"] = {"pooled_rules": rules, "not_worse": gold_ok, "summary": Sm.to_dict("records")}
	else:
	dec["gold"] = {"status": "NO LAYOUTS"}
	E = ece_block(seeds, SPECS, log=log)
	if len(E):
	E.to_csv(outd / "ece.csv", index=False)
	L += ["ECE (10 equal-width bins, gold clean-test pooled rows; descriptive):", "", "\| arm \| ECE s0 / s1 / s2 \| mean score s0 \| base rate \|", "\|---\|---\|---\|---\|"]
	for key in SPECS:
	g = E[E.key == key].sort_values("seed")
	if len(g):
	L.append(f"\| {LABELS[key]} \| {' / '.join(f3(x, 4) for x in g.ece)} \| {f3(g.mean_score.iloc[0], 3)} \| {f3(g.base_rate.iloc[0], 3)} \|")
	L.append(""); dec["ece"] = E.to_dict("records")
	# ------------------------------------------------------------------ 4. timing
	L += ["### 4. Event-level alert timing (eval/alert_timing_eval.py; Recall_{>= δ s early} at 10 / 20 / 40 per h; VAL-fitted thresholds primary)", ""]
	tdir = C.find_timing_dir("v2p_dwfckp=", a.timing_dir)
	if tdir is not None and (tdir / "paired.csv").exists():
	Td, TS, TL = C.timing_block(tdir, TIMING_PAIRS, log)
	Td.to_csv(outd / "timing_pairs.csv", index=False); TS.to_csv(outd / "timing_summary.csv", index=False); TL.to_csv(outd / "timing_levels.csv", index=False)
	L += [f"run `results/alert_timing/{tdir.name}`", ""] + C.md_timing_tables(TS, TL, "v2p_dwfckp − kpf_dwfc", LEVEL_SCORES) + [""]
	dec["timing"] = {"dir": tdir.name, "summary": TS.to_dict("records")}
	else:
	L += ["timing run not available", ""]; dec["timing"] = {"status": "MISSING"}
	# ------------------------------------------------------------------ 5. phone / gaze episodes
	L += ["### 5. Phone / object / gaze-away episodes (TEST routes, EVALUATION ONLY): alert share at the VAL-fitted 20 / 40 per h thresholds, window [t − 5, t + 20] s, 60 s cooldown", ""]
	phone_ok = None
	pp = Path(a.phone)
	if pp.exists():
	PH = json.loads(pp.read_text())
	L += [f"{PH['n_episodes']} episodes (results/tables/pitch_wristface_diag.json); thresholds from `{Path(PH['timing_dir']).name}`", "",
	"\| score \| K/h \| alerted s0 / s1 / s2 \| mean share \| manual / engaged (s0) \| by tag (s0) \|", "\|---\|---\|---\|---\|---\|---\|"]
	for name in PH["scores"]:
	for B_ in (20.0, 40.0):
	rs = [r for r in PH["rows"] if r["score"] == name and r["budget"] == B_]
	if rs:
	r0 = rs[0]
	L.append(f"\| {name} \| {B_:g} \| {' / '.join(str(r['n_alerted']) for r in rs)} / {r0['n_episodes']} \| {np.mean([r['share'] for r in rs]):.3f} \| {r0['share_by_regime']['manual']:.3f} / {r0['share_by_regime']['engaged']:.3f} \| "
	+ ", ".join(f"{tg} {v:.2f} (n {r0['n_by_tag'][tg]})" for tg, v in r0["share_by_tag"].items()) + " \|")
	L += ["", "\| pair (a − b) \| K/h \| Δ share s0 / s1 / s2 \| mean Δ \| a-only / b-only episodes per seed \| a higher in every seed \|", "\|---\|---\|---\|---\|---\|---\|"]
	for pr in PH["pairs"]:
	L.append(f"\| {pr['a']} − {pr['b']} \| {pr['budget']:g} \| {' / '.join(f'{d['delta']:+.3f}' for d in pr['per_seed'])} \| {pr['mean_delta']:+.3f} \| "
	+ "; ".join(f"{d['a_only']}/{d['b_only']}" for d in pr["per_seed"]) + f" \| {pr['a_higher_all_seeds']} \|")
	prim = {pr["budget"]: pr for pr in PH["pairs"] if pr["a"] == "v2p_dwfckp" and pr["b"] == "kpf_dwfc"}
	if 20.0 in prim and 40.0 in prim:
	phone_ok = prim[20.0]["mean_delta"] > 0 and prim[40.0]["mean_delta"] > 0
	L += ["", f"A_W − B_W mean Δ share: {prim[20.0]['mean_delta']:+.3f} @20/h, {prim[40.0]['mean_delta']:+.3f} @40/h -> higher at both: {phone_ok}"]
	L.append(""); dec["phone"] = {"summary": PH["summary"], "pairs": PH["pairs"], "higher_both": phone_ok}
	else:
	L += ["phone-episode evaluation not available", ""]; dec["phone"] = {"status": "MISSING"}
	# ------------------------------------------------------------------ 6. params + GPU minutes
	L += ["### 6. Parameters and GPU minutes (run jsons)", "", "\| run \| params total \| U params \| epochs \| best \| wall s \|", "\|---\|---\|---\|---\|---\|---\|"]
	gpu_s = 0.0; params = {}
	for fam, pat in (("U0", "U0_kin_can_noK_s{seed}_kcv2p_clip8_ln"), ("I1", RUN_A)):
	for s in seeds:
	rd = CWM3 / "results" / "runs" / pat.format(seed=s); rj = sorted(rd.glob("run_train__2.json"))
	if not rj:
	continue
	R = json.loads(rj[-1].read_text()); P_ = R.get("params", {}); wall = R.get("wall_s"); gpu_s += float(wall or 0)
	hist = (rd / "history.jsonl").read_text().splitlines() if (rd / "history.jsonl").exists() else []
	tot = P_.get("total"); up = (P_.get("U", {}) or {}).get("total") if isinstance(P_.get("U"), dict) else P_.get("total")
	L.append(f"\| {rd.name} \| {tot} \| {up} \| {len(hist)} \| {R.get('best_epoch')} \| {wall} \|"); params[rd.name] = {"total": tot, "wall_s": wall, "best_epoch": R.get("best_epoch"), "epochs": len(hist)}
	L += ["", f"GPU wall total (U0 + joint, 3 seeds): {gpu_s / 60:.1f} min", ""]; dec["params"] = params; dec["gpu_minutes"] = gpu_s / 60
	# ------------------------------------------------------------------ verdict
	alt = (gold_ok is True) and (phone_ok is True)
	if joint_ok is None:
	status = "INCOMPLETE (joint pairings missing)"
	elif not joint_ok:
	status = "REJECTED (joint validation gate failed)"
	elif len(paper_beats) >= 1:
	status = "ADOPTION CANDIDATE (joint gate passed; reported driver cell BEATS on validation)"
	elif alt:
	status = "ADOPTION CANDIDATE (joint gate passed; gold not worse AND phone/gaze alert share higher at 20 and 40 per h)"
	elif gold_ok is None or phone_ok is None:
	status = "INCOMPLETE (joint gate passed; read-out / phone endpoints missing)"
	else:
	status = "NOT ACCEPTED (joint gate passed; no reported driver cell BEATS; read-out alternative not met)"
	L += ["### VERDICT (pre-registered rule, one pass)", "", f"{status}. joint gate {joint_ok}; paper driver cells BEATS (val) {paper_beats or 'none'}; gold pooled not worse {gold_ok}; phone/gaze share higher at 20 & 40 per h {phone_ok}. "
	"Adoption, if any, requires the owner's one-shot hold-out; the manuscript is not edited by this lane.", ""]
	dec["verdict"] = {"status": status, "joint_gate": joint_ok, "paper_beats": paper_beats, "gold_not_worse": gold_ok, "phone_higher_both": phone_ok, "sec": round(time.time() - t0, 1)}
	out = Path(a.out); md = "\n".join(L) + "\n"
	if out.exists():
	with open(out, "a") as f:
	f.write("\n---\n\n" + md)
	else:
	out.write_text(md)
	jp = out.with_suffix(".json"); arr = json.loads(jp.read_text()) if jp.exists() else []; arr.append(dec); jp.write_text(json.dumps(arr, indent=1, default=str))
	(outd / "decision.json").write_text(json.dumps(dec, indent=1, default=str))
	print(f"VERDICT: {status}"); print(f"wrote {out} + {jp} (+ {outd})")


	if __name__ == "__main__":
	main()