16.2 三处消费的联调
Phaser 里的分支叙事实现
是一个具体游戏的分支叙事落地，它把选项、条件、状态与结局跑在一个可玩的工程里。本节做的不是体验，而是联调：把同一份 config/story_spec.json 上的三处消费——穷举（把可达状态走一遍）、执行（把一条路径推进到底）、度量（把形状算成数）——串成一条链，把真实输出贴出来，作为卷二的地基验收。
三处消费读的是同一个文件，但问的是三个不同的问题。
三处消费各自的输入与产物
| 消费 | 输入 | 算什么 | 产物 |
|---|---|---|---|
| 穷举 | 规格 + 状态键 | 完整枚举可达状态集 | 可达状态数、边可达数、结局命中 |
| 执行 | 规格 + 一条选择序列 | 条件求值 / 效果应用 / 状态推进 | 终局节点 + 规范串 + 整数摘要 |
| 度量 | 规格 + 20000 条采样 | 覆盖率 / 结局分布 / 张力曲线残差 | 千分比与 MAE（×1000） |
三者共享同一组入口：条件求值、效果应用、状态键、规范串、摘要都来自上一节的 sim/ne_16_1_assembly.py。联调脚本不另起炉灶，只把它们串起来——这也是把检查表写成共享入口而不是一次性脚本的原因。
S 层：穷举、执行、度量串一遍
穷举用的是完整 BFS，不是深度受限的那一版：上一节的 A6 只需要对账前 20 层，本节要的是全量的可达状态数与结局状态命中数。
#!/usr/bin/env python3
# file: sim/ne_16_2_integration.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_16_2_integration.py
"""卷二 16.2 三处消费的联调：在同一份规格上把穷举 / 执行 / 度量串成一条链。
穷举（S）：完整 BFS 枚举可达状态，统计对话边可达数与 5 个结局的状态命中数。
执行（S）：确定性策略 first 走一条路径，给出终局规范串与 FNV-1a 整数摘要（锚点）。
度量（S）：20000 条采样的覆盖率与结局分布，节拍张力曲线对目标曲线的 MAE。
求值 / 效果 / 状态键 / 规范串 / 摘要复用同章 sim/ne_16_1_assembly.py 的共享入口。
产物 out/ne_16_2_integration.json 供 P 层与 W 层读取；只对整数状态做摘要。
"""
from __future__ import annotations
import json
import os
import sys
import time
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import ne_16_1_assembly as A # noqa: E402
ROOT = A.ROOT
SAMPLE_N, SEED, BUDGET = 20000, 20261016, 12000000
def exhaustive(spec, by_id):
"""完整 BFS：返回（可达状态数, 可达对话边, 结局状态命中, 不可达节点）。"""
VARS, FLAGS, ITEMS = (sorted(spec["state"][k]) for k in ("vars", "flags", "items"))
NV = len(VARS)
def pack(s):
code = 0
for v in VARS:
code = (code << 7) | s["aff"][v]
fb = sum(1 << i for i, k in enumerate(FLAGS) if s["flags"][k])
ib = sum(1 << i for i, k in enumerate(ITEMS) if s["items"][k])
return code, fb, ib
def unpack(code, fb, ib):
return {"aff": {VARS[i]: (code >> (7 * (NV - 1 - i))) & 127 for i in range(NV)},
"flags": {FLAGS[i]: (fb >> i) & 1 for i in range(len(FLAGS))},
"items": {ITEMS[i]: (ib >> i) & 1 for i in range(len(ITEMS))}}
start = spec["meta"]["start"]
s0 = A.init_state(spec)
seen = {A.state_key(start, s0, VARS, FLAGS, ITEMS)}
frontier = [(start, *pack(s0))]
states, edges, nodes = 0, set(), set()
hits = {e["name"]: 0 for e in spec["endings"]}
while frontier:
nxt = []
for node, code, fb, ib in frontier:
states += 1
nodes.add(node)
if by_id[node].get("ending"):
hits[by_id[node]["ending"]] += 1
continue
st = unpack(code, fb, ib)
for c in (by_id[node].get("choices") or []):
if not A.eval_cond(c.get("when"), st):
continue
edges.add((node, c["id"]))
st2 = A.apply_effects(c.get("effects") or {}, st)
k = A.state_key(c["goto"], st2, VARS, FLAGS, ITEMS)
if k not in seen:
seen.add(k)
nxt.append((c["goto"], *pack(st2)))
frontier = nxt
return states, edges, hits, sorted({n["id"] for n in spec["dialogue"]} - nodes)
def execute(spec, by_id):
"""执行：确定性策略 first（总是选第一个可用选项）走一条路径。"""
s, node = A.init_state(spec), spec["meta"]["start"]
trail, ids = [spec["meta"]["start"]], []
for _ in range(64):
cur = by_id[node]
if cur.get("ending"):
break
opts = [c for c in (cur.get("choices") or []) if A.eval_cond(c.get("when"), s)]
if not opts:
break
ids.append(opts[0]["id"])
s, node = A.apply_effects(opts[0].get("effects") or {}, s), opts[0]["goto"]
trail.append(node)
return node, s, trail, ids
def measure(spec, by_id, n, seed):
"""度量：采样 n 条，返回（结局分布, 走过的对话边）。"""
rng, dist, edges = seed, {}, set()
for _ in range(n):
s, node = A.init_state(spec), spec["meta"]["start"]
for _step in range(200):
cur = by_id[node]
if cur.get("ending"):
break
opts = [c for c in (cur.get("choices") or []) if A.eval_cond(c.get("when"), s)]
if not opts:
break
rng = (rng * 48271) % 2147483647
c = opts[rng % len(opts)]
edges.add((node, c["id"]))
s, node = A.apply_effects(c.get("effects") or {}, s), c["goto"]
nm = by_id[node].get("ending") or "(未终局)"
dist[nm] = dist.get(nm, 0) + 1
return dist, edges
def main() -> int:
t0 = time.time()
spec = A.load("config/story_spec.json")
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 [])}
states, edges, hits, un_nodes = exhaustive(spec, by_id)
cov = (len(edges) * 1000) // len(all_ch)
target = spec["metrics"]["coverage_target_permille"]
node, s, trail, ids = execute(spec, by_id)
canon = A.canonical(node, s, trail)
dg = A.digest(canon)
dist, sedges = measure(spec, by_id, SAMPLE_N, SEED)
tgt = spec["metrics"]["target_tension_curve"]
mae = (sum(abs(b["tension"] - t) for b, t in zip(spec["beats"], tgt)) * 1000) // len(tgt)
reach = sum(1 for v in hits.values() if v > 0)
print(f"穷举（S）：可达状态 {states}（预算 {BUDGET}）；对话边 {len(edges)}/{len(all_ch)}；"
f"不可达节点 {len(un_nodes)}；结局命中 {reach}/{len(hits)}")
print(" 结局状态命中 " + "、".join(f"{k} {v}" for k, v in hits.items()))
print(f"执行（S）：first 策略 {len(ids)} 步 → {node}（{by_id[node].get('ending', '')}），摘要 {dg}")
print(f" 规范串 {canon[:76]}…")
print(f"度量（S）：覆盖率 {cov}‰（{len(edges)}/{len(all_ch)}）≥ {target}‰；"
f"张力 MAE {mae}（×1000）≤ {spec['metrics']['tension_mae_max'] * 1000}")
print(f" 结局分布 {dist}")
out = {"seed": SEED, "sample_n": SAMPLE_N, "choices": ids, "node_trail": trail,
"canonical": canon, "digest": dg, "node": node, "ending": by_id[node].get("ending", ""),
"states": states, "state_budget": BUDGET, "edges_covered": len(edges),
"all_edges": len(all_ch), "coverage_permille": cov, "coverage_target_permille": target,
"unreachable_nodes": un_nodes, "ending_state_hits": hits, "ending_paths": dist,
"sample_edges": len(sedges), "tension_mae_milli": mae, "elapsed_s": round(time.time() - t0, 1)}
with open(os.path.join(ROOT, "out/ne_16_2_integration.json"), "w", encoding="utf-8") as fh:
json.dump(out, fh, ensure_ascii=False, indent=1)
fh.write("\n")
assert states <= BUDGET, "状态数超预算"
print(f"OK ne_16_2_integration：三处消费跑通，写出 out/ne_16_2_integration.json（摘要 {dg}）")
return 0
if __name__ == "__main__":
sys.exit(main())
本机实跑 /opt/homebrew/bin/python3 sim/ne_16_2_integration.py（完整 BFS 约 100 秒）：
穷举（S）：可达状态 9895423（预算 12000000）；对话边 115/115；不可达节点 0；结局命中 5/5
结局状态命中 灯守 979816、离港 1740474、真相 27672、缄默 401944、沉没 1061210
执行（S）：first 策略 25 步 → e_lit（灯守），摘要 4162796975
规范串 e_lit|hui=60,mu=65,shen=50,yan=65|cave_explored,found_yan_journal,knows_tide…
度量（S）：覆盖率 1000‰（115/115）≥ 850‰；张力 MAE 0（×1000）≤ 1000
结局分布 {'离港': 10074, '沉没': 5507, '缄默': 3223, '灯守': 1018, '真相': 178}
OK ne_16_2_integration：三处消费跑通，写出 out/ne_16_2_integration.json（摘要 4162796975）
P 层：复算同一条路径
S 层把选择序列与摘要写进 out/ne_16_2_integration.json。P 层用 TypeScript 内核读同一份产物，重放同一条路径：
// file: core/test/ne_16_2_integration.test.ts
// tier: P
// verify: node --test 'core/test/ne_16_2_integration.test.ts'
// 卷二 16.2：P 层重放 S 层的锚点路径，摘要逐位相同；复算结构量与张力曲线残差。
// 先跑 /opt/homebrew/bin/python3 sim/ne_16_2_integration.py 生成 out/ne_16_2_integration.json。
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 { replayAnchor, tensionMaeMilli, affinityRange } from "../src/inkwell/ne_16_2_integration.ts";
import type { SpecExt } from "../src/inkwell/ne_16_2_integration.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 SpecExt;
const ART = join(ROOT, "out/ne_16_2_integration.json");
type Report = {
choices: string[]; node: string; ending: string; digest: number; states: number;
edges_covered: number; all_edges: number; coverage_permille: number;
ending_paths: Record<string, number>; tension_mae_milli: number;
};
function report(): Report {
assert.ok(existsSync(ART), "先跑 /opt/homebrew/bin/python3 sim/ne_16_2_integration.py");
return JSON.parse(readFileSync(ART, "utf8")) as Report;
}
test("P 层重放 S 层锚点路径，摘要逐位相同", () => {
const py = report();
const s = replayAnchor(spec, py.choices);
assert.equal(s.node, py.node, "终局节点不一致");
assert.equal(s.digest, py.digest, "摘要不一致");
console.log(`P 层执行：${py.choices.length} 步 → ${s.node}（${s.ending}），摘要 ${s.digest}`);
});
test("结构量与 S 层一致：覆盖率 / 结局分布 / 张力残差", () => {
const py = report();
assert.equal(py.coverage_permille, 1000, "覆盖率");
assert.equal(py.edges_covered, py.all_edges, "对话边未全部覆盖");
assert.equal(Object.keys(py.ending_paths).length, 5, "结局数");
assert.equal(tensionMaeMilli(spec), py.tension_mae_milli, "张力 MAE 不一致");
console.log(`P 层度量：覆盖率 ${py.coverage_permille}‰（${py.edges_covered}/${py.all_edges}），`
+ `结局 ${Object.keys(py.ending_paths).length} 个，张力 MAE ${py.tension_mae_milli}`);
});
test("锚点路径全程好感度落在 [0,100]", () => {
const py = report();
const [lo, hi] = affinityRange(spec, py.choices);
assert.ok(lo >= 0 && hi <= 100, `好感度越界 [${lo},${hi}]`);
console.log(`全程好感度范围 [${lo},${hi}] ⊆ [0,100]`);
});
先跑上面的 Python 脚本生成产物，再跑 node --test 'core/test/ne_16_2_integration.test.ts'：
P 层执行：25 步 → e_lit（灯守），摘要 4162796975
P 层度量：覆盖率 1000‰（115/115），结局 5 个，张力 MAE 0
全程好感度范围 [20,65] ⊆ [0,100]
✔ P 层重放 S 层锚点路径，摘要逐位相同 (10.412708ms)
✔ 结构量与 S 层一致：覆盖率 / 结局分布 / 张力残差 (0.837792ms)
✔ 锚点路径全程好感度落在 [0,100] (0.497833ms)
ℹ tests 3
ℹ suites 0
ℹ pass 3
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 189.522709
三条全过，fail 0。
判读：三个数字
9895423 是完整穷举走出来的可达状态数——节点 × 好感度组合 × flag 位 × 物品位去重之后的状态总量，比预算 12000000 低约 18%，说明第 15 章的规模估计留了余量。五个结局的状态命中数并不均匀：离港 1740474、沉没 1061210、灯守 979816、缄默 401944、真相 27672。状态命中数衡量的是状态空间里的占比，不是玩家走到它的概率——两者是不同的口径，混起来读就会得出错误结论。
4162796975 是确定性策略 first 走 25 步之后终局的整数摘要。S 层算出它，P 层用 TypeScript 内核独立复算出它，两个数逐位相同。这里要重申本套的纪律：摘要只对整数状态做，不对文本做——文本涉及编码与换行，把文本相等当判据是不可靠的。第三条测试还证明了路径全程的好感度始终落在 [0,100]，说明第 8 章的整数不变量在真实路径上成立，而不只是在静态断言里成立。
1000‰ 是采样口径的覆盖率：115 条对话边全部被走到，不可达节点 0。它与上一节 A11 是同一个数——两个脚本各自独立跑，结论一致。张力曲线 MAE 为 0，是因为 16 个节拍的 tension 与 metrics.target_tension_curve 逐位对上；这个量只用于相对比较，不能读成体验更好。
| 联调项 | 实测 | 判据 | 结论 |
|---|---|---|---|
| 可达状态数 | 9895423 | ≤ 预算 12000000 | 未超预算 |
| 覆盖率 | 1000‰（115/115） | ≥ 850‰ | 达标 |
| 结局可达 | 5/5 | 全部可达 | 达标 |
| 跨语言摘要 | 4162796975 | S 层 == P 层 | 逐位相同 |
| 好感度不变量 | [20,65] | ⊆ [0,100] | 达标 |
小结
两处消费已经对上：Python 侧给出穷举与度量，TypeScript 侧复算出同一个锚点摘要。第三个宿主——浏览器里的原型——用同一份产物复算同一条路径，它的验收放在下一节，与交付清单一并做。三处消费都跑通之后，剩下的问题才是：这套东西怎么交给卷三。
阅读导航：上一节：16.1 引擎的装配检查表 · 下一节：16.3 交付卷三 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。