16.2 卷一消费的预演

#!/usr/bin/env python3
# file: sim/na_16_2_rehearsal.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/na_16_2_rehearsal.py
"""卷一 16.2 卷一消费的预演：在同一份规格上把三处消费（穷举 / 执行 / 度量）串一遍。

穷举（S）：完整枚举 → 覆盖率 / 不可达 / 结局可达性。
执行（S）：seed=20261016 第 0 条路径推进到底，输出状态摘要与 digest；P 层复算，两者逐位相同。
度量（S）：节拍张力 vs 目标曲线的 MAE（×1000）+ 结局分布熵（Shannon ×1000）。
产物 out/na16_rehearsal.json 供 P 层读取。复用 sim/na_00_spec_check.py 的求值 / 效果 / 状态键。
"""
from __future__ import annotations

import json
import math
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import na_00_spec_check as C  # noqa: E402

SEED, SAMPLE_N = 20261016, 20000


def init_state(spec: dict) -> dict:
    """按 characters 的 init 建初始状态，并给共享入口设好状态键的位序。"""
    C.VAR_ORDER = sorted(spec["state"]["vars"])
    aff = {v: next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v)
           for v in C.VAR_ORDER}
    return {"aff": aff, "flags": {f: 0 for f in spec["state"]["flags"]},
            "items": {i: 0 for i in spec["state"]["items"]}}


def step(spec: dict, st: dict, node: str, cid: str):
    """推进一步；选项条件不满足即报错。"""
    n = next(x for x in spec["dialogue"] if x["id"] == node)
    c = next(x for x in (n.get("choices") or []) if x["id"] == cid)
    if not C.eval_cond(c.get("when"), st):
        raise ValueError(f"choice unavailable: {cid}")
    return c["goto"], C.apply_effects(c.get("effects") or {}, st)


def walk(spec: dict, seed: int, n: int) -> list[list[str]]:
    """采样 n 条路径（Lehmer MINSTD，与 TS 侧同算法同种子）。"""
    by_id = {x["id"]: x for x in spec["dialogue"]}
    rng, out = seed, []
    for _ in range(n):
        st, node, trail = init_state(spec), spec["meta"]["start"], []
        for _step in range(200):
            if by_id[node].get("ending"):
                break
            opts = [c for c in (by_id[node].get("choices") or []) if C.eval_cond(c.get("when"), st)]
            if not opts:
                break
            rng = (rng * 48271) % 2147483647
            c = opts[rng % len(opts)]
            trail.append(c["id"])
            node, st = step(spec, st, node, c["id"])
        out.append(trail)
    return out


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


def digest(s: str) -> int:
    h = 0x811C9DC5
    for ch in s:
        h = ((h ^ ord(ch)) * 16777619) & 0xFFFFFFFF
    return h


def main() -> int:
    spec = C.load_spec()
    by_id = {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 [])}

    # ---- 穷举：完整枚举可达状态 ----
    init = init_state(spec)
    C.FLAG_ORDER, C.ITEM_ORDER = sorted(spec["state"]["flags"]), sorted(spec["state"]["items"])
    start = spec["meta"]["start"]
    seen, frontier = {C.state_key(start, init)}, [(start, init)]
    edges, nodes = set(), set()
    hits = {e["name"]: 0 for e in spec["endings"]}
    while frontier:
        nxt = []
        for node, st in frontier:
            nodes.add(node)
            if by_id[node].get("ending"):
                hits[by_id[node]["ending"]] += 1
                continue
            for c in (by_id[node].get("choices") or []):
                if not C.eval_cond(c.get("when"), st):
                    continue
                edges.add((node, c["id"]))
                st2 = C.apply_effects(c.get("effects") or {}, st)
                k = C.state_key(c["goto"], st2)
                if k not in seen:
                    seen.add(k)
                    nxt.append((c["goto"], st2))
        frontier = nxt
    cov = (len(edges) * 1000) // len(all_ch)
    un_nodes = sorted({n["id"] for n in spec["dialogue"]} - nodes)

    # ---- 执行：固定路径推进到底 ----
    p = walk(spec, SEED, 1)[0]
    st, node, trail = init_state(spec), start, [start]
    for cid in p:
        node, st = step(spec, st, node, cid)
        trail.append(node)
    canon = canonical(node, st, trail)
    dg = digest(canon)
    ending = by_id[node].get("ending", "")

    # ---- 度量：张力 MAE + 结局分布熵 ----
    curve = [b["tension"] for b in spec["beats"]]
    target = spec["metrics"]["target_tension_curve"]
    mae = (sum(abs(a - b) for a, b in zip(curve, target)) * 1000) // len(target)
    dist: dict[str, int] = {}
    for q in walk(spec, SEED, SAMPLE_N):
        nd, s2 = start, init_state(spec)
        for cid in q:
            nd, s2 = step(spec, s2, nd, cid)
        nm = by_id[nd].get("ending", "(未终局)")
        dist[nm] = dist.get(nm, 0) + 1
    ent = -sum(v / SAMPLE_N * math.log2(v / SAMPLE_N) for v in dist.values())
    ent_milli = int(round(ent * 1000))

    out = {"seed": SEED, "path": p, "canonical": canon, "digest": dg, "node": node, "ending": ending,
           "coverage_permille": cov, "states": len(seen), "unreachable_nodes": un_nodes,
           "ending_states": hits, "tension_mae_milli": mae, "ending_entropy_milli": ent_milli}
    dst = os.path.join(HERE, "../out/na16_rehearsal.json")
    os.makedirs(os.path.dirname(dst), exist_ok=True)
    with open(dst, "w", encoding="utf-8") as fh:
        json.dump(out, fh, ensure_ascii=False, indent=1)
        fh.write("\n")

    print(f"穷举（S）：覆盖率 {cov}‰（{len(edges)}/{len(all_ch)} 条对话边）；"
          f"不可达节点 {len(un_nodes)}；结局可达 {sum(1 for v in hits.values() if v > 0)}/{len(hits)}")
    print(f"执行（S）：seed={SEED} 第 0 条路径，{len(p)} 步，终局 {node}（{ending}），digest {dg}")
    print(f"          状态 {canon.split('|')[1]}；持有 {canon.split('|')[3]}")
    print(f"度量（S）：张力 MAE {mae}（×1000）≤ 上限 {spec['metrics']['tension_mae_max'] * 1000}；"
          f"结局熵 {ent_milli} milli-bits ≥ 下限 {spec['metrics']['ending_entropy_min_milli_bits']}")
    print(f"          结局分布 {dist}")
    print(f"OK na_16_2_rehearsal：三处消费跑通，写出 out/na16_rehearsal.json（digest={dg}）")
    return 0


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

穷举（S）：覆盖率 1000‰（115/115 条对话边）；不可达节点 0；结局可达 5/5
执行（S）：seed=20261016 第 0 条路径，22 步，终局 e_dark（离港），digest 3976226298
          状态 hui=27,mu=50,shen=90,yan=35；持有 fog_glass,lantern,prayer_bead
度量（S）：张力 MAE 0（×1000）≤ 上限 1000；结局熵 1714 milli-bits ≥ 下限 1500
          结局分布 {'离港': 10074, '沉没': 5507, '缄默': 3223, '灯守': 1018, '真相': 178}
OK na_16_2_rehearsal：三处消费跑通，写出 out/na16_rehearsal.json（digest=3976226298）

// file: core/test/na_16_2_rehearsal.test.ts
// tier: P
// verify: node --test 'core/test/na_16_2_rehearsal.test.ts'
// 卷一 16.2：P 层复算 S 层预演路径的状态摘要，digest 与 out/na16_rehearsal.json 逐位相同。
// 先跑 /opt/homebrew/bin/python3 sim/na_16_2_rehearsal.py 生成该产物。

import test from "node:test";
import assert from "node:assert/strict";
import { readFileSync, existsSync } from "node:fs";
import { join, dirname } from "node:path";
import { fileURLToPath } from "node:url";

import { replay, canonical, digest, samplePaths, initState, applyChoice, AFF_LO, AFF_HI } from "../src/inkwell/engine.ts";
import type { Spec } from "../src/inkwell/engine.ts";

const HERE = dirname(fileURLToPath(import.meta.url));
const ROOT = join(HERE, "../..");
const spec = JSON.parse(readFileSync(join(ROOT, "config/story_spec.json"), "utf8")) as Spec;
const ART = join(ROOT, "out/na16_rehearsal.json");

type Row = { seed: number; path: string[]; canonical: string; digest: number; node: string; ending: string };

function load(): Row {
  assert.ok(existsSync(ART), "先跑 /opt/homebrew/bin/python3 sim/na_16_2_rehearsal.py");
  return JSON.parse(readFileSync(ART, "utf8")) as Row;
}

test("S 层与 P 层对同一路径给出逐位相同的 digest", () => {
  const py = load();
  const s = replay(spec, py.path);
  assert.equal(s.node, py.node, "终局节点不一致");
  assert.equal(digest(s), py.digest, "digest 不一致");
  console.log(`P 层：${py.path.length} 步 → ${s.node}（${py.ending}），digest ${digest(s)}`);
  console.log(`      状态 ${canonical(s).split("|")[1]}`);
});

test("两层的采样同源：P 层第一条路径与 S 层逐条相同", () => {
  const py = load();
  const p0 = samplePaths(spec, 1, py.seed)[0];
  assert.deepEqual(p0, py.path, "PRNG 或选路逻辑与 S 层分歧");
  console.log(`两层首条路径 ${py.path.length} 步：${py.path.slice(0, 4).join(",")}…${py.path.slice(-2).join(",")}`);
});

test("预演路径全程满足整数不变量：好感度始终落在 [0,100]", () => {
  const py = load();
  let s = initState(spec);
  let lo = AFF_HI;
  let hi = AFF_LO;
  for (const cid of py.path) {
    s = applyChoice(spec, s, cid);
    for (const k of Object.keys(s.aff)) {
      lo = Math.min(lo, s.aff[k]);
      hi = Math.max(hi, s.aff[k]);
    }
  }
  assert.ok(lo >= AFF_LO && hi <= AFF_HI, `好感度越界 [${lo},${hi}]`);
  console.log(`全程好感度范围 [${lo},${hi}] ⊆ [${AFF_LO},${AFF_HI}]`);
});

P 层：22 步 → e_dark（离港），digest 3976226298
      状态 hui=27,mu=50,shen=90,yan=35
两层首条路径 22 步：c_pier_1b,c_pier_2b,c_chapel_1a,c_chapel_2a…c_dq_1a,c_dq_2e
全程好感度范围 [20,90] ⊆ [0,100]
✔ S 层与 P 层对同一路径给出逐位相同的 digest (44.220916ms)
✔ 两层的采样同源：P 层第一条路径与 S 层逐条相同 (2.666334ms)
✔ 预演路径全程满足整数不变量：好感度始终落在 [0,100] (3.474333ms)
ℹ tests 3
ℹ suites 0
ℹ pass 3
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 284.909375