Blame for tests/benchmark_three_state.py

#!/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())