4.2 同种子同回放

state = state * 48271 % 2147483647

#!/usr/bin/env python3
# file: sim/ne_04_2_replay.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_04_2_replay.py
"""卷二 4.2 同种子同回放（S 层）：把随机性钉死在一个自实现的 PRNG 上。
采样只用 Lehmer MINSTD（state = state * 48271 % 2147483647），禁用内置 random。
本节证明两件事：同种子同回放（同一 seed 采样两次，500 条序列逐条相同）、
异种子异回放（换 seed 后分布改变）。产出 out/ne_04_2_replay.json。
"""
from __future__ import annotations

import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
JSON_PATH = os.path.join(OUT, "ne_04_2_replay.json")
SEED_A, SEED_B, N = 20261016, 20261017, 500


def load_spec() -> dict:
    with open(os.path.join(ROOT, "config/story_spec.json"), encoding="utf-8") as fh:
        return json.load(fh)


def cond_ok(c, s: dict) -> bool:
    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 cond_ok(c[1], s) and cond_ok(c[2], s)
    if k == "or":
        return cond_ok(c[1], s) or cond_ok(c[2], s)
    if k == "not":
        return not cond_ok(c[1], s)
    raise ValueError(f"unknown cond op: {k}")


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


def node_by_id(spec: dict, nid: str) -> dict:
    return next(n for n in spec["dialogue"] if n["id"] == nid)


def apply_choice(spec: dict, s: dict, cid: str) -> dict:
    n = node_by_id(spec, s["node"])
    c = next(x for x in (n.get("choices") or []) if x["id"] == cid)
    if not cond_ok(c.get("when"), s):
        raise ValueError(f"choice unavailable: {cid}")
    aff, flags, items, eff = dict(s["aff"]), dict(s["flags"]), dict(s["items"]), c.get("effects") or {}
    for k in eff.get("set", {}):
        flags[k] = eff["set"][k]
    for k in eff.get("add", {}):
        aff[k] = max(0, min(100, aff.get(k, 0) + eff["add"][k]))
    for k in eff.get("grant", {}):
        items[k] = eff["grant"][k]
    for k in eff.get("revoke", {}):
        items[k] = 0
    return {"node": c["goto"], "aff": aff, "flags": flags, "items": items,
            "path": [*s["path"], c["goto"]]}


def sample_paths(spec: dict, n: int, seed: int) -> list[list[str]]:
    """Lehmer MINSTD：state = state * 48271 % 2147483647。"""
    rng, out = seed, []
    for _ in range(n):
        s, trail = init_state(spec), []
        for _step in range(200):
            if node_by_id(spec, s["node"]).get("ending"):
                break
            opts = [c for c in (node_by_id(spec, s["node"]).get("choices") or []) if cond_ok(c.get("when"), s)]
            if not opts:
                break
            rng = (rng * 48271) % 2147483647
            c = opts[rng % len(opts)]
            trail.append(c["id"])
            s = apply_choice(spec, s, c["id"])
        out.append(trail)
    return out


def ending_dist(spec: dict, paths: list[list[str]]) -> dict:
    ends: dict[str, int] = {}
    for p in paths:
        s = init_state(spec)
        for cid in p:
            s = apply_choice(spec, s, cid)
        e = node_by_id(spec, s["node"]).get("ending") or "(未终局)"
        ends[e] = ends.get(e, 0) + 1
    return dict(sorted(ends.items()))


def main() -> int:
    spec = load_spec()
    run1 = sample_paths(spec, N, SEED_A)
    run2 = sample_paths(spec, N, SEED_A)
    run3 = sample_paths(spec, N, SEED_B)
    same = run1 == run2
    identical = sum(1 for a, b in zip(run1, run3) if a == b)

    print(f"spec={spec['meta']['title']}  PRNG=Lehmer MINSTD  n={N}")
    print(f"1) 同种子同回放：两次 (seed={SEED_A}) 采样 {N} 条序列逐条相同：{same}")
    print(f"   第一条序列长度：run1={len(run1[0])} run2={len(run2[0])}（逐条相同）")
    print(f"2) 异种子异回放：seed={SEED_B} 与 seed={SEED_A} 逐条相同 {identical} / {N} 条")
    dist_a, dist_b = ending_dist(spec, run1), ending_dist(spec, run3)
    print(f"   seed={SEED_A} 结局分布：{dist_a}")
    print(f"   seed={SEED_B} 结局分布：{dist_b}")
    print(f"3) 前 3 条序列（seed={SEED_A}）：")
    for i, seq in enumerate(run1[:3]):
        print(f"   #{i} {' > '.join(seq)}")
    print(f"   前 3 条序列（seed={SEED_B}）：")
    for i, seq in enumerate(run3[:3]):
        print(f"   #{i} {' > '.join(seq)}")

    assert same and identical < N, "同种子未复现或异种子无差异"

    os.makedirs(OUT, exist_ok=True)
    with open(JSON_PATH, "w", encoding="utf-8") as fh:
        json.dump({"seed_a": SEED_A, "seed_b": SEED_B, "n": N, "same_seed_identical": same,
                   "cross_seed_identical": identical, "dist_seed_a": dist_a, "dist_seed_b": dist_b,
                   "first3_seed_a": run1[:3], "first3_seed_b": run3[:3]}, fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    print("wrote out/ne_04_2_replay.json")
    print("OK ne_04_2_replay：同种子 500/500 复现，异种子分布已改变")
    return 0


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

spec=雾港夜航  PRNG=Lehmer MINSTD  n=500
1) 同种子同回放：两次 (seed=20261016) 采样 500 条序列逐条相同：True
   第一条序列长度：run1=22 run2=22（逐条相同）
2) 异种子异回放：seed=20261017 与 seed=20261016 逐条相同 0 / 500 条
   seed=20261016 结局分布：{'沉没': 143, '灯守': 32, '真相': 7, '离港': 250, '缄默': 68}
   seed=20261017 结局分布：{'沉没': 119, '灯守': 27, '真相': 5, '离港': 262, '缄默': 87}
3) 前 3 条序列（seed=20261016）：
   #0 c_pier_1b > c_pier_2b > c_chapel_1a > c_chapel_2a > c_chapel_3b > c_hub_a1b > c_wp_1b > c_wp_2a > c_wp_3a > c_hub_a2b > c_mr_1a > c_mr_2b > c_mr_3b > c_db_1c > c_db_2a > c_db_3a > c_hub_b2b > c_bw_1c > c_bw_2b > c_bw_3a > c_dq_1a > c_dq_2e
   #1 c_pier_1b > c_pier_2b > c_chapel_1a > c_chapel_2a > c_chapel_3b > c_hub_a1b > c_wp_1b > c_wp_2a > c_wp_3a > c_hub_a2a > c_lb_1a > c_lb_2b > c_lb_3a > c_tc_1b > c_tc_2a > c_tc_3b > c_ho_1b > c_ho_2b > c_ho_3a > c_hub_b2b > c_bw_1c > c_bw_2a > c_bw_3a > c_dq_1a > c_dq_2e
   #2 c_pier_1b > c_pier_2a > c_tavern_1b > c_tavern_2b > c_tavern_3a > c_hub_a1b > c_wp_1b > c_wp_2c > c_wp_3a > c_hub_a2a > c_lb_1a > c_lb_2b > c_lb_3a > c_tc_1a > c_tc_2a > c_tc_3a > c_ho_1b > c_ho_2a > c_ho_3a > c_hub_b2a > c_lt_1b > c_lt_2b > c_lt_3c > c_fg_1b > c_fg_2a > c_fg_3b > c_dq_1c > c_dq_2e
   前 3 条序列（seed=20261017）：
   #0 c_pier_1a > c_pier_2b > c_chapel_1c > c_chapel_2a > c_chapel_3a > c_hub_a1a > c_ch_1a > c_ch_2b > c_ch_3a > c_hub_a2b > c_mr_1b > c_mr_2a > c_mr_3a > c_ho_1a > c_ho_2b > c_ho_3a > c_hub_b2b > c_bw_1a > c_bw_2b > c_bw_3a > c_dq_1c > c_dq_2c
   #1 c_pier_1b > c_pier_2b > c_chapel_1a > c_chapel_2a > c_chapel_3b > c_hub_a1a > c_ch_1a > c_ch_2b > c_ch_3a > c_hub_a2a > c_lb_1a > c_lb_2b > c_lb_3b > c_mr_1b > c_mr_2b > c_mr_3b > c_db_1a > c_db_2a > c_db_3a > c_hub_b2b > c_bw_1b > c_bw_2a > c_bw_3a > c_dq_1c > c_dq_2d
   #2 c_pier_1b > c_pier_2b > c_chapel_1a > c_chapel_2b > c_chapel_3b > c_hub_a1a > c_ch_1a > c_ch_2a > c_ch_3a > c_hub_a2a > c_lb_1a > c_lb_2b > c_lb_3a > c_tc_1a > c_tc_2b > c_tc_3a > c_ho_1b > c_ho_2a > c_ho_3a > c_hub_b2a > c_lt_1b > c_lt_2b > c_lt_3b > c_bw_1a > c_bw_2a > c_bw_3b > c_fg_1c > c_fg_2b > c_fg_3b > c_dq_1c > c_dq_2e
wrote out/ne_04_2_replay.json
OK ne_04_2_replay：同种子 500/500 复现，异种子分布已改变

// file: core/test/ne_04_2_replay.test.ts
// tier: P
// verify: node --test 'core/test/ne_04_2_replay.test.ts'
// 卷二 4.2（P 层门禁）：同种子同回放、异种子异回放、PRNG 序列与 Python 侧对齐。

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

import type { Spec } from "../src/inkwell/engine.ts";
import { lehmaRun, sampleSequences, sequenceSignature } from "../src/inkwell/ne_04_2_replay.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 SEED_A = 20261016;
const SEED_B = 20261017;
const N = 500;

test("同种子同回放：两次采样 500 条序列逐条相同", () => {
  const a = sampleSequences(spec, N, SEED_A);
  const b = sampleSequences(spec, N, SEED_A);
  assert.equal(a.length, N);
  assert.deepEqual(a, b, "同种子应逐条复现");
  assert.equal(sequenceSignature(a), sequenceSignature(b));
});

test("异种子异回放：换种子后签名改变、逐条相同数为 0", () => {
  const a = sampleSequences(spec, N, SEED_A);
  const c = sampleSequences(spec, N, SEED_B);
  let same = 0;
  for (let i = 0; i < N; i += 1) if (a[i].join(">") === c[i].join(">")) same += 1;
  assert.equal(same, 0);
  assert.notEqual(sequenceSignature(a), sequenceSignature(c));
});

test("PRNG 是 Lehmer MINSTD，前几个状态可复算", () => {
  const first = lehmaRun(1, 3);
  assert.deepEqual(first, [(1 * 48271) % 2147483647, (48271 * 48271) % 2147483647, (48271 * 48271 % 2147483647 * 48271) % 2147483647]);
});

test("签名与 Python 侧产物一致（同种子同回放跨语言）", () => {
  const j = JSON.parse(readFileSync(join(ROOT, "out/ne_04_2_replay.json"), "utf8")) as {
    seed_a: number; n: number; same_seed_identical: boolean; first3_seed_a: string[][];
  };
  assert.equal(j.seed_a, SEED_A);
  assert.equal(j.same_seed_identical, true);
  const a = sampleSequences(spec, N, SEED_A);
  for (let i = 0; i < 3; i += 1) assert.deepEqual(a[i], j.first3_seed_a[i], `第 ${i} 条序列不一致`);
});

✔ 同种子同回放：两次采样 500 条序列逐条相同 (34.938541ms)
✔ 异种子异回放：换种子后签名改变、逐条相同数为 0 (21.758042ms)
✔ PRNG 是 Lehmer MINSTD，前几个状态可复算 (0.165042ms)
✔ 签名与 Python 侧产物一致（同种子同回放跨语言） (14.065875ms)
ℹ tests 4
ℹ pass 4
ℹ fail 0