6.2 汇流的合法性
站内的独立游戏的叙事设计与分支对话 给的是分支汇聚的设计经验：哪里该合、哪里该分，配的是清单与示意图，代码不可跑。本节把「该不该合」变成一条能跑出计数的判据：什么样的汇聚算非法——两条边合到一起之后，如果下游再也分不出你是从哪条边进来的，这条信息就白丢了。上一节查的是「走进去出不来」，这一节查的是「走进来之后，两条不同的来路被抹平」。
汇流本身不是问题
先破除一个误解：汇聚不等于缺陷。本规格 115 条选项，按「源节点 → 目标」去重后只剩 56 条唯一跳转；也就是说大量选项其实是同一对节点之间的重复边，最后都收进 47 个不同的目标节点。分支叙事本来就会合流——不然节点数会指数爆炸。合流是压缩，是常态。
所以判据不能是「有没有汇聚」，只能是「汇聚有没有丢信息」。一个节点有多条入边，如果下游还能靠状态把「你是从哪条边来的」区分开，这次汇聚就是合法的；如果区分不开，就是冗余汇流——两条边白合了。
什么算非法汇流：可判定口径
「信息有没有丢」听着主观，但可以写成离散判据。对每个汇聚节点 M：
- 取 M 及其下游闭包里的全部条件——包括选项的
when、台词的line_when、以及可达结局的when——组成判据集 D(M)。这是「下游可能拿来区分来路」的全部依据。 - 对 M 的每条入边 e，遍历所有经 e 到达 M 的状态，把每个状态在 D(M) 上求值成一个位向量（D(M) 里第 j 个条件为真则第 j 位为 1），收集成向量集合。
- 两条入边等价，当且仅当它们的向量集合相等。等价即冗余。
为什么「集合相等」是对的判据？因为下游能看到的只有 D(M) 的求值结果。如果两条入边产生的所有可能求值向量一模一样，那么下游无论怎么写条件，都无法把这两条来路分开——来路信息在合流点就被彻底丢弃了。反之，只要存在一个向量只在一条入边出现，下游就有办法区分。
这条判据只依赖条件求值，不依赖任何主观判断，所以能穷举。代价是状态空间大——但和第 6.1 节一样，投影到条件可读变量后只有 3042869 个可达状态，几秒跑完。
S 层：为每条入边收集求值向量集合
脚本 sim/ne_06_2_merge_legality.py 做三件事：① 扫出所有入边 ≥2 的汇聚节点；② 对每个汇聚节点算下游闭包条件集 D(M)；③ 在投影状态上 BFS，为每条入边收集到达向量集合，再按集合相等分等价类。
#!/usr/bin/env python3
# file: sim/ne_06_2_merge_legality.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_06_2_merge_legality.py
"""卷二 6.2 汇流的合法性（S 层）：哪些汇聚把信息丢了。
汇聚（一个节点有多条入边）本身不是问题——本规格 115 条选项去重后只有 56 条唯一跳转
（按「源节点 -> 目标」去重），汇聚是常态。**冗余汇流**指：两条不同入边到达同一节点时，
其后继全部条件的求值结果完全相同，于是「从哪条边进来」这条信息被丢弃，后续条件无法区分。
可判定口径（写死在代码里）：对每个汇聚节点 M，取 M 及其下游闭包里的**全部条件**（选项
when + 台词 line_when + 可达结局的 when）组成判据集 D(M)；对每条入边 e，收集所有经 e
到达 M 的状态在 D(M) 上的**求值向量集合**；两入边等价 <=> 集合相等，等价即冗余。
用「投影到条件可读变量」的完整 BFS 精确枚举到达状态。产出 out/ne_06_2_merge_legality.csv。
"""
from __future__ import annotations
import csv
import json
import os
import sys
from collections import defaultdict
HERE = os.path.dirname(os.path.abspath(__file__))
AFF_LO, AFF_HI = 0, 100
def read_vars(spec):
rf, ri, ra = set(), set(), set()
def walk(c):
if not c:
return
k = c[0]
if k in ("and", "or"):
walk(c[1])
walk(c[2])
elif k == "not":
walk(c[1])
elif k == "flag":
rf.add(c[1])
elif k == "item":
ri.add(c[1])
elif k == "aff":
ra.add(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["endings"]:
walk(e.get("when"))
return sorted(ra), sorted(rf), sorted(ri)
def main() -> int:
with open(os.path.join(HERE, "../config/story_spec.json"), encoding="utf-8") as fh:
spec = json.load(fh)
nodes = spec["dialogue"]
by = {n["id"]: n for n in nodes}
start = spec["meta"]["start"]
RA, RF, RI = read_vars(spec)
AR = {v: i for i, v in enumerate(RA)}
FR = {v: i for i, v in enumerate(RF)}
IR = {v: i for i, v in enumerate(RI)}
ending_of = {n["id"]: n["ending"] for n in nodes if n.get("ending")}
end_when = {e["name"]: e.get("when") for e in spec["endings"]}
def cond(c, aff, fl, it) -> bool:
if c is None:
return True
k = c[0]
if k == "flag":
return ((fl >> FR[c[1]]) & 1) == c[2]
if k == "item":
return ((it >> IR[c[1]]) & 1) == c[2]
if k == "aff":
v, op, n = aff[AR[c[1]]], c[2], c[3]
return {">=": v >= n, "<": v < n, "==": v == n, "!=": v != n}[op]
if k == "and":
return cond(c[1], aff, fl, it) and cond(c[2], aff, fl, it)
if k == "or":
return cond(c[1], aff, fl, it) or cond(c[2], aff, fl, it)
return not cond(c[1], aff, fl, it)
def step(eff, aff, fl, it):
aff = list(aff)
for k, v in (eff.get("set") or {}).items():
if k in FR:
fl = (fl | (1 << FR[k])) if v else (fl & ~(1 << FR[k]))
for k, v in (eff.get("add") or {}).items():
if k in AR:
nv = aff[AR[k]] + v
aff[AR[k]] = AFF_LO if nv < AFF_LO else AFF_HI if nv > AFF_HI else nv
for k, v in (eff.get("grant") or {}).items():
if k in IR:
it = (it | (1 << IR[k])) if v else (it & ~(1 << IR[k]))
for k in (eff.get("revoke") or {}):
if k in IR:
it &= ~(1 << IR[k])
return tuple(aff), fl, it
in_edges = defaultdict(list)
for n in nodes:
for c in (n.get("choices") or []):
in_edges[c["goto"]].append((n["id"], c["id"]))
merges = {m: e for m, e in in_edges.items() if len(e) >= 2}
n_jumps = len({(n["id"], c["goto"]) for n in nodes for c in (n.get("choices") or [])})
def downstream(m):
seen, stack, conds = set(), [m], []
while stack:
x = stack.pop()
if x in seen:
continue
seen.add(x)
for c in (by[x].get("choices") or []):
if c.get("when"):
conds.append(c["when"])
stack.append(c["goto"])
if by[x].get("line_when"):
conds.append(by[x]["line_when"])
if x in ending_of and end_when.get(ending_of[x]):
conds.append(end_when[ending_of[x]])
return conds
D = {m: downstream(m) for m in merges}
aff0 = tuple(next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v) for v in RA)
seen = {(start, aff0, 0, 0)}
frontier = [(start, aff0, 0, 0)]
arr = defaultdict(set)
while frontier:
nxt = []
for node, aff, fl, it in frontier:
if by[node].get("ending"):
continue
for c in by[node]["choices"]:
if not cond(c.get("when"), aff, fl, it):
continue
a2, f2, i2 = step(c.get("effects") or {}, aff, fl, it)
if c["goto"] in merges:
vec = 0
for j, d in enumerate(D[c["goto"]]):
if cond(d, a2, f2, i2):
vec |= 1 << j
arr[(c["goto"], (node, c["id"]))].add(vec)
k = (c["goto"], a2, f2, i2)
if k not in seen:
seen.add(k)
nxt.append(k)
frontier = nxt
rows, pairs, cross_red = [], [], []
for m, ins in sorted(merges.items()):
sets = {e: frozenset(arr[(m, e)]) for e in ins}
classes = {}
for e in ins:
classes.setdefault(sets[e], []).append(e)
cr = False
for i in range(len(ins)):
for j in range(i + 1, len(ins)):
if sets[ins[i]] == sets[ins[j]]:
pairs.append((m, ins[i], ins[j]))
if ins[i][0] != ins[j][0]:
cr = True
if cr:
cross_red.append(m)
rows.append({"node": m, "in_degree": len(ins), "sources": len({s for s, _ in ins}),
"classes": len(classes), "cross_source": int(len({s for s, _ in ins}) >= 2),
"redundant": int(len(classes) < len(ins))})
out = os.path.join(HERE, "../out")
os.makedirs(out, exist_ok=True)
with open(os.path.join(out, "ne_06_2_merge_legality.csv"), "w", encoding="utf-8", newline="") as fh:
w = csv.DictWriter(fh, fieldnames=list(rows[0]))
w.writeheader()
w.writerows(rows)
n_red = sum(r["redundant"] for r in rows)
print(f"选项 115 条 / 唯一跳转（源->目标去重）{n_jumps} 条")
print(f"汇聚节点 {len(merges)} 个（跨源 {sum(r['cross_source'] for r in rows)} / 同源扇入 "
f"{sum(1 - r['cross_source'] for r in rows)}），投影可达状态 {len(seen)}")
print(f"冗余汇流节点 {n_red} 个 / 等价入边对 {len(pairs)} 对；合法（各入边可区分）{len(merges) - n_red} 个")
print(f"其中跨源冗余汇流 {len(cross_red)} 个：{cross_red}")
print("冗余入边对（同一汇聚节点、到达求值向量集合相同）：")
for m, e1, e2 in pairs:
print(f" {m} {e1[0]}/{e1[1]} == {e2[0]}/{e2[1]}")
print("跨源但可区分（合法汇聚，信息保留）：")
for r in rows:
if r["cross_source"] and not r["redundant"]:
print(f" {r['node']} 入边 {r['in_degree']} 条来自 {r['sources']} 源，分 {r['classes']} 类")
print(f"wrote out/ne_06_2_merge_legality.csv rows={len(rows)}")
print("OK ne_06_2_merge_legality：汇流扫描完成")
return 0
if __name__ == "__main__":
sys.exit(main())
本机实跑 /opt/homebrew/bin/python3 sim/ne_06_2_merge_legality.py：
选项 115 条 / 唯一跳转（源->目标去重）56 条
汇聚节点 34 个（跨源 8 / 同源扇入 26），投影可达状态 3042869
冗余汇流节点 22 个 / 等价入边对 23 对；合法（各入边可区分）12 个
其中跨源冗余汇流 3 个：['d_bw_1', 'd_dq_1', 'd_mr_1']
冗余入边对（同一汇聚节点、到达求值向量集合相同）：
d_bw_1 d_hub_b2/c_hub_b2b == d_lt_3/c_lt_3b
d_bw_2 d_bw_1/c_bw_1a == d_bw_1/c_bw_1b
d_bw_3 d_bw_2/c_bw_2a == d_bw_2/c_bw_2b
d_ch_3 d_ch_2/c_ch_2a == d_ch_2/c_ch_2b
d_chapel_2 d_chapel_1/c_chapel_1b == d_chapel_1/c_chapel_1c
d_chapel_3 d_chapel_2/c_chapel_2a == d_chapel_2/c_chapel_2b
d_db_2 d_db_1/c_db_1a == d_db_1/c_db_1b
d_db_3 d_db_2/c_db_2a == d_db_2/c_db_2b
d_dq_1 d_lt_3/c_lt_3a == d_bw_3/c_bw_3a
d_dq_1 d_fg_3/c_fg_3a == d_fg_3/c_fg_3b
d_fg_2 d_fg_1/c_fg_1a == d_fg_1/c_fg_1b
d_fg_3 d_fg_2/c_fg_2a == d_fg_2/c_fg_2b
d_ho_2 d_ho_1/c_ho_1a == d_ho_1/c_ho_1b
d_ho_3 d_ho_2/c_ho_2a == d_ho_2/c_ho_2b
d_hub_b2 d_db_3/c_db_3a == d_db_3/c_db_3b
d_lb_2 d_lb_1/c_lb_1a == d_lb_1/c_lb_1b
d_lt_2 d_lt_1/c_lt_1b == d_lt_1/c_lt_1c
d_lt_3 d_lt_2/c_lt_2a == d_lt_2/c_lt_2b
d_mr_1 d_hub_a2/c_hub_a2b == d_lb_3/c_lb_3b
d_mr_3 d_mr_2/c_mr_2a == d_mr_2/c_mr_2b
d_tavern_2 d_tavern_1/c_tavern_1b == d_tavern_1/c_tavern_1c
d_tc_2 d_tc_1/c_tc_1a == d_tc_1/c_tc_1b
d_wp_2 d_wp_1/c_wp_1a == d_wp_1/c_wp_1c
跨源但可区分（合法汇聚，信息保留）：
d_fg_1 入边 2 条来自 2 源，分 2 类
d_ho_1 入边 3 条来自 2 源，分 3 类
d_hub_a1 入边 4 条来自 2 源，分 4 类
d_hub_a2 入边 4 条来自 2 源，分 4 类
wrote out/ne_06_2_merge_legality.csv rows=34
OK ne_06_2_merge_legality：汇流扫描完成
结果怎么读
34 个汇聚节点里，22 个冗余、12 个合法。 冗余的绝大多数是同源扇入——同一个节点的两个选项都指向同一目标（比如 d_bw_1/c_bw_1a == d_bw_1/c_bw_1b），这两条边本来就没打算传递不同信息，被判定冗余是预期之内的：它们只是同一段对话的两个措辞，效果一样，合流当然分不开。
真正值得盯的是跨源冗余：两条边来自不同节点，却在下游完全不可区分。本规格有 3 个这样的节点——d_bw_1、d_dq_1、d_mr_1。以 d_bw_1 为例，它的一条入边来自 d_hub_b2、另一条来自 d_lt_3，两条来路在 D(d_bw_1) 上的求值向量集合完全相同，意味着「你是从码头那侧来的、还是从灯塔那侧来的」在 d_bw_1 之后再也影响不了任何条件。这就是信息丢失——设计上如果本想让下游对来路有反应，这里就是 bug。
对照着看合法跨源汇聚：d_hub_a1 有 4 条入边来自 2 个源，却分成 4 类——每条边的到达状态在下游都能被区分，信息完整保留。这就是「合流但不丢信息」的正面样本。
P 层：合成规格负例
「34 个汇聚、22 个冗余」这句话要可信，检测器必须能在构造出来的例子上开火。P 层用一个两节点合成规格做正反两例：a 有两个选项都跳到 m，m 有一个带 when 的选项跳到结局。
- 负例一：
a的两个选项效果相同（都空）→ 到达m的状态集合相同 → 判冗余（classes1 类、redundanttrue）。 - 负例二：给其中一个选项加
set: {f: 1}→ 两条入边到达m的f不同 → 判合法（classes2 类、redundantfalse）。
同一个合成规格，只改一个选项的 effects，判定就从冗余翻成合法——说明检测器读的确实是「到达状态的求值差异」，不是别的。本机实跑 node --test 'core/test/ne_06_2_merge_legality.test.ts'：
汇聚 34，冗余 22，等价入边对 23
✔ 真实规格：34 个汇聚节点，22 个冗余，跨源冗余 3 个 (5888.018459ms)
✔ 负例一 · 两入边效果相同：合成规格的汇聚被判定为冗余 (0.255791ms)
✔ 负例二 · 两入边效果不同：同一合成规格的汇聚不再冗余 (0.735667ms)
ℹ tests 3
ℹ suites 0
ℹ pass 3
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 6178.891292
判读
第一，「冗余」不等于「要改」。 22 个冗余里绝大多数是同源扇入（措辞不同的同一跳转），它们冗余是正常的。判据的价值在于把跨源冗余单独拎出来——只有那 3 个才是「两条不同来路被抹平」的真问题。
第二，判据是保守的。 它只问「下游条件能不能区分」，不预测玩家体验。一个跨源冗余节点可能永远没人注意，也可能正是设计想要的效果。它给的是一份可复核的清单，不是判决。
第三，口径可以收紧或放松。 D(M) 取的是「下游闭包的全部条件」。如果把 line_when 排除，只留选项 when，冗余数会变——因为有些区分只体现在台词层。口径必须写死在代码里、随结论一起报告，否则「22 个」这个数字无从复现。
自检表
| 检查项 | 合格标准 |
|---|---|
| 先破除误解 | 明确汇聚是常态（115 条选项 → 56 条唯一跳转），判据只针对「丢信息」 |
| 口径可判定 | D(M) = 下游闭包全部条件；等价 = 到达求值向量集合相等 |
| 精确非抽样 | 在全部 3042869 个投影可达状态上收集向量，不靠采样 |
| 跨源单列 | 冗余拆成同源扇入与跨源冗余，只后者是信息丢失 |
| 负例必给 | 合成规格改一处 effects 即翻转判定，证明检测器开火 |
汇流查完，还剩最后一问：所有结局都够得着吗？下一节给 5 个结局各找一条真实到达路径，并把「可达」与「易达」分开报。
阅读导航：上一节：6.1 死路检测 · 下一节：6.3 结局的可达性 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。