16.1 引擎的装配检查表

#!/usr/bin/env python3
# file: sim/ne_16_1_assembly.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_16_1_assembly.py
"""卷二 16.1 引擎的装配检查表：把全卷交付的零件装成一张可断言的检查表。

十二项检查，每项返回真或假，任一项 FAIL 即 exit 非 0。判据分两类来源：阈值
（覆盖率下限）从 config/story_spec.json 的 metrics 段读；基线（逐层累计可达状态、
确定性策略摘要）从 out/ 下既有产物读 —— 两者都不写死在脚本里。
本模块同时是本章的共享入口：16.2 / 16.3 复用这里的求值 / 效果 / 状态键 / 规范串。
"""
from __future__ import annotations

import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
SECTIONS = ["meta", "world", "characters", "beats", "scene_graph",
            "dialogue", "conditions", "state", "endings", "metrics"]
DIST = ["web/dist/ne09.html", "web/dist/ne10.html"]
SAMPLE_N, SEED, BFS_LEVELS = 20000, 20261016, 20
AFF_LO, AFF_HI = 0, 100


def load(rel):
    with open(os.path.join(ROOT, rel), encoding="utf-8") as fh:
        return json.load(fh)


def eval_cond(c, s):
    if c is None:
        return True
    k = c[0]
    if k == "flag":
        return s["flags"].get(c[1], 0) == c[2]
    if k == "item":
        return s["items"].get(c[1], 0) == c[2]
    if k == "aff":
        v, op, n = s["aff"].get(c[1], 0), c[2], c[3]
        return v >= n if op == ">=" else v < n if op == "<" else v == n if op == "==" else v != n
    if k == "and":
        return eval_cond(c[1], s) and eval_cond(c[2], s)
    if k == "or":
        return eval_cond(c[1], s) or eval_cond(c[2], s)
    return not eval_cond(c[1], s)


def apply_effects(eff, s):
    aff, flags, items = dict(s["aff"]), dict(s["flags"]), dict(s["items"])
    for k, v in (eff.get("set") or {}).items():
        flags[k] = v
    for k, v in (eff.get("add") or {}).items():
        aff[k] = min(AFF_HI, max(AFF_LO, aff.get(k, 0) + v))
    for k, v in (eff.get("grant") or {}).items():
        items[k] = v
    for k in (eff.get("revoke") or {}):
        items[k] = 0
    return {"aff": aff, "flags": flags, "items": items}


def init_state(spec):
    aff = {v: next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v)
           for v in sorted(spec["state"]["vars"])}
    return {"aff": aff, "flags": {f: 0 for f in sorted(spec["state"]["flags"])},
            "items": {i: 0 for i in sorted(spec["state"]["items"])}}


def state_key(node, s, VARS, FLAGS, ITEMS):
    """去重键 = 节点 × 好感度 × flag 位 × 物品位（不含路径）。"""
    return (node, tuple(s["aff"][v] for v in VARS), tuple(s["flags"][f] for f in FLAGS),
            tuple(s["items"][i] for i in ITEMS))


def bfs_layers(spec, by_id, max_level):
    """逐层 BFS 到 max_level，返回每层累计状态数（与 5.2 基线同口径）。"""
    VARS, FLAGS, ITEMS = (sorted(spec["state"][k]) for k in ("vars", "flags", "items"))
    frontier = [state_key(spec["meta"]["start"], init_state(spec), VARS, FLAGS, ITEMS)]
    seen, out, states = set(frontier), [], 0
    for _ in range(max_level + 1):
        out.append(states + len(frontier))
        nxt = []
        for node, av, fv, iv in frontier:
            states += 1
            st = {"aff": dict(zip(VARS, av)), "flags": dict(zip(FLAGS, fv)),
                  "items": dict(zip(ITEMS, iv))}
            for c in (by_id[node].get("choices") or []):
                if not eval_cond(c.get("when"), st):
                    continue
                k = state_key(c["goto"], apply_effects(c.get("effects") or {}, st), VARS, FLAGS, ITEMS)
                if k not in seen:
                    seen.add(k)
                    nxt.append(k)
        frontier = nxt
    return out


def sample(spec, by_id, n, seed):
    """有界采样：返回（结局分布, 走过的对话边, 卡死路径数）。"""
    rng, dist, edges, dead = seed, {}, set(), 0
    for _ in range(n):
        s, node = init_state(spec), spec["meta"]["start"]
        for _step in range(200):
            cur = by_id[node]
            if cur.get("ending"):
                nm = cur["ending"]
                break
            opts = [c for c in (cur.get("choices") or []) if eval_cond(c.get("when"), s)]
            if not opts:
                nm = "(卡死)"
                break
            rng = (rng * 48271) % 2147483647
            c = opts[rng % len(opts)]
            edges.add((node, c["id"]))
            s, node = apply_effects(c.get("effects") or {}, s), c["goto"]
        else:
            nm = "(未终局)"
        dist[nm] = dist.get(nm, 0) + 1
        dead += 1 if nm in ("(卡死)", "(未终局)") else 0
    return dist, edges, dead


def canonical(node, s, path):
    aff = ",".join(f"{k}={s['aff'][k]}" for k in sorted(s["aff"]))
    fl = ",".join(k for k in sorted(s["flags"]) if s["flags"][k])
    it = ",".join(k for k in sorted(s["items"]) if s["items"][k])
    return f"{node}|{aff}|{fl}|{it}|{'>'.join(path)}"


def digest(text):
    h = 2166136261
    for ch in text:
        h = ((h ^ ord(ch)) * 16777619) & 0xFFFFFFFF
    return h


def main():
    spec = load("config/story_spec.json")
    st, by_id = spec["state"], {n["id"]: n for n in spec["dialogue"]}
    all_ch = {(n["id"], c["id"]) for n in spec["dialogue"] for c in (n.get("choices") or [])}
    snodes = {x["id"] for x in spec["scene_graph"]["nodes"]}
    ends = {e["name"] for e in spec["endings"]}
    refs = {}

    def walk(c):
        if not c:
            return
        if c[0] in ("flag", "aff", "item"):
            refs.setdefault(c[0], set()).add(c[1])
        elif c[0] in ("and", "or"):
            walk(c[1]), walk(c[2])
        elif c[0] == "not":
            walk(c[1])

    for n in spec["dialogue"]:
        walk(n.get("line_when"))
        for c in (n.get("choices") or []):
            walk(c.get("when"))
    for e in spec["scene_graph"]["edges"] + spec["endings"]:
        walk(e.get("when"))

    bad_write = []
    for n in spec["dialogue"]:
        for c in (n.get("choices") or []):
            e = c.get("effects") or {}
            bad_write += [k for k in (e.get("set") or {}) if k not in set(st["flags"])]
            bad_write += [k for k in (e.get("add") or {}) if k not in set(st["vars"])]
            bad_write += [k for k in list(e.get("grant") or {}) + list(e.get("revoke") or {})
                          if k not in set(st["items"])]
    bad_ref = ((refs.get("flag", set()) - set(st["flags"])) | (refs.get("aff", set()) - set(st["vars"]))
               | (refs.get("item", set()) - set(st["items"])))
    goto_bad = [c["id"] for n in spec["dialogue"] for c in (n.get("choices") or []) if c["goto"] not in by_id]
    edge_bad = [e for e in spec["scene_graph"]["edges"] if e["from"] not in snodes or e["to"] not in snodes]

    resid = sum(1 for c in spec["characters"] if not (AFF_LO <= c["affinity"]["init"] <= AFF_HI)
                or list(c["affinity"]["range"]) != [AFF_LO, AFF_HI])
    resid += sum(1 for b in spec["beats"] if not (1 <= b["tension"] <= 5))
    for seg in (spec["dialogue"], spec["beats"], spec["endings"], spec["characters"]):
        ids = [x["id"] for x in seg]
        resid += 1 if len(ids) != len(set(ids)) else 0

    actual = {"scenes": len(spec["scene_graph"]["nodes"]), "characters": len(spec["characters"]),
              "dialogue_nodes": len(spec["dialogue"]), "choices": len(all_ch),
              "flags": len(st["flags"]), "affinity_vars": len(st["vars"]), "items": len(st["items"]),
              "factions": len(spec["world"]["factions"]), "endings": len(spec["endings"])}
    layers = bfs_layers(spec, by_id, BFS_LEVELS)
    base = [r["cumulative"] for r in load("out/ne_05_2_reachable.json")["layers"]]

    dist, edges, dead = sample(spec, by_id, SAMPLE_N, SEED)
    cov = (len(edges) * 1000) // len(all_ch)
    target = spec["metrics"]["coverage_target_permille"]

    s, node, trail, i = init_state(spec), spec["meta"]["start"], [spec["meta"]["start"]], 0
    while i < 64 and not by_id[node].get("ending"):
        opts = [c for c in (by_id[node].get("choices") or []) if eval_cond(c.get("when"), s)]
        if not opts:
            break
        s, node = apply_effects(opts[0].get("effects") or {}, s), opts[0]["goto"]
        trail.append(node)
        i += 1
    anchor = digest(canonical(node, s, trail))

    ext, dist_ok = [], True
    for rel in DIST:
        p = os.path.join(ROOT, rel)
        dist_ok = dist_ok and os.path.exists(p)
        ext += re.findall(r"https?://", open(p, encoding="utf-8").read()) if os.path.exists(p) else []

    rows = [
        ("A1  规格可解析", all(x in spec for x in SECTIONS), f"{len(SECTIONS)}/{len(SECTIONS)} 段"),
        ("A2  引用完整", not (bad_ref or edge_bad or goto_bad),
         f"条件引用 {sum(len(v) for v in refs.values())} 名、场景边 {len(spec['scene_graph']['edges'])} 条"),
        ("A3  写入域合法", not bad_write, f"set/add/grant 越界 {len(bad_write)}"),
        ("A4  整数不变量残差 0", resid == 0, f"残差 {resid}"),
        ("A5  scale 对账", all(spec["meta"]["scale"].get(k) == v for k, v in actual.items()),
         f"{len(actual)}/{len(actual)} 项一致"),
        ("A6  BFS 与基线一致", layers == base[:BFS_LEVELS + 1], f"深度 {BFS_LEVELS} 累计 {layers[-1]}"),
        ("A7  goto 全部存在", not goto_bad, f"{len(all_ch)}/{len(all_ch)}"),
        ("A8  结局全部可达", sum(1 for k in dist if k in ends) == len(ends),
         f"{sum(1 for k in dist if k in ends)}/{len(ends)}"),
        ("A9  跨语言锚点一致", anchor == load("out/ne_01_3_py.json")[0]["digest"], f"锚点 {anchor}"),
        ("A10 W 层零外部依赖", dist_ok and not ext, f"{len(DIST)}/{len(DIST)} 产物，外链 {len(ext)}"),
        ("A11 覆盖率达目标", cov >= target, f"{cov}‰ ≥ {target}‰"),
        ("A12 无卡死路径", dead == 0, f"采样 {SAMPLE_N} 条，卡死 {dead}"),
    ]
    print("装配检查表（《雾港夜航》config/story_spec.json）")
    for name, ok, ev in rows:
        print(f"{name:<20} {'PASS' if ok else 'FAIL':<5} {ev}")
    passed = sum(1 for _, ok, _ in rows if ok)
    print(f"装配结果 {'PASS' if passed == len(rows) else 'FAIL'}（{passed}/{len(rows)}）")

    report = {"passed": passed, "total": len(rows),
              "rows": [{"id": n.split()[0], "ok": ok, "evidence": ev} for n, ok, ev in rows],
              "scale_actual": actual, "coverage_permille": cov, "dead_ends": dead,
              "bfs_depth": BFS_LEVELS, "bfs_cumulative": layers[-1], "anchor_digest": anchor,
              "ending_paths": dist, "sample_n": SAMPLE_N, "seed": SEED}
    with open(os.path.join(ROOT, "out/ne_16_1_assembly.json"), "w", encoding="utf-8") as fh:
        json.dump(report, fh, ensure_ascii=False, indent=1)
        fh.write("\n")
    return 0 if passed == len(rows) else 1


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

装配检查表（《雾港夜航》config/story_spec.json）
A1  规格可解析            PASS  10/10 段
A2  引用完整             PASS  条件引用 22 名、场景边 16 条
A3  写入域合法            PASS  set/add/grant 越界 0
A4  整数不变量残差 0        PASS  残差 0
A5  scale 对账         PASS  9/9 项一致
A6  BFS 与基线一致        PASS  深度 20 累计 357845
A7  goto 全部存在        PASS  115/115
A8  结局全部可达           PASS  5/5
A9  跨语言锚点一致          PASS  锚点 4162796975
A10 W 层零外部依赖         PASS  2/2 产物，外链 0
A11 覆盖率达目标           PASS  1000‰ ≥ 850‰
A12 无卡死路径            PASS  采样 20000 条，卡死 0
装配结果 PASS（12/12）