1"""E8: adaptive representation policy fitted on development records only.23Features (deployment-observable): context tokens (receiver tokenizer), question-access regime, task family, receiver model,4reuse count N, measured LAN bandwidth (constant from E6), and the candidate action's development cost model.5Action = condition id among a feasible set. Quality constraint = lower 95% bootstrap bound of dev mean quality for the (family,6context-bin, pair) group >= tau * best static lower bound; cost = dev mean amortized latency (or bytes). Policy = argmin cost subject7to constraint, with fallback to the dev-best static action if no action satisfies the constraint. Frozen before test evaluation.8"""9from __future__ import annotations10import json, yaml11import numpy as np, pandas as pd12from pathlib import Path13from .util import ROOT, write_json, read_json14from .analysis.load import load_experiments15from .analysis.stats import cluster_bootstrap_mean1617def _ctx_bin(tokens):18 return "short" if tokens <= 3000 else ("medium" if tokens <= 10000 else "long")1920def fit(dev_df: pd.DataFrame, actions: list[str], cost_col: str, tau: float, n_boot: int = 500) -> dict:21 policy = {"actions": actions, "cost_col": cost_col, "tau": tau, "table": {}}22 dev_df = dev_df[dev_df["condition"].isin(actions)].copy()23 dev_df["ctx_bin"] = dev_df["ctx_tokens"].apply(_ctx_bin)24 for (fam, cb, pair, acc), g in dev_df.groupby(["task_family", "ctx_bin", "pair", "question_access_group"]):25 rows = []26 for a, ga in g.groupby("condition"):27 q = cluster_bootstrap_mean(ga, "quality_operational", n_boot=n_boot); c = ga[cost_col].mean()28 rows.append({"action": a, "q_mean": q["mean"], "q_lo": q["lo"], "cost": float(c), "n": q["n"]})29 if not rows: continue30 best = max(rows, key=lambda r: r["q_lo"])31 feasible = [r for r in rows if r["q_lo"] >= tau * best["q_lo"]]32 chosen = min(feasible, key=lambda r: r["cost"]) if feasible else best33 policy["table"][f"{fam}|{cb}|{pair}|{acc}"] = {"chosen": chosen["action"], "best_static": best["action"], "candidates": rows}34 return policy3536def choose(policy: dict, fam: str, ctx_tokens: int, pair: str, acc: str) -> tuple[str, bool]:37 key = f"{fam}|{_ctx_bin(ctx_tokens)}|{pair}|{acc}"38 if key in policy["table"]:39 return policy["table"][key]["chosen"], False40 # fallback: nearest context bin for the same family/pair, else global best static41 for k, v in policy["table"].items():42 if k.startswith(f"{fam}|") and k.endswith(f"|{pair}|{acc}"):43 return v["chosen"], True44 return policy.get("global_best_static", "A1"), True4546def main(config_path: str) -> int:47 cfg = yaml.safe_load(open(config_path))48 df = load_experiments(cfg["dev_patterns"])49 df = df[df["split"] == "dev"]50 df["ctx_tokens"] = df["message_utf8_bytes"].fillna(0) / 4.0 # proxy when receiver source tokens absent51 if "source_tokens_receiver" in df: df["ctx_tokens"] = df["source_tokens_receiver"].fillna(df["ctx_tokens"])52 df["question_access_group"] = df["question_access"].replace({"none": "blind"})53 pol = fit(df, cfg["actions"], cfg["cost_col"], cfg["tau"])54 g = df[df["condition"].isin(cfg["actions"])].groupby("condition")["quality_operational"].mean()55 pol["global_best_static"] = str(g.idxmax()); pol["fitted_on"] = cfg["dev_patterns"]; pol["n_dev_rows"] = int(len(df))56 # dev-best static action per access regime (aware-eligible actions for the aware regime)57 pol["global_best_static_by_access"] = {}58 for acc, ga in df[df["condition"].isin(cfg["actions"])].groupby("question_access_group"):59 gg = ga.groupby("condition")["quality_operational"].mean(); pol["global_best_static_by_access"][acc] = str(gg.idxmax())60 write_json(ROOT / "runs/checkpoints/router_policy.json", pol)61 print("router fitted:", len(pol["table"]), "cells; global best static:", pol["global_best_static"])62 return 063