4.1 选择序列与路径

#!/usr/bin/env python3
# file: sim/ne_04_1_path.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_04_1_path.py
"""卷二 4.1 选择序列与路径（S 层）：一条路径有两种表示——玩家按下的是「选择序列」
（一串 choice_id），引擎走出来的却是「路径」（一串 node_id）。本节给出双表示对照、
115->56 的汇聚观察，以及对话图是 DAG 的拓扑证据。产出 out/ne_04_1_path.csv。
PRNG 用自实现 Lehmer MINSTD，与 core/src/inkwell/engine.ts 同算法同种子。
"""
from __future__ import annotations

import csv
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")
CSV_PATH = os.path.join(OUT, "ne_04_1_path.csv")
SEED = 20261016


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_one(spec: dict, seed: int) -> list[str]:
    """Lehmer MINSTD 采样一条完整路径，返回选择序列（choice_id 列表）。"""
    rng, s, trail = seed, init_state(spec), []
    for _ 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"])
    return trail


def trace(spec: dict, choices: list[str]) -> list[dict]:
    """把选择序列展开成逐步的 (选择, 起点, 终点) 三元组。"""
    s, rows = init_state(spec), []
    for step, cid in enumerate(choices, 1):
        frm = s["node"]
        s = apply_choice(spec, s, cid)
        rows.append({"step": step, "choice_id": cid, "from_node": frm, "to_node": s["node"]})
    return rows


def topo_sort(adj: dict) -> list[str]:
    """Kahn 拓扑排序；返回序列长度 < 节点数即说明图有环。"""
    indeg = {k: 0 for k in adj}
    for u in adj:
        for v in adj[u]:
            indeg[v] += 1
    ready = sorted(k for k in adj if indeg[k] == 0)
    order: list[str] = []
    while ready:
        u = ready.pop(0)
        order.append(u)
        for v in adj[u]:
            indeg[v] -= 1
            if indeg[v] == 0:
                ready.append(v)
                ready.sort()
    return order


def main() -> int:
    spec = load_spec()
    adj = {n["id"]: sorted({c["goto"] for c in (n.get("choices") or [])}) for n in spec["dialogue"]}
    edges = {(n["id"], c["goto"]) for n in spec["dialogue"] for c in (n.get("choices") or [])}
    n_nodes = len(adj)
    n_choices = sum(len(n.get("choices") or []) for n in spec["dialogue"])

    choices = sample_one(spec, SEED)
    rows = trace(spec, choices)
    nodes = [spec["meta"]["start"], *(r["to_node"] for r in rows)]
    print(f"spec={spec['meta']['title']}  起点 {spec['meta']['start']}  seed={SEED}")
    print(f"1) 样例路径：选择序列 {len(choices)} 步 -> 节点路径 {len(nodes)} 个节点")
    print(f"   选择序列（choice_id）：{' > '.join(choices)}")
    print(f"   节点路径（node_id）  ：{' > '.join(nodes)}")
    print(f"   终点 {nodes[-1]}  是否结局：{bool(node_by_id(spec, nodes[-1]).get('ending'))}")

    print(f"2) 规模：{n_nodes} 个对话节点，{n_choices} 条选项，去重后唯一跳转 {len(edges)} 条")
    print(f"   选项数 / 唯一跳转 = {n_choices} / {len(edges)}（多条不同选项汇到同一后继节点）")

    order = topo_sort(adj)
    is_dag = len(order) == n_nodes
    print(f"3) Kahn 拓扑排序：排入 {len(order)} / {n_nodes} 个节点；图是 DAG：{is_dag}")
    print(f"   拓扑序前 8：{order[:8]}")
    print(f"   拓扑序末 5：{order[-5:]}")

    known = {n["id"] for n in spec["dialogue"]}
    assert all(x in known for x in nodes) and is_dag, "路径含未知节点或图有环"

    os.makedirs(OUT, exist_ok=True)
    with open(CSV_PATH, "w", encoding="utf-8", newline="") as fh:
        w = csv.DictWriter(fh, fieldnames=["step", "choice_id", "from_node", "to_node"], lineterminator="\n")
        w.writeheader()
        w.writerows(rows)
    print(f"wrote out/ne_04_1_path.csv rows={len(rows)}")
    print("OK ne_04_1_path：双表示对照已输出，115->56 汇聚确认，对话图为 DAG")
    return 0


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

spec=雾港夜航  起点 d_pier_1  seed=20261016
1) 样例路径：选择序列 22 步 -> 节点路径 23 个节点
   选择序列（choice_id）：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
   节点路径（node_id）  ：d_pier_1 > d_pier_2 > d_chapel_1 > d_chapel_2 > d_chapel_3 > d_hub_a1 > d_wp_1 > d_wp_2 > d_wp_3 > d_hub_a2 > d_mr_1 > d_mr_2 > d_mr_3 > d_db_1 > d_db_2 > d_db_3 > d_hub_b2 > d_bw_1 > d_bw_2 > d_bw_3 > d_dq_1 > d_dq_2 > e_dark
   终点 e_dark  是否结局：True
2) 规模：48 个对话节点，115 条选项，去重后唯一跳转 56 条
   选项数 / 唯一跳转 = 115 / 56（多条不同选项汇到同一后继节点）
3) Kahn 拓扑排序：排入 48 / 48 个节点；图是 DAG：True
   拓扑序前 8：['d_pier_1', 'd_pier_2', 'd_chapel_1', 'd_chapel_2', 'd_chapel_3', 'd_tavern_1', 'd_tavern_2', 'd_tavern_3']
   拓扑序末 5：['e_dark', 'e_lit', 'e_loss', 'e_silence', 'e_truth']
wrote out/ne_04_1_path.csv rows=22
OK ne_04_1_path：双表示对照已输出，115->56 汇聚确认，对话图为 DAG

step,choice_id,from_node,to_node
1,c_pier_1b,d_pier_1,d_pier_2
...
22,c_dq_2e,d_dq_2,e_dark

// file: core/test/ne_04_1_path.test.ts
// tier: P
// verify: node --test 'core/test/ne_04_1_path.test.ts'
// 卷二 4.1（P 层门禁）：双表示一致、115->56 汇聚、对话图是 DAG、与 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 { nodeById } from "../src/inkwell/engine.ts";
import type { Spec } from "../src/inkwell/engine.ts";
import { pathOf, trace, adjacency, uniqueGotos, topoSort, firstSequence } from "../src/inkwell/ne_04_1_path.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 = 20261016;

test("选择序列与节点路径是同一趟推进的两种表示", () => {
  const choices = firstSequence(spec, SEED);
  const path = pathOf(spec, choices);
  assert.equal(path[0], spec.meta.start, "路径首节点应为起点");
  assert.equal(path.length, choices.length + 1, "路径节点数 = 选择步数 + 1");
  const steps = trace(spec, choices);
  for (let i = 0; i < steps.length; i += 1) {
    assert.equal(steps[i].from, path[i], `第 ${i} 步起点错`);
    assert.equal(steps[i].to, path[i + 1], `第 ${i} 步终点错`);
  }
  assert.ok(nodeById(spec, path[path.length - 1]).ending, "终点应为结局节点");
  for (const n of path) assert.ok(spec.dialogue.some((d) => d.id === n), `未知节点 ${n}`);
});

test("115 条选项去重后只有 56 条唯一跳转", () => {
  const nChoices = spec.dialogue.reduce((a, n) => a + (n.choices ?? []).length, 0);
  assert.equal(spec.dialogue.length, 48);
  assert.equal(nChoices, 115);
  assert.equal(uniqueGotos(spec.dialogue), 56);
  assert.ok(nChoices > uniqueGotos(spec.dialogue), "选项数应大于唯一跳转数（存在汇聚）");
});

test("对话图是 DAG：Kahn 拓扑排序吃掉全部节点", () => {
  const adj = adjacency(spec.dialogue);
  const order = topoSort(adj);
  assert.equal(order.length, spec.dialogue.length, "拓扑序长度应等于节点数（无环）");
  const pos: Record<string, number> = {};
  order.forEach((k, i) => (pos[k] = i));
  for (const u of Object.keys(adj)) for (const v of adj[u]) assert.ok(pos[u] < pos[v], `边 ${u}->${v} 违反拓扑序`);
});

test("与 Python 侧产物逐条同序（同种子）", () => {
  const rows = readFileSync(join(ROOT, "out/ne_04_1_path.csv"), "utf8")
    .trim().split("\n").slice(1)
    .map((l) => {
      const [step, choice_id, from_node, to_node] = l.split(",");
      return { step: Number(step), choice_id, from_node, to_node };
    });
  const path = pathOf(spec, rows.map((r) => r.choice_id));
  for (const r of rows) {
    assert.equal(path[r.step - 1], r.from_node, `第 ${r.step} 步起点不一致`);
    assert.equal(path[r.step], r.to_node, `第 ${r.step} 步终点不一致`);
  }
});

✔ 选择序列与节点路径是同一趟推进的两种表示 (1.144042ms)
✔ 115 条选项去重后只有 56 条唯一跳转 (0.197291ms)
✔ 对话图是 DAG：Kahn 拓扑排序吃掉全部节点 (0.293416ms)
✔ 与 Python 侧产物逐条同序（同种子） (0.690625ms)
ℹ tests 4
ℹ pass 4
ℹ fail 0