Buckets:
| """cwm3/eval/kill_report_cwm3.py — the CWM3 kill / claim report (roadmap E9): one row per cwm2 K1–K10 analogue, rendered from the result CSVs. | |
| Every verdict is computed by cwm2/eval/kill_checks.beats (paired Δ CI excludes 0 in >= 2/3 seeds, same sign in all seeds, |mean Δ| > seed sd) | |
| over rows that already exist in results/paired_compare.csv, results/shift_twin.csv, results/warning_gold/warning_gold.csv and the cwm2 grid | |
| reference file; the only thing this script RUNS is the no-DMS firewall test (pytest tests/test_no_dms_inputs.py, CPU, seconds). Nothing is | |
| recomputed from predictions. Run names / tags are parameters so the same report re-renders on the `_final` runs of the final-assembly agent. | |
| Rows (claim -> rule -> evidence tag(s) -> verdict -> one-line wording): | |
| K1 world-model claim NOT CLAIMED (CWM3 has no rollout arm; direct 25-step decoder, B3 chunked decoder FAILED) | |
| K2 joint != composition NOT CLAIMED (no joint first-passage head; the warning head is a learned read-out of the current state) | |
| K3 vehicle sanity vs CAN-only main model's manual brake / gas / steer_phase onset PR-AUC@3s (value_a of its paired rows, 3 seeds) not below | |
| the cwm2 can_only_direct reference (grid4_v2, 3 seeds) minus 2 sd [reference rows live on the cwm2 grid row subset] | |
| K4 causal inputs (K4) + B1 beats over --k4-input-tag (causal vs old inputs) and --b1-tag (survivor U vs k0 heuristic), 6 driver cells | |
| K5 identity / shifted twins beats over --twin-tag twin_pair_auroc@3s (a gain that survives same-route twins does not ride on identity); | |
| twin-vs-random gap and identity_probe_acc reported | |
| K6 no-DMS lineage every main / base run.json has dms_fields_used == [] and the data-layer file list is DMS-free | |
| K7 demand-head sanity main model's d_onset PR-AUC@3s per family vs the cwm2 current_signal_S0 reference (level), and P onset vs the | |
| margin rule (beats over --p-margin-tag, p_onset cells) | |
| K8 firewall tests pytest tests/test_no_dms_inputs.py (all tests pass) | |
| K9 provenance TRAIN_DONE / PRED_DONE / BEST == layout model_id / config sha / inputs_sha256 listing for every main run | |
| K10 warning head vs DMS att@3s beats over warning_gold paired rows (--warning-column vs legacy_att, clean_test AUROC, --warning-item) | |
| Outputs: results/tables/kill_report.md (+ results/kill_report/kill_report_<label>_<stamp>.json); exit 0 always (INSUFFICIENT rows are reported). | |
| python eval/kill_report_cwm3.py --label pre-final # existing v2 tags | |
| python eval/kill_report_cwm3.py --label final --main-run-prefix I1_all_minus_road2drv_s{k}_final --indep-run-prefix I1_indep_s{k}_final \ | |
| --warning-item cwm3_final --twin-tag final_s{k}_allminus_vs_indep --b1-tag ... --out results/tables/kill_report.md | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import os | |
| import re | |
| import subprocess | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| sys.path.insert(0, str(CWM2 / "eval")) | |
| sys.path.insert(0, str(CWM3 / "data")) | |
| import kill_checks as KC # noqa: E402 | |
| PAIRED = CWM3 / "results" / "paired_compare.csv" | |
| TWIN = CWM3 / "results" / "shift_twin.csv" | |
| WGOLD = CWM3 / "results" / "warning_gold" / "warning_gold.csv" | |
| REF = CWM2 / "results" / "grid" / "metrics.csv" | |
| RUNS = CWM3 / "results" / "runs" | |
| OUT_DIR = CWM3 / "results" / "kill_report" | |
| PY = sys.executable | |
| VEH = ("brake_on", "gas_on", "steer_phase_on") | |
| DRV = ("hand_leave", "hand_return", "head_turn_onset") | |
| FAM = ("steer_rate", "decel", "ttc", "ttc_drop", "lane", "badas") | |
| PFAM = ("ttc", "ttc_drop", "cut_in", "lead_braking", "lane_excursion", "badas") | |
| # ----------------------------------------------------------------------------- helpers | |
| def latest_rows(df, keys): | |
| """keep the latest run_stamp per key (append-only CSVs may hold re-runs).""" | |
| if not len(df): | |
| return df | |
| return df.sort_values("run_stamp").groupby(keys, as_index=False).tail(1) | |
| def beats_rows(g, a_col="value_a", b_col="value_b"): | |
| """kill_checks.beats over per-seed paired rows g (delta / CI / values) -> verdict dict.""" | |
| if g is None or not len(g): | |
| return {"verdict": "INSUFFICIENT", "n_seeds": 0} | |
| pr = pd.DataFrame({"seed": g["seed"].to_numpy(), "delta": pd.to_numeric(g["delta"]), "delta_ci_lo": pd.to_numeric(g["delta_ci_lo"]), | |
| "delta_ci_hi": pd.to_numeric(g["delta_ci_hi"]), "status": "secondary"}) | |
| return KC.beats(pr, pd.DataFrame({"value": pd.to_numeric(g[a_col])}), pd.DataFrame({"value": pd.to_numeric(g[b_col])}), min_seeds=3) | |
| def fmt_d(v): | |
| return "—" if v is None or not np.isfinite(v) else f"{v:+.3f}" | |
| def fmt_v(v): | |
| return "—" if v is None or not np.isfinite(v) else f"{v:.3f}" | |
| def ms(vals): | |
| v = np.asarray([x for x in vals if x is not None and np.isfinite(x)], float) | |
| return "—" if not len(v) else (f"{v.mean():.3f} ± {v.std(ddof=1):.3f}" if len(v) > 1 else f"{v[0]:.3f}") | |
| def expand(pattern, seeds): | |
| return [pattern.format(k=k) for k in seeds] | |
| def load_csv(p): | |
| return pd.read_csv(p, low_memory=False) if Path(p).exists() else pd.DataFrame() | |
| def tag_rows(df, pattern, seeds, cell=None, metric="pr_auc", regime=None): | |
| """rows of the per-seed tags pattern.format(k=k) (adds a `seed` column); latest stamp per (tag, regime, cell, metric).""" | |
| if not len(df): | |
| return df | |
| tags = expand(pattern, seeds) | |
| x = df[df.tag.isin(tags)].copy() | |
| if cell is not None: | |
| x = x[x.cell == cell] | |
| if metric is not None: | |
| x = x[x.metric == metric] | |
| if regime is not None: | |
| x = x[x.regime == regime] | |
| if not len(x): | |
| return x | |
| x["seed"] = x.tag.map({t: k for t, k in zip(tags, seeds)}) | |
| return latest_rows(x, ["tag", "regime", "cell", "metric"]).sort_values("seed") | |
| def run_values(df, run_pattern, seeds, cell, regime, metric="pr_auc"): | |
| """per-seed values of a run as the A side of ANY paired row (value_a; latest stamp) -> DataFrame(seed, value, tag).""" | |
| out = [] | |
| for k in seeds: | |
| r = run_pattern.format(k=k) | |
| x = df[(df.a == r) & (df.cell == cell) & (df.metric == metric) & (df.regime == regime)] | |
| if not len(x): | |
| x = df[(df.b == r) & (df.cell == cell) & (df.metric == metric) & (df.regime == regime)] | |
| if len(x): | |
| x = x.sort_values("run_stamp").tail(1) | |
| out.append({"seed": k, "value": float(x.value_b.iloc[0]), "tag": str(x.tag.iloc[0]), "side": "b"}) | |
| continue | |
| x = x.sort_values("run_stamp").tail(1) | |
| out.append({"seed": k, "value": float(x.value_a.iloc[0]), "tag": str(x.tag.iloc[0]), "side": "a"}) | |
| return pd.DataFrame(out) | |
| # ----------------------------------------------------------------------------- checks | |
| def k3_vehicle(paired, ref, main_pat, seeds, margin_sd=2.0): | |
| """manual brake / gas / steer_phase of the main model vs can_only_direct (cwm2 grid4_v2): PASS iff mean_main >= mean_ref - margin_sd * sd_ref.""" | |
| res = {"rule": f"manual veh_onset PR-AUC@3s of the main model (3-seed mean) >= can_only_direct mean − {margin_sd:g}·sd (cwm2 grid4_v2, 3 seeds)", "cells": {}} | |
| ok_all, insuff = True, False | |
| for reg in ("manual", "engaged"): | |
| for ev in VEH: | |
| mv = run_values(paired, main_pat, seeds, f"veh_onset|{ev}|3s", reg) | |
| rv = ref[(ref.arm == "can_only_direct") & (ref.endpoint == "veh_onset") & (ref.metric == "pr_auc") & (ref.comparator.isna()) & (ref.regime == reg) | |
| & (ref.event_or_family == ev) & np.isclose(ref.tau_s.astype(float), 3.0)] if len(ref) else pd.DataFrame() | |
| if len(ref) and "item" in ref: | |
| rv = rv[rv.item == "grid4_v2"] | |
| rv = rv.drop_duplicates(subset=["seed"]) if len(rv) else rv | |
| cell = {"main": mv.value.tolist() if len(mv) else [], "ref": rv.value.astype(float).tolist() if len(rv) else []} | |
| if len(mv) < 3 or len(rv) < 3: | |
| cell["verdict"] = "INSUFFICIENT" | |
| if reg == "manual": | |
| insuff = True | |
| else: | |
| m, r, sd = float(mv.value.mean()), float(rv.value.astype(float).mean()), float(rv.value.astype(float).std(ddof=1)) | |
| cell.update(main_mean=m, ref_mean=r, ref_sd=sd, threshold=r - margin_sd * sd, verdict="OK" if m >= r - margin_sd * sd else "BELOW") | |
| if reg == "manual" and cell["verdict"] == "BELOW": | |
| ok_all = False | |
| res["cells"][f"{reg}|{ev}"] = cell | |
| res["verdict"] = "INSUFFICIENT" if insuff else ("PASS" if ok_all else "FAIL") | |
| res["note"] = "reference rows come from the cwm2 grid (its own row subset; not a paired delta); engaged rows descriptive" | |
| return res | |
| def k4_causal(paired, in_tag, b1_tag, seeds): | |
| res = {"rule": "K4: causal-ROI inputs vs OLD route-median inputs (both on causal targets) — BEATS per cell; B1: survivor U vs the k0 current-state heuristic — BEATS / NOT WORSE per cell", | |
| "input_effect": {}, "b1_vs_k0": {}} | |
| for reg in ("manual", "engaged"): | |
| for ev in DRV: | |
| cell = f"drv_onset|{ev}|3s" | |
| g = tag_rows(paired, in_tag, seeds, cell=cell, regime=reg) | |
| res["input_effect"][f"{reg}|{ev}"] = dict(beats_rows(g), deltas_fmt=[fmt_d(d) for d in pd.to_numeric(g.delta)] if len(g) else []) | |
| g2 = tag_rows(paired, b1_tag, seeds, cell=cell, regime=reg) | |
| res["b1_vs_k0"][f"{reg}|{ev}"] = dict(beats_rows(g2), a_mean=float(pd.to_numeric(g2.value_a).mean()) if len(g2) else np.nan, | |
| b_mean=float(pd.to_numeric(g2.value_b).mean()) if len(g2) else np.nan) | |
| ie = [v["verdict"] for v in res["input_effect"].values()] | |
| b1 = [v["verdict"] for v in res["b1_vs_k0"].values()] | |
| res["k4_verdict"] = "INSUFFICIENT" if all(v == "INSUFFICIENT" for v in ie) else ("PASS" if ie.count("BEATS") >= 4 and "WORSE" not in ie else ("FLAG" if "WORSE" in ie else "PARTIAL")) | |
| res["b1_verdict"] = "INSUFFICIENT" if all(v == "INSUFFICIENT" for v in b1) else (f"{b1.count('BEATS')}/6 BEATS" + (", WORSE: " + ", ".join(k for k, v in res["b1_vs_k0"].items() if v["verdict"] == "WORSE") if "WORSE" in b1 else "")) | |
| res["verdict"] = res["k4_verdict"] if res["k4_verdict"] in ("INSUFFICIENT", "FLAG") else ("PASS" if "WORSE" not in b1 else "FLAG") | |
| return res | |
| def k5_twins(twin, twin_tag, seeds): | |
| res = {"rule": "twin_pair_auroc@3s (same-route twins share identity; 0.5 = identity-only) A − B: WORSE in any cell -> FLAG; identity_probe_acc A − B > 0.05 -> FLAG", "cells": {}, "identity": {}} | |
| if not len(twin): | |
| res["verdict"] = "INSUFFICIENT" | |
| return res | |
| tags = expand(twin_tag, seeds) | |
| x = twin[twin.tag.isin(tags)].copy() | |
| if not len(x): | |
| res["verdict"] = "INSUFFICIENT" | |
| return res | |
| x["seed"] = x.tag.map({t: k for t, k in zip(tags, seeds)}) | |
| x = latest_rows(x, ["tag", "regime", "event", "statistic"]) | |
| worse, any_ok = [], False | |
| for (reg, ev), g in x[x.statistic == "twin_pair_auroc@3s"].groupby(["regime", "event"]): | |
| g = g.sort_values("seed") | |
| v = beats_rows(g) | |
| rnd = x[(x.statistic == "random_pair_auroc@3s") & (x.regime == reg) & (x.event == ev)] | |
| gap_a = float(pd.to_numeric(rnd.value_a).mean() - pd.to_numeric(g.value_a).mean()) if len(rnd) else np.nan | |
| res["cells"][f"{reg}|{ev}"] = dict(v, twin_a=float(pd.to_numeric(g.value_a).mean()), twin_b=float(pd.to_numeric(g.value_b).mean()), random_minus_twin_a=gap_a, | |
| deltas_fmt=[fmt_d(d) for d in pd.to_numeric(g.delta)]) | |
| any_ok |= v["verdict"] != "INSUFFICIENT" | |
| if v["verdict"] == "WORSE": | |
| worse.append(f"{reg}|{ev}") | |
| idf = x[x.statistic == "identity_probe_acc"] | |
| flag_id = False | |
| for reg, g in idf.groupby("regime"): | |
| d = pd.to_numeric(g.delta) | |
| res["identity"][reg] = {"acc_a": float(pd.to_numeric(g.value_a).mean()), "acc_b": float(pd.to_numeric(g.value_b).mean()), "delta_mean": float(d.mean()), "n_seeds": int(len(g))} | |
| flag_id |= bool(d.mean() > 0.05) | |
| res["worse"] = worse | |
| res["verdict"] = "INSUFFICIENT" if not any_ok else ("FLAG" if worse or flag_id else "PASS") | |
| return res | |
| def k6_lineage(main_runs, indep_runs): | |
| res = {"rule": "run.json dms_fields_used == [] for every main / indep run (and their base runs when base_ckpts.json exists); data-layer file names carry no DMS path", "runs": {}} | |
| ok, found = True, False | |
| for r in main_runs + indep_runs: | |
| rd = RUNS / r | |
| rj = sorted(rd.glob("run_*_latest.json")) if rd.exists() else [] | |
| if not rj: | |
| res["runs"][r] = "run.json missing" | |
| continue | |
| found = True | |
| z = json.loads(rj[-1].read_text()) | |
| fields = z.get("dms_fields_used", None) | |
| bases = {} | |
| bc = rd / "base_ckpts.json" | |
| if bc.exists(): | |
| try: | |
| for name, path in json.loads(bc.read_text()).get("base_ckpts", {}).items(): | |
| bd = Path(path).parent if str(path).endswith(".pt") else Path(path) | |
| bj = sorted(bd.glob("run_*_latest.json")) | |
| bases[name] = json.loads(bj[-1].read_text()).get("dms_fields_used", "?") if bj else "?" | |
| except Exception as e: # noqa: BLE001 | |
| bases["error"] = str(e)[:80] | |
| res["runs"][r] = {"dms_fields_used": fields, "roi": z.get("roi", ""), "kin_source": z.get("kin_source", ""), "bases": bases} | |
| ok &= fields == [] and all(v == [] for v in bases.values() if v != "?") | |
| try: | |
| import cwm3_dataset as ds | |
| bad = [str(p) for reg in ds.REGIMES for sp in ds.SPLITS for p in ds.files(reg, sp).values() if "dms" in str(p).lower() or any(tok in str(p) for tok in ds.FORBIDDEN_KEYS)] | |
| res["data_layer_dms_paths"] = bad | |
| ok &= not bad | |
| except Exception as e: # noqa: BLE001 | |
| res["data_layer_error"] = str(e)[:120] | |
| res["verdict"] = "INSUFFICIENT" if not found else ("PASS" if ok else "FAIL") | |
| return res | |
| def k7_demand(paired, ref, main_pat, p_tag, seeds): | |
| res = {"rule": "E/D chains: main model d_onset PR-AUC@3s (3-seed mean) vs the cwm2 current_signal_S0 reference level per family (below -> listed); P chains: P onset vs the margin rule, beats", | |
| "d_vs_S0": {}, "p_vs_margin": {}} | |
| below, n_d = [], 0 | |
| for reg in ("manual", "engaged"): | |
| for fam in FAM: | |
| mv = run_values(paired, main_pat, seeds, f"d_onset|{fam}|3s", reg) | |
| rv = ref[(ref.arm == "current_signal_S0") & (ref.endpoint == "d_onset") & (ref.metric == "pr_auc") & (ref.comparator.isna()) & (ref.regime == reg) | |
| & (ref.event_or_family == fam) & np.isclose(ref.tau_s.astype(float), 3.0)] if len(ref) else pd.DataFrame() | |
| if len(ref) and "item" in ref: | |
| rv = rv[rv.item == "grid4_v2"] | |
| cell = {"main": mv.value.tolist() if len(mv) else [], "ref": float(rv.value.astype(float).iloc[0]) if len(rv) else np.nan, | |
| "ref_status": str(rv.status.iloc[0]) if len(rv) and "status" in rv else ""} | |
| if len(mv) >= 3 and np.isfinite(cell["ref"]): | |
| n_d += 1 | |
| cell["main_mean"] = float(mv.value.mean()) | |
| cell["verdict"] = "ABOVE" if cell["main_mean"] >= cell["ref"] else "BELOW" | |
| if cell["verdict"] == "BELOW": | |
| below.append(f"{reg}|{fam}") | |
| else: | |
| cell["verdict"] = "INSUFFICIENT" | |
| res["d_vs_S0"][f"{reg}|{fam}"] = cell | |
| n_p = 0 | |
| for reg in ("manual", "engaged"): | |
| for fam in PFAM: | |
| g = tag_rows(paired, p_tag, seeds, cell=f"p_onset|{fam}|3s", regime=reg) | |
| v = beats_rows(g) | |
| n_p += v["verdict"] != "INSUFFICIENT" | |
| res["p_vs_margin"][f"{reg}|{fam}"] = dict(v, a_mean=float(pd.to_numeric(g.value_a).mean()) if len(g) else np.nan, b_mean=float(pd.to_numeric(g.value_b).mean()) if len(g) else np.nan) | |
| res["below_reference"] = below | |
| res["p_beats"] = [k for k, v in res["p_vs_margin"].items() if v["verdict"] == "BEATS"] | |
| res["p_worse"] = [k for k, v in res["p_vs_margin"].items() if v["verdict"] == "WORSE"] | |
| res["verdict"] = "INSUFFICIENT" if (n_d == 0 and n_p == 0) else ("FLAG" if below or res["p_worse"] else "PASS") | |
| res["note"] = "the S0 reference is a single cwm2 grid value on its own row subset (level comparison, not paired); P rows are paired (identical rows)" | |
| return res | |
| def k8_pytest(skip=False): | |
| res = {"rule": "pytest tests/test_no_dms_inputs.py passes (static DMS-token ban, grid10 column rule, no weak/ import, single DMS reader, DMS-free data layer)"} | |
| if skip: | |
| res.update(verdict="NOT_RUN", note="--skip-pytest") | |
| return res | |
| t0 = time.time() | |
| p = subprocess.run([PY, "-m", "pytest", "tests/test_no_dms_inputs.py", "-q", "--color=no", "-p", "no:cacheprovider"], cwd=str(CWM3), capture_output=True, text=True) | |
| tail = re.sub(r"\x1b\[[0-9;]*m", "", (p.stdout.strip().splitlines() or [""])[-1]) | |
| failed = [l.split("::")[-1].split(" ")[0] for l in p.stdout.splitlines() if l.startswith("FAILED")] | |
| m = re.search(r"(\d+) passed", tail) | |
| f = re.search(r"(\d+) failed", tail) | |
| res.update(returncode=p.returncode, summary=tail[:200], failed_tests=failed, n_passed=int(m.group(1)) if m else 0, n_failed=int(f.group(1)) if f else (0 if p.returncode == 0 else -1), wall_s=round(time.time() - t0, 1)) | |
| res["verdict"] = "PASS" if p.returncode == 0 and res["n_passed"] > 0 else "FAIL" | |
| return res | |
| def k9_provenance(main_runs): | |
| res = {"rule": "every main run dir: TRAIN_DONE + PRED_DONE, layout model_id checkpoint == BEST, layout meta.cfg sha == config.json cfg sha, inputs_sha256 listing present", "runs": {}} | |
| sha = lambda o: hashlib.sha256(json.dumps(o, sort_keys=True, default=str).encode()).hexdigest() | |
| ok, found = True, False | |
| for r in main_runs: | |
| rd = RUNS / r | |
| if not rd.exists(): | |
| res["runs"][r] = {"status": "missing"} | |
| continue | |
| found = True | |
| rec = {"TRAIN_DONE": (rd / "TRAIN_DONE").exists(), "PRED_DONE": (rd / "PRED_DONE").exists(), "inputs_sha256": bool(list(rd.glob("inputs_sha256_*.txt"))), | |
| "layouts": {}} | |
| best = (rd / "BEST").read_text().strip() if (rd / "BEST").exists() else "" | |
| cfg_sha = sha(json.load(open(rd / "config.json"))["cfg"]) if (rd / "config.json").exists() else "" | |
| for reg in ("manual", "engaged"): | |
| lay = rd / f"pred_layout_{reg}_test.npz" | |
| if not lay.exists(): | |
| rec["layouts"][reg] = "missing" | |
| ok = False | |
| continue | |
| meta = json.loads(str(np.load(lay, allow_pickle=True)["meta"])) | |
| mid = str(meta.get("model_id", "")) | |
| ck = mid.split(":", 1)[1] if ":" in mid else "" | |
| good = (ck == best or ck == "base") and (sha(meta.get("cfg", {})) == cfg_sha) | |
| rec["layouts"][reg] = {"model_id": mid, "best": best, "cfg_match": sha(meta.get("cfg", {})) == cfg_sha, "source_sha256": str(meta.get("inputs_digest", ""))[:16], | |
| "mtime_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(os.path.getmtime(lay))), "ok": good} | |
| ok &= good | |
| ok &= rec["TRAIN_DONE"] and rec["PRED_DONE"] and rec["inputs_sha256"] | |
| res["runs"][r] = rec | |
| res["verdict"] = "INSUFFICIENT" if not found else ("PASS" if ok else "FAIL") | |
| return res | |
| def k10_warning(w, item, column, comparator, seeds): | |
| res = {"rule": f"clean_test AUROC of {column} vs {comparator} (att@3s): paired Δ (warning_gold, device-cluster bootstrap) -> kill_checks.beats", "item": item, "column": column} | |
| if not len(w): | |
| res["verdict"] = "INSUFFICIENT" | |
| return res | |
| w = w.assign(seed=pd.to_numeric(w.seed, errors="coerce")) # the CSV holds per-seed rows AND 'agg' rows in the seed column | |
| w = w[w.seed.isin(seeds)].sort_values("run_stamp") | |
| x = w[(w.item == item) & (w.arm == "cwm3_W") & (w.event_or_family == column) & (w.metric == "auroc") & (w.stratum == "clean_test")] | |
| pr = x[x.comparator == comparator].drop_duplicates(subset=["seed"], keep="last").sort_values("seed") | |
| va = x[x.comparator.isna()].drop_duplicates(subset=["seed"], keep="last") | |
| vb = w[(w.item == item) & (w.arm == comparator) & (w.metric == "auroc") & (w.stratum == "clean_test") & (w.comparator.isna())].drop_duplicates(subset=["seed"], keep="last") | |
| if len(pr) < 3: | |
| res.update(verdict="INSUFFICIENT", n_seeds=int(len(pr))) | |
| return res | |
| prs = pd.DataFrame({"seed": pr.seed, "delta": pd.to_numeric(pr.delta), "delta_ci_lo": pd.to_numeric(pr.delta_ci_lo), "delta_ci_hi": pd.to_numeric(pr.delta_ci_hi), "status": pr.status.astype(str)}) | |
| v = KC.beats(prs, pd.DataFrame({"value": pd.to_numeric(va.value)}), pd.DataFrame({"value": pd.to_numeric(vb.value)}), min_seeds=3) | |
| res.update(v, a_values=pd.to_numeric(va.value).tolist(), b_values=pd.to_numeric(vb.value).tolist(), deltas_fmt=[f"{d:+.3f} [{lo:+.3f},{hi:+.3f}]" for d, lo, hi in zip(prs.delta, prs.delta_ci_lo, prs.delta_ci_hi)]) | |
| res["wording"] = "outperforms the DMS attention chain" if v["verdict"] == "BEATS" else ("below the DMS attention chain" if v["verdict"] == "WORSE" else "matches the DMS attention chain within the CI (delta, CI reported)") | |
| return res | |
| # ----------------------------------------------------------------------------- render | |
| def wording(k, r): | |
| v = r.get("verdict") | |
| if k == "K1": | |
| return "no world-model claim: CWM3 uses a direct 25-step decoder (the chunked decoder B3 was WORSE on manual aEgo MAE / CRPS)" | |
| if k == "K2": | |
| return "no joint-first-passage claim: the warning head is a learned read-out of the current state (+ P onset + margin rule)" | |
| if k == "K3": | |
| cells = r.get("cells", {}) | |
| man = [f"{ev.split('_on')[0]} {fmt_v(cells.get(f'manual|{ev}', {}).get('main_mean', np.nan))} vs {fmt_v(cells.get(f'manual|{ev}', {}).get('ref_mean', np.nan))}" for ev in VEH] | |
| return ("vehicle chains keep the CAN-only level: " if v == "PASS" else "vehicle chain below CAN-only: ") + "; ".join(man) + " (manual, model vs can_only_direct; reference on the cwm2 row subset)" | |
| if k == "K4": | |
| ie = r.get("input_effect", {}) | |
| n = sum(1 for x in ie.values() if x.get("verdict") == "BEATS") | |
| return f"causal-ROI inputs BEAT the old-input control in {n}/6 driver cells (K4); survivor U vs k0: {r.get('b1_verdict')} (B1)" | |
| if k == "K5": | |
| c = r.get("cells", {}) | |
| w = r.get("worse", []) | |
| gaps = [x.get("random_minus_twin_a", np.nan) for x in c.values()] | |
| g = f"twin-vs-random gap {np.nanmean(gaps):.2f}" if len(gaps) and np.isfinite(np.nanmean(gaps)) else "gap n/a" | |
| return ("no cell loses on same-route twins -> the gains do not ride on identity; " if not w else f"twin AUROC WORSE in {', '.join(w)}; ") + g + " (device shortcut present in both arms, reported as a secondary statistic)" | |
| if k == "K6": | |
| return "every run.json declares dms_fields_used = []; the data layer names no DMS file" if v == "PASS" else "lineage check failed or runs missing" | |
| if k == "K7": | |
| b = r.get("below_reference", []) | |
| return (f"D chains below the current-signal rule on {', '.join(b)}; " if b else "D chains at or above the current-signal rule; ") + \ | |
| f"P onset BEATS the margin rule on {', '.join(r.get('p_beats', [])) or 'none'}" + (f", WORSE on {', '.join(r['p_worse'])}" if r.get("p_worse") else "") + \ | |
| " — honest wording: the learned E/P heads win on dynamics-defined families and lose on state-threshold families" | |
| if k == "K8": | |
| return f"{r.get('n_passed', 0)} firewall tests pass" if v == "PASS" else f"firewall tests: {r.get('summary', v)}" + (f" — failing: {', '.join(r['failed_tests'])}" if r.get("failed_tests") else "") | |
| if k == "K9": | |
| return "all main runs complete; layouts match BEST and config; inputs hashed" if v == "PASS" else "provenance incomplete (see json)" | |
| if k == "K10": | |
| return f"HGB(current state) vs DMS att@3s: Δ {', '.join(r.get('deltas_fmt', []))} -> {r.get('wording', v)}" if r.get("deltas_fmt") else f"warning-head rows missing for item {r.get('item')}" | |
| return str(v) | |
| def render(report, out_md): | |
| lab = report["label"] | |
| L = [f"# CWM3 kill / claim report — {lab} ({report['stamp']})", "", | |
| f"Generated by `eval/kill_report_cwm3.py`; every verdict is `cwm2/eval/kill_checks.beats` over rows already in `results/paired_compare.csv`, `results/shift_twin.csv`, " | |
| f"`results/warning_gold/warning_gold.csv` and the cwm2 grid reference (`can_only_direct`, `current_signal_S0`); the only executed check is the no-DMS pytest. " | |
| f"Main runs: `{report['params']['main_run_prefix']}` (indep `{report['params']['indep_run_prefix']}`, U `{report['params']['u_run_prefix']}`), seeds {report['params']['seeds']}; " | |
| f"tags K4 `{report['params']['k4_input_tag']}` / B1 `{report['params']['b1_tag']}` / twins `{report['params']['twin_tag']}` / P `{report['params']['p_margin_tag']}`; " | |
| f"warning item `{report['params']['warning_item']}` column `{report['params']['warning_column']}`.", "", | |
| "| K | claim | rule | evidence | verdict | wording |", "|---|---|---|---|---|---|"] | |
| C = report["checks"] | |
| P = report["params"] | |
| ev = {"K1": "—", "K2": "—", "K3": f"paired_compare value_a of {P['main_run_prefix']} (veh_onset cells) + cwm2 grid4_v2 can_only_direct", | |
| "K4": f"{P['k4_input_tag']}; {P['b1_tag']}", "K5": f"shift_twin.csv {P['twin_tag']}", "K6": "run_*_latest.json dms_fields_used; cwm3_dataset.files", | |
| "K7": f"paired_compare value_a of {P['main_run_prefix']} (d_onset) + cwm2 current_signal_S0; {P['p_margin_tag']} (p_onset)", "K8": "pytest tests/test_no_dms_inputs.py", | |
| "K9": f"results/runs/{P['main_run_prefix']}/", "K10": f"warning_gold.csv item {P['warning_item']} {P['warning_column']} vs legacy_att"} | |
| claims = {"K1": "world model (rollout beats direct)", "K2": "joint first passage beats composition", "K3": "vehicle chains not below CAN-only", "K4": "causal driver inputs (K4) / U beats k0 (B1)", | |
| "K5": "gains survive same-route twins (identity)", "K6": "no DMS field in the lineage", "K7": "demand / pressure heads vs current-signal rules", "K8": "no-DMS firewall tests", | |
| "K9": "run provenance / layout guard", "K10": "warning head vs DMS att@3s"} | |
| for k in [f"K{i}" for i in range(1, 11)]: | |
| r = C[k] | |
| L.append(f"| {k} | {claims[k]} | {r.get('rule', '')} | {ev[k]} | **{r.get('verdict')}** | {wording(k, r)} |") | |
| # detail sections | |
| L += ["", "## K3 detail (onset PR-AUC@3s; main = 3-seed values from the paired rows; reference = cwm2 grid4_v2 can_only_direct, 3 seeds)", "", "| regime | event | main s0/s1/s2 | main mean | can_only mean ± sd | threshold (−2 sd) | status |", "|---|---|---|---|---|---|---|"] | |
| for key, c in C["K3"].get("cells", {}).items(): | |
| reg, evn = key.split("|") | |
| L.append(f"| {reg} | {evn} | {' / '.join(fmt_v(x) for x in c.get('main', [])) or '—'} | {fmt_v(c.get('main_mean', np.nan))} | {ms(c.get('ref', []))} | {fmt_v(c.get('threshold', np.nan))} | {c.get('verdict')} |") | |
| L += ["", "## K4 detail (driver onset PR-AUC@3s)", "", "| regime | event | input effect Δ s0/s1/s2 | mean Δ | verdict | survivor U | k0 | U − k0 verdict |", "|---|---|---|---|---|---|---|---|"] | |
| for key in C["K4"].get("input_effect", {}): | |
| a, b = C["K4"]["input_effect"][key], C["K4"]["b1_vs_k0"][key] | |
| reg, evn = key.split("|") | |
| L.append(f"| {reg} | {evn} | {' / '.join(a.get('deltas_fmt', [])) or '—'} | {fmt_d(a.get('mean_delta', np.nan))} | {a.get('verdict')} | {fmt_v(b.get('a_mean', np.nan))} | {fmt_v(b.get('b_mean', np.nan))} | {b.get('verdict')} |") | |
| L += ["", "## K5 detail (shifted twins)", "", "| regime | event | twin AUROC A | twin AUROC B | Δ s0/s1/s2 | verdict | random − twin (A) |", "|---|---|---|---|---|---|---|"] | |
| for key, c in C["K5"].get("cells", {}).items(): | |
| reg, evn = key.split("|") | |
| L.append(f"| {reg} | {evn} | {fmt_v(c.get('twin_a'))} | {fmt_v(c.get('twin_b'))} | {' / '.join(c.get('deltas_fmt', []))} | {c.get('verdict')} | {fmt_v(c.get('random_minus_twin_a'))} |") | |
| for reg, c in C["K5"].get("identity", {}).items(): | |
| L.append(f"| {reg} | all | identity probe acc A {fmt_v(c['acc_a'])} | B {fmt_v(c['acc_b'])} | Δ {fmt_d(c['delta_mean'])} | {'FLAG' if c['delta_mean'] > 0.05 else 'ok'} | — |") | |
| L += ["", "## K7 detail", "", "| regime | family | main d_onset s0/s1/s2 | mean | current_signal_S0 | status |", "|---|---|---|---|---|---|"] | |
| for key, c in C["K7"].get("d_vs_S0", {}).items(): | |
| reg, fam = key.split("|") | |
| L.append(f"| {reg} | {fam} | {' / '.join(fmt_v(x) for x in c.get('main', [])) or '—'} | {fmt_v(c.get('main_mean', np.nan))} | {fmt_v(c.get('ref', np.nan))} {('(' + c['ref_status'] + ')') if c.get('ref_status') and c['ref_status'] != 'sanity' else ''} | {c.get('verdict')} |") | |
| L += ["", "| regime | P family | P onset (model) | margin rule | mean Δ | verdict |", "|---|---|---|---|---|---|"] | |
| for key, c in C["K7"].get("p_vs_margin", {}).items(): | |
| reg, fam = key.split("|") | |
| L.append(f"| {reg} | {fam} | {fmt_v(c.get('a_mean', np.nan))} | {fmt_v(c.get('b_mean', np.nan))} | {fmt_d(c.get('mean_delta', np.nan))} | {c.get('verdict')} |") | |
| L += ["", "## K10 detail", "", f"- item `{C['K10'].get('item')}` column `{C['K10'].get('column')}`: A (per seed) {', '.join(fmt_v(x) for x in C['K10'].get('a_values', []))}; " | |
| f"legacy att@3s {', '.join(fmt_v(x) for x in C['K10'].get('b_values', []))}; Δ {', '.join(C['K10'].get('deltas_fmt', [])) or '—'} -> {C['K10'].get('verdict')}"] | |
| L += ["", "## K8 / K9 detail", "", f"- pytest: {C['K8'].get('summary', C['K8'].get('verdict'))}", | |
| "- provenance per run: " + "; ".join(f"{r}: " + ("missing" if v.get("status") == "missing" else f"TRAIN_DONE {v['TRAIN_DONE']}, PRED_DONE {v['PRED_DONE']}, inputs_sha256 {v['inputs_sha256']}, layouts " + ", ".join(f"{reg} {'ok' if isinstance(l, dict) and l['ok'] else l if not isinstance(l, dict) else 'MISMATCH'}" for reg, l in v["layouts"].items())) for r, v in C["K9"].get("runs", {}).items())] | |
| L += ["", f"Source files: `{PAIRED.relative_to(CWM3)}`, `{TWIN.relative_to(CWM3)}`, `{WGOLD.relative_to(CWM3)}`, `{REF}`; json: `{report['json']}`.", ""] | |
| Path(out_md).write_text("\n".join(L)) | |
| return "\n".join(L) | |
| def main(argv=None): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--label", default="pre-final") | |
| ap.add_argument("--seeds", default="0,1,2") | |
| ap.add_argument("--main-run-prefix", default="I1_all_minus_road2drv_s{k}_v2") | |
| ap.add_argument("--indep-run-prefix", default="I1_indep_s{k}_v2") | |
| ap.add_argument("--u-run-prefix", default="U0_kin_can_noK_s{k}_kcv2_clip8_ln") | |
| ap.add_argument("--k4-input-tag", default="K4v2in_s{k}_causal_vs_oldin") | |
| ap.add_argument("--b1-tag", default="K4v2c_s{k}_clip8ln_vs_k0") | |
| ap.add_argument("--twin-tag", default="I1_s{k}_all_vs_indep") | |
| ap.add_argument("--p-margin-tag", default="exp3P_s{k}_pooledP_vs_margin") | |
| ap.add_argument("--warning-item", default="cwm3_v1") | |
| ap.add_argument("--warning-column", default="W_learned_curhgb_t15_w10_n979") | |
| ap.add_argument("--warning-comparator", default="legacy_att") | |
| ap.add_argument("--paired", default=str(PAIRED)) | |
| ap.add_argument("--twin", default=str(TWIN)) | |
| ap.add_argument("--wgold", default=str(WGOLD)) | |
| ap.add_argument("--ref", default=str(REF)) | |
| ap.add_argument("--out", default=str(CWM3 / "results" / "tables" / "kill_report.md")) | |
| ap.add_argument("--json-dir", default=str(OUT_DIR)) | |
| ap.add_argument("--skip-pytest", action="store_true") | |
| a = ap.parse_args(argv) | |
| seeds = [int(s) for s in a.seeds.split(",")] | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| paired, twin, w, ref = load_csv(a.paired), load_csv(a.twin), load_csv(a.wgold), load_csv(a.ref) | |
| main_runs, indep_runs = expand(a.main_run_prefix, seeds), expand(a.indep_run_prefix, seeds) | |
| Path(a.json_dir).mkdir(parents=True, exist_ok=True) | |
| jpath = Path(a.json_dir) / f"kill_report_{a.label}_{stamp}.json" | |
| report = {"label": a.label, "stamp": stamp, "params": {k: v for k, v in vars(a).items()}, "json": str(jpath), "checks": {}} | |
| report["params"]["seeds"] = seeds | |
| C = report["checks"] | |
| C["K1"] = {"verdict": "NOT_CLAIMED", "rule": "rollout_coupled beats direct_multitask on primary cells — no rollout arm exists in CWM3"} | |
| C["K2"] = {"verdict": "NOT_CLAIMED", "rule": "joint first-passage beats own product composition — no joint first-passage head in CWM3"} | |
| C["K3"] = k3_vehicle(paired, ref, a.main_run_prefix, seeds) | |
| C["K4"] = k4_causal(paired, a.k4_input_tag, a.b1_tag, seeds) | |
| C["K5"] = k5_twins(twin, a.twin_tag, seeds) | |
| C["K6"] = k6_lineage(main_runs, indep_runs) | |
| C["K7"] = k7_demand(paired, ref, a.main_run_prefix, a.p_margin_tag, seeds) | |
| C["K8"] = k8_pytest(a.skip_pytest) | |
| C["K9"] = k9_provenance(main_runs) | |
| C["K10"] = k10_warning(w, a.warning_item, a.warning_column, a.warning_comparator, seeds) | |
| jpath.write_text(json.dumps(report, indent=1, default=lambda o: o.tolist() if hasattr(o, "tolist") else str(o))) | |
| md = render(report, a.out) | |
| print("\n".join(l for l in md.splitlines() if l.startswith("| K"))) | |
| print(f"\nwritten {a.out} and {jpath}") | |
| return report | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 34.1 kB
- Xet hash:
- a9102cd3c3c17359a23bba8368263ac2d2a256f7cf2acb7beed7318422f7ffa6
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.