1"""Build every empirical table and figure from raw records. Each output gets a provenance entry (source run files + script)."""2from __future__ import annotations3import json, sys, glob4from pathlib import Path5import numpy as np, pandas as pd6import matplotlib7matplotlib.use("Agg")8import matplotlib.pyplot as plt9from ..util import ROOT, read_json, write_json, read_jsonl, utc_now10from .load import load_experiments, LABELS11from .stats import cluster_bootstrap_mean, paired_difference, holm, summarize_latency1213# fixed hue per method family (colorblind-validated palette; see results/figures/PALETTE.txt)14PAL = {"A": "#2a78d6", "B_quant": "#eb6834", "B_sel": "#1baf7a", "B_ref": "#6250d6", "C": "#e87ba4", "A_struct": "#eda100", "neg": "#e34948", "hosted": "#008300"}15METHOD_COLOR = {"A0": "#7a7a78", "A1": "#2a78d6", "A2": "#2a78d6", "A3": "#2a78d6", "A4": "#2a78d6", "A5": "#eda100", "A6": "#eda100", "A7": "#2a78d6", "A8": "#2a78d6",16 "B0": "#6250d6", "B1": "#6250d6", "B2": "#eb6834", "B3": "#1baf7a", "B4": "#e34948", "B5": "#1baf7a", "B6": "#1baf7a", "B9": "#eb6834", "C0": "#e34948", "C2": "#e87ba4", "C3": "#e87ba4"}17MARKER = {"A0": "x", "A1": "o", "A2": "v", "A3": "^", "A4": "s", "A5": "D", "A6": "P", "A7": "o", "A8": "<", "B0": "o", "B1": "o", "B2": "s", "B3": "^", "B4": "x", "B5": "D", "B6": "P", "B9": "*", "C0": "x", "C2": "s", "C3": "D"}18TAB = ROOT / "results/tables"; FIG = ROOT / "results/figures"19PROV: dict = {}2021def _prov(path: Path, sources: list[str], note: str = ""):22 PROV[str(path.relative_to(ROOT))] = {"sources": sources, "script": "src/wp30/analysis/analyze.py", "generated_utc": utc_now(), "note": note}2324def _save_fig(fig, name: str, sources: list[str], note: str = ""):25 FIG.mkdir(parents=True, exist_ok=True)26 for ext in ("pdf", "png"):27 p = FIG / f"{name}.{ext}"; fig.savefig(p, bbox_inches="tight", dpi=160 if ext == "png" else None)28 _prov(p, sources, note)29 plt.close(fig)3031def _style():32 plt.rcParams.update({"font.size": 9, "axes.spines.top": False, "axes.spines.right": False, "axes.grid": True, "grid.alpha": 0.25, "grid.linewidth": 0.5,33 "legend.frameon": False, "figure.dpi": 110, "axes.titlesize": 9.5, "axes.labelsize": 9})3435def fmt_bytes(b):36 return f"{b/1e6:.1f} MB" if b >= 1e6 else (f"{b/1e3:.1f} kB" if b >= 1e3 else f"{b:.0f} B")3738# ------------------------------------------------------------------------------------------------------------- tables39def table_primary(df: pd.DataFrame, exp: str, sources: list[str]) -> pd.DataFrame:40 rows = []41 for (fam, pair, cond), g in df.groupby(["task_family", "pair", "condition"]):42 q = cluster_bootstrap_mean(g, "quality_operational"); ok = (g["status"] == "success").mean()43 rows.append({"family": fam, "pair": pair, "condition": cond, "method": g["method_id"].iloc[0], "access": g["question_access"].iloc[0], "n": q["n"], "n_docs": q["n_clusters"],44 "quality": q["mean"], "q_lo": q["lo"], "q_hi": q["hi"], "success_rate": ok, "payload_bytes_median": g["serialized_payload_bytes"].median(),45 "cache_bytes_median": g["cache_tensor_bytes"].median(), "cold_ms_p50": g["cost_cold_ms"].median(), "warm_ms_p50": g["cost_warm_query_ms"].median(),46 "amortized_ms_p50": g["cost_amortized_ms"].median(), "ttft_ms_p50": g["ttft_ms"].median(), "peak_decode_mem_median": g["peak_decode_memory"].median(),47 "fetched_bytes_mean": g["fetched_bytes"].mean()})48 t = pd.DataFrame(rows).sort_values(["family", "pair", "condition"])49 TAB.mkdir(parents=True, exist_ok=True); p = TAB / f"{exp}_primary.csv"; t.to_csv(p, index=False); _prov(p, sources, "per-condition quality (cluster bootstrap CI) and resource medians")50 return t5152def table_failures(df: pd.DataFrame, exp: str, sources):53 t = df.groupby(["method_id", "status"]).size().unstack(fill_value=0)54 p = TAB / f"{exp}_failures.csv"; t.to_csv(p); _prov(p, sources, "attempt status counts by method"); return t5556def table_qtype(df: pd.DataFrame, exp: str, sources):57 d = df[df["task_family"] == "synthetic"]58 rows = []59 for (cond, qt), g in d.groupby(["condition", "qtype"]):60 q = cluster_bootstrap_mean(g, "quality_operational", n_boot=500); rows.append({"condition": cond, "qtype": qt, "quality": q["mean"], "lo": q["lo"], "hi": q["hi"], "n": q["n"]})61 t = pd.DataFrame(rows); p = TAB / f"{exp}_by_question_type.csv"; t.to_csv(p, index=False); _prov(p, sources, "synthetic accuracy by question type"); return t6263def primary_tests(df: pd.DataFrame, exp: str, sources, tests: list[dict]):64 rows = []65 for t in tests:66 sub = df[df["task_family"].isin(t["families"])] if t.get("families") else df67 if t.get("pair"): sub = sub[sub["pair"] == t["pair"]]68 r = paired_difference(sub, t.get("value", "quality_operational"), "condition", t["a"], t["b"], keys=("document_id", "question_id", "pair"))69 r.update({"name": t["name"], "hypothesis": t.get("hypothesis", "")}); rows.append(r)70 out = pd.DataFrame(rows)71 if len(out):72 out["p_holm"] = holm([p if not np.isnan(p) else 1.0 for p in out["p"]])73 p = TAB / f"{exp}_primary_tests.csv"; out.to_csv(p, index=False); _prov(p, sources, "registered primary comparisons, paired, cluster bootstrap, Holm"); return out7475# ------------------------------------------------------------------------------------------------------------ figures76def fig_quality_vs_bytes(df: pd.DataFrame, exp: str, sources, access_filter=None, name="fig2_quality_vs_bytes"):77 _style()78 fams = [f for f in ["synthetic", "exec", "public_qasper", "public_2wikimqa"] if f in set(df["task_family"])]79 pairs = sorted(df["pair"].unique())80 fig, axes = plt.subplots(len(pairs), len(fams), figsize=(3.1 * len(fams), 2.7 * len(pairs)), squeeze=False, sharey="row")81 for i, pair in enumerate(pairs):82 for j, fam in enumerate(fams):83 ax = axes[i][j]; d = df[(df["pair"] == pair) & (df["task_family"] == fam)]84 if access_filter: d = d[d["question_access"].isin(access_filter)]85 for cond, g in d.groupby("condition"):86 m = g["method_id"].iloc[0]; q = cluster_bootstrap_mean(g, "quality_operational", n_boot=300)87 x = g["serialized_payload_bytes"].median()88 if m == "A0" or x is None or np.isnan(x): continue89 ax.errorbar(max(x, 1), q["mean"], yerr=[[q["mean"] - q["lo"]], [q["hi"] - q["mean"]]], fmt=MARKER.get(m, "o"), color=METHOD_COLOR.get(m, "#444"), ms=4.5, lw=0.9, capsize=1.5, mfc="white" if m.startswith("B") else None)90 ax.set_xscale("log"); ax.set_title(f"{fam} · {pair}", fontsize=8.5); ax.set_ylim(-0.03, 1.03)91 if i == len(pairs) - 1: ax.set_xlabel("serialized payload bytes (log)")92 if j == 0: ax.set_ylabel("quality (acc. or F1)")93 handles = [plt.Line2D([], [], marker=MARKER[m], color=METHOD_COLOR[m], ls="", ms=5, mfc="white" if m.startswith("B") else None, label=LABELS[m]) for m in ["A1", "A2", "A3", "A4", "A5", "A6", "A7", "A8", "B0", "B1", "B2", "B3", "B4", "B5", "B6", "B9", "C0", "C2", "C3"] if m in set(df["method_id"])]94 fig.legend(handles=handles, loc="lower center", ncol=5, fontsize=7, bbox_to_anchor=(0.5, -0.06 - 0.02 * (len(handles) // 5)))95 fig.suptitle("Quality vs measured serialized payload bytes (medians; 95% cluster-bootstrap CIs)", fontsize=9.5, y=1.0)96 _save_fig(fig, name, sources, "points = conditions; x = median payload bytes; y = mean quality")9798def fig_quality_vs_latency(df: pd.DataFrame, exp: str, sources, name="fig3_quality_vs_latency_memory"):99 _style()100 fams = [f for f in ["synthetic", "exec", "public_qasper", "public_2wikimqa"] if f in set(df["task_family"])]101 fig, axes = plt.subplots(2, len(fams), figsize=(3.1 * len(fams), 5.2), squeeze=False)102 for j, fam in enumerate(fams):103 d = df[df["task_family"] == fam]104 for cond, g in d.groupby("condition"):105 m = g["method_id"].iloc[0]; q = g["quality_operational"].mean()106 if m == "A0": continue107 axes[0][j].scatter(g["cost_cold_ms"].median(), q, marker=MARKER.get(m, "o"), color=METHOD_COLOR.get(m, "#444"), s=22, facecolors="white" if m.startswith("B") else METHOD_COLOR.get(m, "#444"), lw=0.9)108 axes[1][j].scatter(g["peak_decode_memory"].median() / 1e9, q, marker=MARKER.get(m, "o"), color=METHOD_COLOR.get(m, "#444"), s=22, facecolors="white" if m.startswith("B") else METHOD_COLOR.get(m, "#444"), lw=0.9)109 axes[0][j].set_xscale("log"); axes[0][j].set_title(fam, fontsize=8.5); axes[0][j].set_xlabel("cold end-to-end latency, ms (log)"); axes[1][j].set_xlabel("peak MLX memory during trial, GB")110 axes[0][j].set_ylim(-0.03, 1.03); axes[1][j].set_ylim(-0.03, 1.03)111 axes[0][0].set_ylabel("quality"); axes[1][0].set_ylabel("quality")112 fig.suptitle("Quality vs cold latency (top) and peak memory (bottom); all model pairs pooled", fontsize=9.5)113 _save_fig(fig, name, sources)114115def fig_latency_decomposition(df: pd.DataFrame, exp: str, sources, name="fig4_latency_decomposition"):116 _style()117 d = df[(df["task_family"] == "synthetic") & (df["status"] == "success")]118 conds = [c for c in ["A1", "A4|budget_tokens=512|blind", "A5|budget_tokens=512|blind", "A7|codec=zstd", "B0", "B1", "B2|bits=8", "B3|keep_frac=0.25", "B5|keep_frac=0.25|aware"] if c in set(d["condition"])]119 comps = [("span_encode_ms", "sender encode (text)"), ("span_doc_prefill_ms", "document prefill (cache)"), ("span_compress_ms", "compress/select"), ("span_serialize_ms", "serialize"), ("span_load_ms", "deserialize/load"), ("span_receiver_prefill_ms", "receiver prefill"), ("ttft_ms", "query prefill→first token"), ("span_decode_ms", "decode")]120 cols = ["#2a78d6", "#6250d6", "#1baf7a", "#eda100", "#e87ba4", "#eb6834", "#008300", "#7a7a78"]121 fig, ax = plt.subplots(figsize=(7.2, 3.2))122 for i, c in enumerate(conds):123 g = d[d["condition"] == c]; left = 0124 for (col, lab), color in zip(comps, cols):125 v = g[col].median() if col in g and g[col].notna().any() else 0.0126 if v and v > 0:127 ax.barh(i, v, left=left, color=color, height=0.6, label=lab if i == 0 else None, edgecolor="white", linewidth=0.8); left += v128 ax.set_yticks(range(len(conds))); ax.set_yticklabels([c.replace("|", " ") for c in conds], fontsize=7.5); ax.invert_yaxis(); ax.set_xlabel("median span duration, ms (serial accounting; one question per document build)")129 ax.legend(fontsize=7, ncol=4, loc="lower right"); ax.set_title("Latency decomposition — synthetic, cold scenario (document build counted once)", fontsize=9)130 _save_fig(fig, name, sources)131132def fig_handoffs(df: pd.DataFrame, exp: str, sources, name="fig5_handoffs"):133 _style()134 d = df[df["topology"].astype(str).str.startswith("chain")]135 if d.empty: return136 fig, axes = plt.subplots(1, 3, figsize=(9.6, 2.8))137 for cid, g in d.groupby("run_group_id"):138 pass139 d["chain"] = d["run_group_id"].str.split(":").str[1]140 for chain, g in d.groupby("chain"):141 rep = g["method_id"].iloc[0]; mode = g["topology"].iloc[0].replace("chain_", "")142 color = "#eda100" if rep == "A5" else "#2a78d6"; ls = "-" if mode == "closed" else "--"143 hop = g.groupby("handoff_index")144 ret = hop["retention_rate"].mean(); uns = hop["unsupported"].mean(); acc = hop["quality_operational"].mean()145 axes[0].plot(ret.index, ret.values, ls, color=color, marker="o", ms=4, label=f"{rep} {mode}")146 axes[1].plot(uns.index, uns.values, ls, color=color, marker="o", ms=4)147 axes[2].plot(acc.index, acc.values, ls, color=color, marker="o", ms=4)148 for ax, t in zip(axes, ["retained correct facts (ledger)", "unsupported numbers + identifiers per message", "question accuracy from message"]):149 ax.set_xlabel("handoffs"); ax.set_title(t, fontsize=8.5); ax.set_xticks(sorted(d["handoff_index"].unique()))150 axes[0].set_ylim(0, 1.02); axes[2].set_ylim(0, 1.02); axes[0].legend(fontsize=7)151 _save_fig(fig, name, sources, "means over documents; chains defined in configs/chains*.yaml")152153def fig_breakeven(net_rows: list[dict], exp: str, sources, name="fig6_breakeven"):154 """Measured two-node transfer vs remote text prefill; modeled curves from measured bandwidth."""155 _style()156 if not net_rows: return157 n = pd.DataFrame(net_rows)158 fig, axes = plt.subplots(1, 2, figsize=(8.4, 3.0))159 ax = axes[0]160 for (meth, bits, codec), g in n[n["method"] != "local_prefill"].groupby(["method", "bits", "codec"], dropna=False):161 if meth == "text_remote_prefill":162 tot = g["roundtrip_ms"]; lab = "text → remote prefill"; color = "#2a78d6"; mk = "o"163 else:164 tot = g["roundtrip_ms"]; lab = f"{meth}" + (f" {int(bits)}-bit" if pd.notna(bits) else "") + (f" +{codec}" if codec != "none" else ""); color = "#eb6834" if pd.notna(bits) else "#6250d6"; mk = "s" if pd.notna(bits) else "D"165 gg = g.groupby("prefix_tokens")["roundtrip_ms"].median()166 ax.plot(gg.index, gg.values, marker=mk, ms=4, color=color, lw=1, label=lab, alpha=0.9 if codec == "none" else 0.55, ls="-" if codec == "none" else ":")167 ax.set_xlabel("context tokens"); ax.set_ylabel("remote answer round trip, ms (measured)"); ax.set_title("Measured: M3U96a → M3U96b, 1 GbE", fontsize=8.5); ax.legend(fontsize=6.5)168 ax = axes[1]169 bw = n[n["method"].isin(["B1", "B2"])]["effective_gbps"].median()170 L = np.linspace(500, 32000, 50)171 for bits, bpp in [(16, 2), (8, 1.07), (4, 0.57)]:172 pass173 # bytes per token measured from B1 rows (Qwen3B)174 b1 = n[(n["method"] == "B1") & (n["codec"] == "none")]175 bpt = (b1["serialized_bytes"] / b1["prefix_tokens"]).median() if len(b1) else np.nan176 tr = n[n["method"] == "text_remote_prefill"]177 if len(tr) and not np.isnan(bpt):178 slope = (tr["remote_prefill_ms"] / tr["prefix_tokens"]).median()179 for frac, lab, color in [(1.0, "bf16 cache", "#6250d6"), (0.53, "8-bit", "#eb6834"), (0.28, "4-bit", "#eb6834")]:180 for bw_g, ls in [(bw, "-"), (10.0, "--")]:181 t_ms = L * bpt * frac * 8 / (bw_g * 1e9) * 1000182 ax.plot(L, t_ms, ls, color=color, lw=1, alpha=0.9 if ls == "-" else 0.5, label=f"{lab} transfer @ {bw_g:.1f} Gb/s" if True else None)183 ax.plot(L, L * slope, "-", color="#2a78d6", lw=1.5, label="remote text prefill (measured slope)")184 ax.set_xlabel("context tokens"); ax.set_ylabel("ms"); ax.set_title("Modeled from measured bandwidth and prefill slope", fontsize=8.5); ax.legend(fontsize=6, ncol=2)185 _save_fig(fig, name, sources, "left = measured; right = analytic model (dashed 10 Gb/s is hypothetical)")186187# ------------------------------------------------------------------------------------------------------------ main188def main(protocol_path: str) -> int:189 _style()190 prot = read_json(ROOT / protocol_path) if (ROOT / protocol_path).exists() else {}191 core_exp = prot.get("core_experiment_ids", ["E1_core"]) if prot else ["E1_core"]192 exps = {p.stem: p.name for p in (ROOT / "runs/raw").glob("*.jsonl")}193 (FIG).mkdir(parents=True, exist_ok=True); (TAB).mkdir(parents=True, exist_ok=True)194 (FIG / "PALETTE.txt").write_text("categorical palette validated with dataviz validate_palette.js (light): #2a78d6,#eb6834,#1baf7a,#eda100,#e87ba4,#008300,#6250d6,#e34948 — ALL CHECKS PASS\n")195 # choose the experiment set to analyze: core if present, else pilot196 main_exp = "E1_core" if "E1_core" in exps else ("E1_pilot" if "E1_pilot" in exps else None)197 if main_exp:198 src = [f"runs/raw/{main_exp}.jsonl"]199 df = load_experiments([f"{main_exp}.jsonl"])200 table_primary(df, main_exp, src); table_failures(df, main_exp, src); table_qtype(df, main_exp, src)201 tests = [202 {"name": "T1 A5 vs A4 (512, blind, synthetic)", "a": "A5|budget_tokens=512|blind", "b": "A4|budget_tokens=512|blind", "families": ["synthetic"], "hypothesis": "RQ5/H1"},203 {"name": "T2 B2-8bit vs B0", "a": "B2|bits=8", "b": "B0", "hypothesis": "H2"},204 {"name": "T3 B2-4bit vs B0", "a": "B2|bits=4", "b": "B0", "hypothesis": "H2"},205 {"name": "T4 B5-aware vs B3 (1/4)", "a": "B5|keep_frac=0.25|aware", "b": "B3|keep_frac=0.25", "hypothesis": "H3"},206 {"name": "T6 A1 vs B0 (text vs native cache, same content)", "a": "A1", "b": "B0", "hypothesis": "consistency"},207 ]208 primary_tests(df, main_exp, src, tests)209 fig_quality_vs_bytes(df, main_exp, src)210 fig_quality_vs_bytes(df[df["question_access"].isin(["blind", "none"])], main_exp, src, name="figA_quality_vs_bytes_blind")211 fig_quality_vs_bytes(df[df["question_access"] == "aware"], main_exp, src, name="figA_quality_vs_bytes_aware")212 fig_quality_vs_latency(df, main_exp, src); fig_latency_decomposition(df, main_exp, src)213 chain_exp = "E4_core" if "E4_core" in exps else ("E4_pilot" if "E4_pilot" in exps else None)214 if chain_exp:215 dfc = load_experiments([f"{chain_exp}.jsonl"]); src = [f"runs/raw/{chain_exp}.jsonl"]216 fig_handoffs(dfc, chain_exp, src)217 rows = []218 for (grp, h), g in dfc.groupby(["run_group_id", "handoff_index"]):219 q = cluster_bootstrap_mean(g, "quality_operational", n_boot=500); r = cluster_bootstrap_mean(g.drop_duplicates("document_id"), "retention_rate", n_boot=500)220 rows.append({"chain": grp.split(":")[1], "handoffs": h, "accuracy": q["mean"], "acc_lo": q["lo"], "acc_hi": q["hi"], "retention": r["mean"], "ret_lo": r["lo"], "ret_hi": r["hi"],221 "unsupported_mean": g.drop_duplicates("document_id")["unsupported"].mean(), "fetched_bytes_mean": g["fetched_bytes"].mean(), "msg_bytes_mean": g["msg_bytes_hop"].mean(), "n": q["n"], "n_docs": q["n_clusters"]})222 t = pd.DataFrame(rows); p = TAB / f"{chain_exp}_handoffs.csv"; t.to_csv(p, index=False); _prov(p, src, "repeated handoffs")223 primary_tests(dfc[dfc["handoff_index"] == 4].assign(condition=dfc["run_group_id"].str.split(":").str[1]), chain_exp + "_h4", src,224 [{"name": "T5 A5 chain vs A4 chain at 4 handoffs (closed)", "a": "A5|closed|qwen3b", "b": "A4|closed|qwen3b", "hypothesis": "H5"}])225 fan_exp = "E5_core" if "E5_core" in exps else ("E5_pilot" if "E5_pilot" in exps else None)226 if fan_exp:227 dff = load_experiments([f"{fan_exp}.jsonl"]); src = [f"runs/raw/{fan_exp}.jsonl"]228 rows = []229 for (cond, qt), g in dff.groupby(["condition", "qtype"]):230 q = cluster_bootstrap_mean(g, "quality_operational", n_boot=500); rows.append({"condition": cond, "qtype": qt, "quality": q["mean"], "lo": q["lo"], "hi": q["hi"], "n": q["n"], "payload_bytes_median": g["serialized_payload_bytes"].median(), "fetched_bytes_mean": g["fetched_bytes"].mean()})231 t = pd.DataFrame(rows); p = TAB / f"{fan_exp}_fanin.csv"; t.to_csv(p, index=False); _prov(p, src, "fan-in by method and question type")232 net_files = [p for p in (ROOT / "runs/raw").glob("E6_*.jsonl") if "smoke" not in p.name] or list((ROOT / "runs/raw").glob("E6_*.jsonl"))233 if net_files:234 rows = [r for p in net_files for r in read_jsonl(p)]235 fig_breakeven(rows, "E6", [str(p.relative_to(ROOT)) for p in net_files])236 n = pd.DataFrame(rows)237 agg = n.groupby(["label", "method", "bits", "codec", "target_tokens"], dropna=False).agg(n=("repeat", "size"), bytes=("serialized_bytes", "median"), roundtrip_ms_p50=("roundtrip_ms", "median"), send_ms_p50=("send_ms", "median"),238 remote_load_ms_p50=("remote_load_ms", "median"), remote_ttft_p50=("remote_ttft_ms", "median"), remote_prefill_p50=("remote_prefill_ms", "median"), eff_gbps=("effective_gbps", "median"), correct=("correct", "mean")).reset_index()239 p = TAB / "E6_network.csv"; agg.to_csv(p, index=False); _prov(p, [str(x.relative_to(ROOT)) for x in net_files], "two-node transfer measurements")240 hosted = [p for p in (ROOT / "runs/raw").glob("ED_*.jsonl")]241 if hosted:242 dh = load_experiments([p.name for p in hosted]); src = [str(p.relative_to(ROOT)) for p in hosted]243 rows = []244 for (fam, recv, cond), g in dh.groupby(["task_family", "receiver", "condition"]):245 q = cluster_bootstrap_mean(g, "quality_operational", n_boot=500)246 rows.append({"family": fam, "provider_model": recv, "condition": cond, "quality": q["mean"], "lo": q["lo"], "hi": q["hi"], "n": q["n"], "cost_usd_mean": g["cost_usd"].mean(), "ttft_ms_p50": g["ttft_ms"].median(), "total_ms_p50": g["total_wall_ms"].median(), "success_rate": (g["status"] == "success").mean()})247 t = pd.DataFrame(rows); p = TAB / "ED_hosted.csv"; t.to_csv(p, index=False); _prov(p, src, "hosted receivers; cost estimated from provider price tables")248 write_json(ROOT / "results/provenance.json", PROV)249 print(f"analysis done: {len(PROV)} artifacts with provenance")250 return 0251