headless / tests /benchmark_three_state.py

Upload folder using huggingface_hub

216004d verified 10 days ago

19.9 kB

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""benchmark_three_state.py — **三态 epistemic state(好/坏/未知)的验收台**。
## 为什么需要这个文件
leo 的判断: 当前最大的缺口不是精度, 是 **epistemic state** ——
系统必须能区分"好、坏、**未知**", 而在此之前 `advq == 0` 被一律读成"不坏"。
那个 `0` 在两种完全不同的情况下出现:
    (a) 证据充分, 真的没有显著差异           -> 好
    (b) **奖励流为空, Φ 学不到东西**
        (harm 故障下整臂全灭时 `Q_adv ≡ 0.0000` 精确零)  -> **未知**
把 (b) 读成"不坏"就是**自欺**。这个文件的核心判据就是**直接测这个自欺**:
在 agent 正在整体失败的场景里, 旧报告会不会说"一切正常"?
## 三个固定场景(leo 阶段一)
    normal   无故障        —— 正常 option 应该被保留(判定为"好")
    inert    惰性故障      —— 看着效率差, 但**不应误杀**
    harm     有害故障      —— 真正有害, 应该被检出(判"坏", 或至少不是"好")
## 八项固定指标(leo 阶段一)
    ① harm 检测延迟   ② inert 误杀率   ③ 退休后恢复时间   ④ fallback 比例
    ⑤ 未知状态占比    ⑥ 主任务成功率/步数
    ⑦ 证据是否持续增长   ⑧ **自欺率**(失败时仍报"好"的比例)
## 两个臂: 三态本身是一条消融轴
    b2-nostate  冻结基线(GVF + advq + 遗忘 + 原谅)但 `rr_three_state=False`
    b2-3state   同上 + `rr_three_state=True`(未知 -> 一等状态)
这样"把未知提升为一等状态到底改变了什么"是可归因的。
"""
import sys
import json
import argparse
import numpy as np
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from hibs_lnn.rr_agent import RRSkillAgent
from hibs_lnn.skill_mdp import KeyDoorMDP, LEFT, RIGHT
from hibs_lnn.skill_agent import DEFAULT_REGIMES

FAULT_OPT = "rr[door]"

# 固定场景: (regime 序列, 故障, 从第几段起施加故障)
SCENARIOS = {
    "normal": dict(chain=[DEFAULT_REGIMES[i] for i in (0, 1, 2)],
                   fault=None, fault_at=99),
    "inert":  dict(chain=[DEFAULT_REGIMES[i % 3] for i in range(6)],
                   fault="inert", fault_at=3),
    "harm":   dict(chain=[DEFAULT_REGIMES[i % 3] for i in range(6)],
                   fault="harm", fault_at=3),
    # ★★ "不稳定" 故障 —— 与 `harm`("稳定但错误")形成对照。
    #    leo 点名的否证实验就在这里: 如果 A 类只对 `chaos` 有反应、
    #    对 `harm` 无反应, 那它学到的是**可预测性**, 不是**任务价值**,
    #    因此只能当辅助/风险通道, 不能单独驱动退休或选择。
    "chaos":  dict(chain=[DEFAULT_REGIMES[i % 3] for i in range(6)],
                   fault="chaos", fault_at=3),
}


class _RandPolicy:
    """`chaos` 故障: **每步随机出动作** —— 动力学不可预测。
    与 `harm`(把 π_o 反向后仍然完全确定)正好构成一对:
      · `harm`  = 稳定 + 错误   -> 可预测性高, 任务价值负
      · `chaos` = 不稳定 + 无向 -> 可预测性低, 任务价值也是负
    两者对比可以把"熟悉度/可预测性"和"任务价值"**分离**开 —— 这是 A 类
    唯一能证明自己是值信号还是辅助信号的办法。
    """

    def __init__(self, seed=0):
        self.rng = np.random.RandomState(seed)

    def __getitem__(self, s):
        return int(self.rng.randint(2))

    def __len__(self):
        return 64

# ── 冻结基线(leo 契约): GVF + advq + 遗忘 + 原谅 ──────────────────────
FROZEN = dict(rr_select_rule="calibrated", rr_pot="gvf", rr_retire_rule="advq",
              rr_score="advq", rr_forget_hl=20, rr_forgive_p=0.10)

ARMS = {
    "uniform":    dict(rr_select_rule="uniform"),
    "b2-nostate": dict(FROZEN, rr_three_state=False),
    "b2-3state":  dict(FROZEN, rr_three_state=True),
    # ★ 阶段四 A: 三候选稠密动力学证据。其余**全部**沿用冻结契约,
    #   所以臂间差异只能归因到 cumulant 本身(leo 的归因纪律)。
    "a1": dict(FROZEN, rr_three_state=True, rr_dyn="A1"),   # 转移不确定性
    "a2": dict(FROZEN, rr_three_state=True, rr_dyn="A2"),   # 转移预测误差
    "a3": dict(FROZEN, rr_three_state=True, rr_dyn="A3"),   # 已知稳定区域
}


def apply_fault(ag, kind):
    """把故障加到目标 option 上。故障**只在行为上**, 不改台账键。"""
    st = next((s for s in ag.rr_lib_obj if getattr(s, "name", None) == FAULT_OPT), None)
    if st is None:
        return False
    if kind == "inert":                   # 看着坏: 达不成子目标...
        ag._rr_cap[FAULT_OPT] = 1         # ...但那一步仍是 π_o 的最优步
    elif kind == "harm":                  # 真的坏: 主动朝反方向走
        #   ★ 这个故障的性质是 **"稳定 + 错误"**: 转移完全确定、极易预测,
        #     但任务价值是负的。它同时是 A 类的**否证实验**(见判据⑧)。
        p = np.asarray(st.policy).copy()
        p2 = p.copy()
        p2[p == LEFT] = RIGHT
        p2[p == RIGHT] = LEFT
        st.policy = p2
    elif kind == "chaos":                 # 不稳定: 每步随机动作
        #   转移**不可预测**, 与 `harm` 的可预测性形成对照。
        st.policy = _RandPolicy(seed=12345)
        ag._rr_cap[FAULT_OPT] = 12        # 不让它无限跑
    else:
        raise ValueError(f"unknown fault: {kind}")
    return True


def run_arm(kw, scen, episodes_per=300, seed=0, thresh=0.8, window=20):
    sc = SCENARIOS[scen]
    mdp = KeyDoorMDP(n_pos=8, horizon=30, **sc["chain"][0])
    ag = RRSkillAgent(mdp, mode="rr", seed=seed, rr_lib="rr", opt_prob=1.0, **kw)
    segs = []
    for ci, reg in enumerate(sc["chain"]):
        mdp.set_regime(reg["key_pos"], reg["door_pos"], reg["goal_pos"])
        ag._rr_ensure()
        faulted = (sc["fault"] is not None and ci >= sc["fault_at"])
        if faulted:
            apply_fault(ag, sc["fault"])
        n0 = ag.rr_outcome.get(FAULT_OPT, {}).get("n", 0)
        broken_at = None
        succ, steps_ok = [], []
        for _ in range(episodes_per):
            ok, stp = ag.run_episode()
            succ.append(1.0 if ok else 0.0)
            if ok:
                steps_ok.append(stp)
            if faulted and broken_at is None and FAULT_OPT in ag.rr_broken():
                broken_at = ag.rr_outcome[FAULT_OPT]["n"] - n0
        s = np.array(succ)
        t_adapt = episodes_per
        for e in range(window - 1, episodes_per):
            if float(s[e - window + 1:e + 1].mean()) >= thresh:
                t_adapt = e + 1
                break
        cs = ag.rr_calib_stats()
        o = cs["per_option"].get(FAULT_OPT, {})
        others = {k: v for k, v in cs["per_option"].items() if k != FAULT_OPT}
        segs.append({
            "seg": ci, "tc": reg.get("goal_pos"), "faulted": bool(faulted),
            "t_adapt": int(t_adapt),
            "succ": float(s[-window:].mean()),
            "steps_go": float(np.mean(steps_ok)) if steps_ok else float("nan"),
            # ① 检测: 目标是坏的
            "detect": broken_at,
            "bad_state": o.get("state"),
            # ② 误杀: 非目标的 option 被判坏
            "false_kill": sum(1 for v in others.values() if v["state"] == "bad"),
            # ③ 恢复: 目标从坏恢复过几次
            "recovered": o.get("recovered_n", 0),
            "retired": o.get("retired_n", 0),
            # ④ fallback = 退回基元层的**选择尝试占比**。分母必须是 `sel_attempts`
            #   (尝试总数), 不能用 `selections`(只在真的选中时 +1) —— 否则
            #   比值会 > 1(实测 6.827), 指标失去意义。
            "fallback": ag.rr_stats.get("starved_sel", 0)
            / max(1, ag.rr_stats.get("sel_attempts", 1)),
            # ⑤ 三态占比
            "states": cs["states"],
            "starved": cs["starved"],
            "since_rew": cs["since_rew"],
            # ⑦ 证据增长
            "evidence": float(sum(v["advq_n"] for v in cs["per_option"].values())),
            # ★ A 类通道: 转移数 / 基线压力 / 每 option 的证据量与风险标记
            "dyn": (ag.rr_dyn_stats() if hasattr(ag, "rr_dyn_stats") else None),
            "no_base": o.get("no_base", 0),
            "advq": o.get("advq", 0.0),
            "advq_lo": o.get("advq_lo", 0.0),
            "advq_n": o.get("advq_n", 0.0),
            "usage": o.get("n", 0) / max(1, sum(v["n"] for v in cs["per_option"].values())),
        })
    return segs, ag


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--episodes-per", type=int, default=300)
    ap.add_argument("--seeds", type=int, default=1)
    ap.add_argument("--scenarios", default="normal,inert,harm")
    ap.add_argument("--arms", default="uniform,b2-nostate,b2-3state")
    ap.add_argument("--out", default=None)
    a = ap.parse_args()
    scen_names = a.scenarios.split(",")
    arms = {k: v for k, v in ARMS.items() if k in a.arms.split(",")}

    out, agents = {}, {}
    for sc in scen_names:
        for name, kw in arms.items():
            runs = [run_arm(kw, sc, a.episodes_per, sd) for sd in range(a.seeds)]
            out[(sc, name)] = [r[0] for r in runs]
            agents[(sc, name)] = runs[-1][1]

    # ── 打印 ────────────────────────────────────────────────────────
    for sc in scen_names:
        print("\n" + "=" * 108)
        print(f"场景 {sc}   故障={SCENARIOS[sc]['fault']}   从第 {SCENARIOS[sc]['fault_at']} 段起")
        print("=" * 108)
        for name in arms:
            runs = out[(sc, name)]
            print(f"\n── {name} " + "─" * max(0, 96 - len(name)))
            print(f"   {'段':>3}{'故障':>6}{'T_adapt':>9}{'步数':>8}{'成功率':>8}"
                  f"{'使用率':>8}{'advq':>9}{'下界':>9}{'advq_n':>8}"
                  f"{'判定':>8}{'好/坏/未知':>14}{'检测':>7}{'误杀':>6}")
            for si in range(len(SCENARIOS[sc]["chain"])):
                g = [r[si] for r in runs]
                m = lambda f: float(np.nanmean([x[f] for x in g]))
                dv = [x["detect"] for x in g if x["detect"] is not None]
                dl = (f"{np.mean(dv):.0f}" if dv
                      else ("未检测" if g[0]["faulted"] else "—"))
                st = m_states = g[0]["states"]
                st_s = f"{m_states['good']}/{m_states['bad']}/{m_states['unknown']}"
                print(f"   {si:>3}{'是' if g[0]['faulted'] else '否':>6}"
                      f"{m('t_adapt'):>9.1f}{m('steps_go'):>8.2f}{m('succ'):>8.3f}"
                      f"{m('usage'):>8.3f}{m('advq'):>9.4f}{m('advq_lo'):>9.4f}"
                      f"{m('advq_n'):>8.1f}{str(g[0]['bad_state']):>8}{st_s:>14}"
                      f"{dl:>7}{m('false_kill'):>6.0f}")

    # ── 事前判据 ────────────────────────────────────────────────────
    print("\n" + "=" * 108)
    print("事前判据")
    print("=" * 108)
    ok = []
    # ★ `info` = **不计入通过率**的观察项。用于"判据本身还不成立/已被取代"的条目:
    #   留在 `ok` 里会让它永远红, 从而淹没真正的回归(退出码失去分辨力)。
    #   一条判据只有在**它声称验的那件事可以被验**时才有资格当门。
    info = []

    def segs_(sc, nm):
        return out[(sc, nm)][0]

    def last_faulted(sc):
        ch = SCENARIOS[sc]["chain"]
        return [i for i in range(len(ch)) if i >= SCENARIOS[sc]["fault_at"]]

    # ★★ ① 自欺门: 任何"奖励流为空"的段, 都**不得报"好"**
    #    这是 leo 点的核心缺口的直接检验: `advq == 0` 在"证据充分无差异"和
    #    "奖励流为空根本没数据"两种情况下都出现; 前者可以报"好", 后者不行。
    #    注意这个判据对**所有臂**成立 —— 它是报告层的不变量, 与决策规则无关。
    lies = []
    for sc in scen_names:
        for nm in arms:
            for si, x in enumerate(segs_(sc, nm)):
                if x["starved"] and x["states"]["good"] > 0:
                    lies.append((sc, nm, si, x["states"], x["since_rew"]))
    ok.append(("① 自欺门: 奖励流为空时从不报'好'", len(lies) == 0,
               f"违约 {len(lies)} 处" + (f" 例: {lies[0]}" if lies else "")))

    # ★★ ② 决策有效性: 三态开着 -> agent 会因"坏/未知"停止使用有害 option 而**恢复**
    #    这是在测"把未知提升为一等状态"到底改变了什么: 不是报告更好看,
    #    是行为上能自救。
    if "harm" in scen_names and {"b2-nostate", "b2-3state"} <= set(arms):
        nf = [segs_("harm", "b2-nostate")[i] for i in last_faulted("harm")]
        tf = [segs_("harm", "b2-3state")[i] for i in last_faulted("harm")]
        su_n = float(np.mean([x["succ"] for x in nf]))
        su_t = float(np.mean([x["succ"] for x in tf]))
        ok.append(("② 三态使 agent 在失败场景下恢复 (成功率 3state 显著 > nostate)",
                   su_t > su_n + 0.2, f"{su_n:.3f} -> {su_t:.3f}"))

    # ③ 未知不得触发退休 / 不得增加误杀
    #    ★ 守卫必须有: 臂子集是常规用法(--arms a1,a2,a3), 缺守卫会直接
    #      KeyError 崩在**评测阶段** —— 数据全跑完了才崩, 最浪费的一种。
    if "harm" in scen_names and {"b2-nostate", "b2-3state"} <= set(arms):
        print(f"     harm 末段 误杀(非目标 option 被判坏): "
              f"nostate {segs_('harm','b2-nostate')[-1]['false_kill']} / "
              f"3state {segs_('harm','b2-3state')[-1]['false_kill']}")
        ok.append(("③ 三态不增加误杀",
                   segs_("harm", "b2-3state")[-1]["false_kill"]
                   <= segs_("harm", "b2-nostate")[-1]["false_kill"]))

    # ④ inert 不应被系统性误杀
    if "inert" in scen_names:
        for nm in ("b2-nostate", "b2-3state"):
            if nm in arms:
                fk = sum(segs_("inert", nm)[i]["false_kill"] for i in last_faulted("inert"))
                print(f"     inert 故障段累计误杀 {nm}: {fk}")
        ok.append(("④ inert 场景下三态臂不误杀",
                   all(segs_("inert", "b2-3state")[i]["false_kill"] == 0
                       for i in last_faulted("inert"))))

    # ⑤ 正常场景应判"好"(三态不能退化成一律未知)
    if "normal" in scen_names:
        c = segs_("normal", "b2-3state")[-1]
        print(f"     normal 末段: 三态 {c['states']}  (starved={c['starved']})")
        ok.append(("⑤ normal 场景下目标 option 最终判'好'", c["states"]["good"] >= 1))

    # ⑥ 证据持续增长
    for sc in scen_names:
        if "b2-3state" not in arms:
            continue
        ev = [x["evidence"] for x in segs_(sc, "b2-3state")]
        ok.append((f"⑥ [{sc}] 证据持续增长 (advq_n 单调)",
                   all(ev[i + 1] >= ev[i] - 1e-9 for i in range(len(ev) - 1))))

    # ★★ ⑨ **反熟悉度否证实验**(leo 点名的最重要一条; 编号接在既有 ⑧ 之后)
    #    对两个镜像故障各问一个问题:
    #      chaos(不稳定): A 类**应该**察觉到 -> 证明它测的是"可预测性"
    #      harm (稳定+错误): A 类**应该**察觉不到 -> 证明它测的**不是**
    #                        "任务价值", 因此只能当辅助/风险通道
    #    如果 A 在 harm 上也"察觉到了", 那是更可疑的情况(需另查来源);
    #    如果 A 在 chaos 上毫无反应, 那它连"可预测性"都没测到 -> 直接淘汰。
    if "chaos" in scen_names:
        def dyn_flag(sc, nm):
            """故障段里 A 类触发风险的 (命中, 总数)。"""
            hit = tot = 0
            for x in segs_(sc, nm):
                dy = x.get("dyn")
                if not dy or not x["faulted"]:
                    continue
                for v in dy["per_option"].values():
                    tot += 1
                    hit += int(v["risk"])
            return hit, tot
        for nm in ("a1", "a2", "a3"):
            if nm not in arms:
                continue
            hc, tc = dyn_flag("chaos", nm)
            hh, th = dyn_flag("harm", nm)
            # ★★ **本条已从"门"降级为"观察项"(INFO)** —— 理由是它**不可满足**:
            #    本台跑在**确定性** KeyDoorMDP 上, 实测所有 n>=2 的 (s,a) 后继分布
            #    最大熵**逐位 0.000000** ⇒ H(T|S,A) ≡ 0 ⇒ 转移预测误差 / 不确定性 /
            #    稳定区域 **在数学上坍缩成同一个量 = 新颖度计数**。
            #    于是 A 在 chaos 与 harm 上必然给出**同一签名** —— 这不是
            #    "A 类不行", 是**本环境没有能力分辨**。
            #    留它在 `ok` 里会让整台永远 3 条红, 退出码失去分辨力
            #    (真正的回归会被淹没)。
            #    新定义(T_shift/T_detect/T_recover/FP/FN/AUC)与可辨识环境见
            #    `tests/benchmark_a_prime.py` + `docs/a_prime_decision.md`。
            info.append((f"⑨ [{nm}] (已被取代, 见 benchmark_a_prime.py) "
                         f"本台不可满足: 确定性环境 H(T|S,A)≡0, A 类 ≡ 新颖度",
                         f"chaos {hc}/{tc}   harm {hh}/{th}"))

    # ⑦ 主任务不劣于安全基线 —— **只在 uniform 真正能跑的场景比**
    #    (harm 下 uniform 整体失败 steps_go=nan, 那时比较没有意义;
    #     失败场景由判据 ② 用成功率来测)
    for sc in scen_names:
        if "uniform" not in arms or "b2-3state" not in arms:
            continue
        fu = np.nanmean([x["steps_go"] for x in segs_(sc, "uniform")])
        fs = np.nanmean([x["steps_go"] for x in segs_(sc, "b2-3state")])
        # ★ 安全基线的可用性必须**只在故障段上**判: harm 下 uniform 在故障段
        #   成功率 0.000 / 步数 nan(整体失败), 那时"比步数"没有意义 ——
        #   用全段均值会把健康段的成功混进来, 误以为基线可用。
        uf = [x for x in segs_(sc, "uniform") if x["faulted"]]
        us_f = np.mean([x["succ"] for x in uf]) if uf else 1.0
        if np.isnan(fu) or np.isnan(fs) or us_f < 0.5:
            ok.append((f"⑦ [{sc}] 主任务不劣于 uniform", True,
                       f"uniform 在故障段基线不可用 (成功率 {us_f:.3f}), 由判据②覆盖"))
        else:
            ok.append((f"⑦ [{sc}] 主任务步数不劣于 uniform (增幅 < 15%)",
                       fs <= fu * 1.15, f"{fu:.2f} -> {fs:.2f}"))

    n_pass = 0
    for item in ok:
        nm, v = item[0], item[1]
        extra = item[2] if len(item) > 2 else None
        n_pass += int(bool(v))
        print(f"  {'PASS' if v else 'FAIL'}  {nm}" + (f"   [{extra}]" if extra else ""))
    for item in info:                       # ★ 不计入通过率, 不影响退出码
        nm = item[0]
        extra = item[1] if len(item) > 1 else None
        print(f"  INFO  {nm}" + (f"   [{extra}]" if extra else ""))
    print(f"\n  {n_pass}/{len(ok)} 通过" +
          (f"   (+{len(info)} 条 INFO, 不计入)" if info else ""))

    if a.out:
        Path(a.out).parent.mkdir(parents=True, exist_ok=True)
        ser = {f"{sc}|{nm}": runs for (sc, nm), runs in out.items()}
        Path(a.out).write_text(json.dumps(ser, indent=2, default=str, ensure_ascii=False))
        print(f"原始数据: {a.out}")
    return 0 if n_pass == len(ok) else 1


if __name__ == "__main__":
    sys.exit(main())

	#!/usr/bin/env python3
	# -- coding: utf-8 --
	"""benchmark_three_state.py — 三态 epistemic state(好/坏/未知)的验收台。

	## 为什么需要这个文件

	leo 的判断: 当前最大的缺口不是精度, 是 epistemic state ——
	系统必须能区分"好、坏、未知", 而在此之前 `advq == 0` 被一律读成"不坏"。

	那个 `0` 在两种完全不同的情况下出现:

	(a) 证据充分, 真的没有显著差异 -> 好
	(b) 奖励流为空, Φ 学不到东西
	(harm 故障下整臂全灭时 `Q_adv ≡ 0.0000` 精确零) -> 未知

	把 (b) 读成"不坏"就是自欺。这个文件的核心判据就是直接测这个自欺:
	在 agent 正在整体失败的场景里, 旧报告会不会说"一切正常"?

	## 三个固定场景(leo 阶段一)

	normal 无故障 —— 正常 option 应该被保留(判定为"好")
	inert 惰性故障 —— 看着效率差, 但不应误杀
	harm 有害故障 —— 真正有害, 应该被检出(判"坏", 或至少不是"好")

	## 八项固定指标(leo 阶段一)

	① harm 检测延迟 ② inert 误杀率 ③ 退休后恢复时间 ④ fallback 比例
	⑤ 未知状态占比 ⑥ 主任务成功率/步数
	⑦ 证据是否持续增长 ⑧ 自欺率(失败时仍报"好"的比例)

	## 两个臂: 三态本身是一条消融轴

	b2-nostate 冻结基线(GVF + advq + 遗忘 + 原谅)但 `rr_three_state=False`
	b2-3state 同上 + `rr_three_state=True`(未知 -> 一等状态)

	这样"把未知提升为一等状态到底改变了什么"是可归因的。
	"""
	import sys
	import json
	import argparse
	import numpy as np
	from pathlib import Path

	sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

	from hibs_lnn.rr_agent import RRSkillAgent
	from hibs_lnn.skill_mdp import KeyDoorMDP, LEFT, RIGHT
	from hibs_lnn.skill_agent import DEFAULT_REGIMES

	FAULT_OPT = "rr[door]"

	# 固定场景: (regime 序列, 故障, 从第几段起施加故障)
	SCENARIOS = {
	"normal": dict(chain=[DEFAULT_REGIMES[i] for i in (0, 1, 2)],
	fault=None, fault_at=99),
	"inert": dict(chain=[DEFAULT_REGIMES[i % 3] for i in range(6)],
	fault="inert", fault_at=3),
	"harm": dict(chain=[DEFAULT_REGIMES[i % 3] for i in range(6)],
	fault="harm", fault_at=3),
	# ★★ "不稳定" 故障 —— 与 `harm`("稳定但错误")形成对照。
	# leo 点名的否证实验就在这里: 如果 A 类只对 `chaos` 有反应、
	# 对 `harm` 无反应, 那它学到的是可预测性, 不是任务价值,
	# 因此只能当辅助/风险通道, 不能单独驱动退休或选择。
	"chaos": dict(chain=[DEFAULT_REGIMES[i % 3] for i in range(6)],
	fault="chaos", fault_at=3),
	}


	class _RandPolicy:
	"""`chaos` 故障: 每步随机出动作 —— 动力学不可预测。

	与 `harm`(把 π_o 反向后仍然完全确定)正好构成一对:
	· `harm` = 稳定 + 错误 -> 可预测性高, 任务价值负
	· `chaos` = 不稳定 + 无向 -> 可预测性低, 任务价值也是负
	两者对比可以把"熟悉度/可预测性"和"任务价值"分离开 —— 这是 A 类
	唯一能证明自己是值信号还是辅助信号的办法。
	"""

	def __init__(self, seed=0):
	self.rng = np.random.RandomState(seed)

	def __getitem__(self, s):
	return int(self.rng.randint(2))

	def __len__(self):
	return 64

	# ── 冻结基线(leo 契约): GVF + advq + 遗忘 + 原谅 ──────────────────────
	FROZEN = dict(rr_select_rule="calibrated", rr_pot="gvf", rr_retire_rule="advq",
	rr_score="advq", rr_forget_hl=20, rr_forgive_p=0.10)

	ARMS = {
	"uniform": dict(rr_select_rule="uniform"),
	"b2-nostate": dict(FROZEN, rr_three_state=False),
	"b2-3state": dict(FROZEN, rr_three_state=True),
	# ★ 阶段四 A: 三候选稠密动力学证据。其余全部沿用冻结契约,
	# 所以臂间差异只能归因到 cumulant 本身(leo 的归因纪律)。
	"a1": dict(FROZEN, rr_three_state=True, rr_dyn="A1"), # 转移不确定性
	"a2": dict(FROZEN, rr_three_state=True, rr_dyn="A2"), # 转移预测误差
	"a3": dict(FROZEN, rr_three_state=True, rr_dyn="A3"), # 已知稳定区域
	}


	def apply_fault(ag, kind):
	"""把故障加到目标 option 上。故障只在行为上, 不改台账键。"""
	st = next((s for s in ag.rr_lib_obj if getattr(s, "name", None) == FAULT_OPT), None)
	if st is None:
	return False
	if kind == "inert": # 看着坏: 达不成子目标...
	ag._rr_cap[FAULT_OPT] = 1 # ...但那一步仍是 π_o 的最优步
	elif kind == "harm": # 真的坏: 主动朝反方向走
	# ★ 这个故障的性质是 "稳定 + 错误": 转移完全确定、极易预测,
	# 但任务价值是负的。它同时是 A 类的否证实验(见判据⑧)。
	p = np.asarray(st.policy).copy()
	p2 = p.copy()
	p2[p == LEFT] = RIGHT
	p2[p == RIGHT] = LEFT
	st.policy = p2
	elif kind == "chaos": # 不稳定: 每步随机动作
	# 转移不可预测, 与 `harm` 的可预测性形成对照。
	st.policy = _RandPolicy(seed=12345)
	ag._rr_cap[FAULT_OPT] = 12 # 不让它无限跑
	else:
	raise ValueError(f"unknown fault: {kind}")
	return True


	def run_arm(kw, scen, episodes_per=300, seed=0, thresh=0.8, window=20):
	sc = SCENARIOS[scen]
	mdp = KeyDoorMDP(n_pos=8, horizon=30, **sc["chain"][0])
	ag = RRSkillAgent(mdp, mode="rr", seed=seed, rr_lib="rr", opt_prob=1.0, **kw)
	segs = []
	for ci, reg in enumerate(sc["chain"]):
	mdp.set_regime(reg["key_pos"], reg["door_pos"], reg["goal_pos"])
	ag._rr_ensure()
	faulted = (sc["fault"] is not None and ci >= sc["fault_at"])
	if faulted:
	apply_fault(ag, sc["fault"])
	n0 = ag.rr_outcome.get(FAULT_OPT, {}).get("n", 0)
	broken_at = None
	succ, steps_ok = [], []
	for _ in range(episodes_per):
	ok, stp = ag.run_episode()
	succ.append(1.0 if ok else 0.0)
	if ok:
	steps_ok.append(stp)
	if faulted and broken_at is None and FAULT_OPT in ag.rr_broken():
	broken_at = ag.rr_outcome[FAULT_OPT]["n"] - n0
	s = np.array(succ)
	t_adapt = episodes_per
	for e in range(window - 1, episodes_per):
	if float(s[e - window + 1:e + 1].mean()) >= thresh:
	t_adapt = e + 1
	break
	cs = ag.rr_calib_stats()
	o = cs["per_option"].get(FAULT_OPT, {})
	others = {k: v for k, v in cs["per_option"].items() if k != FAULT_OPT}
	segs.append({
	"seg": ci, "tc": reg.get("goal_pos"), "faulted": bool(faulted),
	"t_adapt": int(t_adapt),
	"succ": float(s[-window:].mean()),
	"steps_go": float(np.mean(steps_ok)) if steps_ok else float("nan"),
	# ① 检测: 目标是坏的
	"detect": broken_at,
	"bad_state": o.get("state"),
	# ② 误杀: 非目标的 option 被判坏
	"false_kill": sum(1 for v in others.values() if v["state"] == "bad"),
	# ③ 恢复: 目标从坏恢复过几次
	"recovered": o.get("recovered_n", 0),
	"retired": o.get("retired_n", 0),
	# ④ fallback = 退回基元层的选择尝试占比。分母必须是 `sel_attempts`
	# (尝试总数), 不能用 `selections`(只在真的选中时 +1) —— 否则
	# 比值会 > 1(实测 6.827), 指标失去意义。
	"fallback": ag.rr_stats.get("starved_sel", 0)
	/ max(1, ag.rr_stats.get("sel_attempts", 1)),
	# ⑤ 三态占比
	"states": cs["states"],
	"starved": cs["starved"],
	"since_rew": cs["since_rew"],
	# ⑦ 证据增长
	"evidence": float(sum(v["advq_n"] for v in cs["per_option"].values())),
	# ★ A 类通道: 转移数 / 基线压力 / 每 option 的证据量与风险标记
	"dyn": (ag.rr_dyn_stats() if hasattr(ag, "rr_dyn_stats") else None),
	"no_base": o.get("no_base", 0),
	"advq": o.get("advq", 0.0),
	"advq_lo": o.get("advq_lo", 0.0),
	"advq_n": o.get("advq_n", 0.0),
	"usage": o.get("n", 0) / max(1, sum(v["n"] for v in cs["per_option"].values())),
	})
	return segs, ag


	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("--episodes-per", type=int, default=300)
	ap.add_argument("--seeds", type=int, default=1)
	ap.add_argument("--scenarios", default="normal,inert,harm")
	ap.add_argument("--arms", default="uniform,b2-nostate,b2-3state")
	ap.add_argument("--out", default=None)
	a = ap.parse_args()
	scen_names = a.scenarios.split(",")
	arms = {k: v for k, v in ARMS.items() if k in a.arms.split(",")}

	out, agents = {}, {}
	for sc in scen_names:
	for name, kw in arms.items():
	runs = [run_arm(kw, sc, a.episodes_per, sd) for sd in range(a.seeds)]
	out[(sc, name)] = [r[0] for r in runs]
	agents[(sc, name)] = runs[-1][1]

	# ── 打印 ────────────────────────────────────────────────────────
	for sc in scen_names:
	print("\n" + "=" * 108)
	print(f"场景 {sc} 故障={SCENARIOS[sc]['fault']} 从第 {SCENARIOS[sc]['fault_at']} 段起")
	print("=" * 108)
	for name in arms:
	runs = out[(sc, name)]
	print(f"\n── {name} " + "─" * max(0, 96 - len(name)))
	print(f" {'段':>3}{'故障':>6}{'T_adapt':>9}{'步数':>8}{'成功率':>8}"
	f"{'使用率':>8}{'advq':>9}{'下界':>9}{'advq_n':>8}"
	f"{'判定':>8}{'好/坏/未知':>14}{'检测':>7}{'误杀':>6}")
	for si in range(len(SCENARIOS[sc]["chain"])):
	g = [r[si] for r in runs]
	m = lambda f: float(np.nanmean([x[f] for x in g]))
	dv = [x["detect"] for x in g if x["detect"] is not None]
	dl = (f"{np.mean(dv):.0f}" if dv
	else ("未检测" if g[0]["faulted"] else "—"))
	st = m_states = g[0]["states"]
	st_s = f"{m_states['good']}/{m_states['bad']}/{m_states['unknown']}"
	print(f" {si:>3}{'是' if g[0]['faulted'] else '否':>6}"
	f"{m('t_adapt'):>9.1f}{m('steps_go'):>8.2f}{m('succ'):>8.3f}"
	f"{m('usage'):>8.3f}{m('advq'):>9.4f}{m('advq_lo'):>9.4f}"
	f"{m('advq_n'):>8.1f}{str(g[0]['bad_state']):>8}{st_s:>14}"
	f"{dl:>7}{m('false_kill'):>6.0f}")

	# ── 事前判据 ────────────────────────────────────────────────────
	print("\n" + "=" * 108)
	print("事前判据")
	print("=" * 108)
	ok = []
	# ★ `info` = 不计入通过率的观察项。用于"判据本身还不成立/已被取代"的条目:
	# 留在 `ok` 里会让它永远红, 从而淹没真正的回归(退出码失去分辨力)。
	# 一条判据只有在它声称验的那件事可以被验时才有资格当门。
	info = []

	def segs_(sc, nm):
	return out[(sc, nm)][0]

	def last_faulted(sc):
	ch = SCENARIOS[sc]["chain"]
	return [i for i in range(len(ch)) if i >= SCENARIOS[sc]["fault_at"]]

	# ★★ ① 自欺门: 任何"奖励流为空"的段, 都不得报"好"
	# 这是 leo 点的核心缺口的直接检验: `advq == 0` 在"证据充分无差异"和
	# "奖励流为空根本没数据"两种情况下都出现; 前者可以报"好", 后者不行。
	# 注意这个判据对所有臂成立 —— 它是报告层的不变量, 与决策规则无关。
	lies = []
	for sc in scen_names:
	for nm in arms:
	for si, x in enumerate(segs_(sc, nm)):
	if x["starved"] and x["states"]["good"] > 0:
	lies.append((sc, nm, si, x["states"], x["since_rew"]))
	ok.append(("① 自欺门: 奖励流为空时从不报'好'", len(lies) == 0,
	f"违约 {len(lies)} 处" + (f" 例: {lies[0]}" if lies else "")))

	# ★★ ② 决策有效性: 三态开着 -> agent 会因"坏/未知"停止使用有害 option 而恢复
	# 这是在测"把未知提升为一等状态"到底改变了什么: 不是报告更好看,
	# 是行为上能自救。
	if "harm" in scen_names and {"b2-nostate", "b2-3state"} <= set(arms):
	nf = [segs_("harm", "b2-nostate")[i] for i in last_faulted("harm")]
	tf = [segs_("harm", "b2-3state")[i] for i in last_faulted("harm")]
	su_n = float(np.mean([x["succ"] for x in nf]))
	su_t = float(np.mean([x["succ"] for x in tf]))
	ok.append(("② 三态使 agent 在失败场景下恢复 (成功率 3state 显著 > nostate)",
	su_t > su_n + 0.2, f"{su_n:.3f} -> {su_t:.3f}"))

	# ③ 未知不得触发退休 / 不得增加误杀
	# ★ 守卫必须有: 臂子集是常规用法(--arms a1,a2,a3), 缺守卫会直接
	# KeyError 崩在评测阶段 —— 数据全跑完了才崩, 最浪费的一种。
	if "harm" in scen_names and {"b2-nostate", "b2-3state"} <= set(arms):
	print(f" harm 末段误杀(非目标 option 被判坏): "
	f"nostate {segs_('harm','b2-nostate')[-1]['false_kill']} / "
	f"3state {segs_('harm','b2-3state')[-1]['false_kill']}")
	ok.append(("③ 三态不增加误杀",
	segs_("harm", "b2-3state")[-1]["false_kill"]
	<= segs_("harm", "b2-nostate")[-1]["false_kill"]))

	# ④ inert 不应被系统性误杀
	if "inert" in scen_names:
	for nm in ("b2-nostate", "b2-3state"):
	if nm in arms:
	fk = sum(segs_("inert", nm)[i]["false_kill"] for i in last_faulted("inert"))
	print(f" inert 故障段累计误杀 {nm}: {fk}")
	ok.append(("④ inert 场景下三态臂不误杀",
	all(segs_("inert", "b2-3state")[i]["false_kill"] == 0
	for i in last_faulted("inert"))))

	# ⑤ 正常场景应判"好"(三态不能退化成一律未知)
	if "normal" in scen_names:
	c = segs_("normal", "b2-3state")[-1]
	print(f" normal 末段: 三态 {c['states']} (starved={c['starved']})")
	ok.append(("⑤ normal 场景下目标 option 最终判'好'", c["states"]["good"] >= 1))

	# ⑥ 证据持续增长
	for sc in scen_names:
	if "b2-3state" not in arms:
	continue
	ev = [x["evidence"] for x in segs_(sc, "b2-3state")]
	ok.append((f"⑥ [{sc}] 证据持续增长 (advq_n 单调)",
	all(ev[i + 1] >= ev[i] - 1e-9 for i in range(len(ev) - 1))))

	# ★★ ⑨ 反熟悉度否证实验(leo 点名的最重要一条; 编号接在既有 ⑧ 之后)
	# 对两个镜像故障各问一个问题:
	# chaos(不稳定): A 类应该察觉到 -> 证明它测的是"可预测性"
	# harm (稳定+错误): A 类应该察觉不到 -> 证明它测的不是
	# "任务价值", 因此只能当辅助/风险通道
	# 如果 A 在 harm 上也"察觉到了", 那是更可疑的情况(需另查来源);
	# 如果 A 在 chaos 上毫无反应, 那它连"可预测性"都没测到 -> 直接淘汰。
	if "chaos" in scen_names:
	def dyn_flag(sc, nm):
	"""故障段里 A 类触发风险的 (命中, 总数)。"""
	hit = tot = 0
	for x in segs_(sc, nm):
	dy = x.get("dyn")
	if not dy or not x["faulted"]:
	continue
	for v in dy["per_option"].values():
	tot += 1
	hit += int(v["risk"])
	return hit, tot
	for nm in ("a1", "a2", "a3"):
	if nm not in arms:
	continue
	hc, tc = dyn_flag("chaos", nm)
	hh, th = dyn_flag("harm", nm)
	# ★★ 本条已从"门"降级为"观察项"(INFO) —— 理由是它不可满足:
	# 本台跑在确定性 KeyDoorMDP 上, 实测所有 n>=2 的 (s,a) 后继分布
	# 最大熵逐位 0.000000 ⇒ H(T\|S,A) ≡ 0 ⇒ 转移预测误差 / 不确定性 /
	# 稳定区域在数学上坍缩成同一个量 = 新颖度计数。
	# 于是 A 在 chaos 与 harm 上必然给出同一签名 —— 这不是
	# "A 类不行", 是本环境没有能力分辨。
	# 留它在 `ok` 里会让整台永远 3 条红, 退出码失去分辨力
	# (真正的回归会被淹没)。
	# 新定义(T_shift/T_detect/T_recover/FP/FN/AUC)与可辨识环境见
	# `tests/benchmark_a_prime.py` + `docs/a_prime_decision.md`。
	info.append((f"⑨ [{nm}] (已被取代, 见 benchmark_a_prime.py) "
	f"本台不可满足: 确定性环境 H(T\|S,A)≡0, A 类 ≡ 新颖度",
	f"chaos {hc}/{tc} harm {hh}/{th}"))

	# ⑦ 主任务不劣于安全基线 —— 只在 uniform 真正能跑的场景比
	# (harm 下 uniform 整体失败 steps_go=nan, 那时比较没有意义;
	# 失败场景由判据 ② 用成功率来测)
	for sc in scen_names:
	if "uniform" not in arms or "b2-3state" not in arms:
	continue
	fu = np.nanmean([x["steps_go"] for x in segs_(sc, "uniform")])
	fs = np.nanmean([x["steps_go"] for x in segs_(sc, "b2-3state")])
	# ★ 安全基线的可用性必须只在故障段上判: harm 下 uniform 在故障段
	# 成功率 0.000 / 步数 nan(整体失败), 那时"比步数"没有意义 ——
	# 用全段均值会把健康段的成功混进来, 误以为基线可用。
	uf = [x for x in segs_(sc, "uniform") if x["faulted"]]
	us_f = np.mean([x["succ"] for x in uf]) if uf else 1.0
	if np.isnan(fu) or np.isnan(fs) or us_f < 0.5:
	ok.append((f"⑦ [{sc}] 主任务不劣于 uniform", True,
	f"uniform 在故障段基线不可用 (成功率 {us_f:.3f}), 由判据②覆盖"))
	else:
	ok.append((f"⑦ [{sc}] 主任务步数不劣于 uniform (增幅 < 15%)",
	fs <= fu * 1.15, f"{fu:.2f} -> {fs:.2f}"))

	n_pass = 0
	for item in ok:
	nm, v = item[0], item[1]
	extra = item[2] if len(item) > 2 else None
	n_pass += int(bool(v))
	print(f" {'PASS' if v else 'FAIL'} {nm}" + (f" [{extra}]" if extra else ""))
	for item in info: # ★ 不计入通过率, 不影响退出码
	nm = item[0]
	extra = item[1] if len(item) > 1 else None
	print(f" INFO {nm}" + (f" [{extra}]" if extra else ""))
	print(f"\n {n_pass}/{len(ok)} 通过" +
	(f" (+{len(info)} 条 INFO, 不计入)" if info else ""))

	if a.out:
	Path(a.out).parent.mkdir(parents=True, exist_ok=True)
	ser = {f"{sc}\|{nm}": runs for (sc, nm), runs in out.items()}
	Path(a.out).write_text(json.dumps(ser, indent=2, default=str, ensure_ascii=False))
	print(f"原始数据: {a.out}")
	return 0 if n_pass == len(ok) else 1


	if __name__ == "__main__":
	sys.exit(main())