14.3 回归与告警
站内服务端的叙事分支状态架构
用版本号与幂等重放保证状态在服务端不乱，那是架构层的正确性。本节是引擎层的回归：规格不是一次写死的，它会被人改——补一句台词、调一个 add、动一个阈值。每改一次，昨天还成立的结论就可能不再成立。本节把「当前规格算出来的东西」存成回归基线，让改动可见、可告警。全程不涉及网络、数据库、并发，只在一份规格与一个内核里跑。
基线是什么：三个量
基线不是一个数字，而是三个不同粒度的量：
| 量 | 定义 | 作用 |
|---|---|---|
摘要 summary_digest | 固定批次（N 条同种子路径）终局状态摘要的 FNV-1a 汇总 | 最敏感：任何状态变化都改它 |
分布 ending_distribution | 同一批路径的 5 结局计数 | 定位：告诉你哪个结局动了 |
残差 residual | 整数不变量残差（越界 + 静态死路 + 零选项状态 + 缺失结局） | 判真伪：规格是否已内部破掉 |
三者的敏感度与信息量相反：摘要在最前面报警（什么都可能触发它），分布紧随其后（指出漂移方向），残差最后（只在规格真的坏了时才非零）。所以三类告警不是互斥的——一次改动常常同时触发摘要失配与分布漂移；残差非零则是最严重的信号。
三类告警
| 告警 | 触发条件 | 强度 |
|---|---|---|
summary_mismatch | summary_digest 与基线不同 | 软 |
distribution_drift | 任一结局占比偏离基线超过 5‰ | 软 |
residual_nonzero | 整数不变量残差 != 0 | 软 |
三类都是软告警：打印 WARN、exit 0，与 B 层口径一致——因为「分布变了」不等于「规格错了」，它可能正是这次改动的目的。真正该硬失败的，是负例测试：如果注入一个已知破坏，告警却一声不响，那才必须 exit 1。
S 层：建立基线并比对
S 层脚本先看基线文件在不在：不在就写一份、在就读进来比对；随后跑三组内存内注入负例，演示三类告警各自会开火：
#!/usr/bin/env python3
# file: sim/ne_14_3_regression.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_14_3_regression.py
"""卷二 14.3 回归与告警（S 层）。
把「基线摘要 + 结局分布 + 残差」落成回归基线文件，再跑一次比对：
· 摘要摘要 summary_digest —— 固定批次（N 条同种子路径）终局状态摘要的 FNV-1a 汇总；
· 结局分布 ending_distribution —— 同一批路径的 5 结局计数；
· 残差 residual —— 整数不变量残差（好感度越界 + 静态死路 + 零选项状态 + 缺失结局），恒为 0。
三类告警（**软告警**：打印 WARN、exit 0，与 B 层口径一致）：
① 摘要失配 summary_mismatch —— summary_digest 与基线不同；
② 分布漂移 distribution_drift —— 任一结局占比偏离基线超过阈值（千分比）；
③ 残差非零 residual_nonzero —— 整数不变量残差 != 0。
同时跑三组**内存内**注入负例（不改 config/story_spec.json），演示三类告警确实会开火。
随机一律自实现可播种 PRNG（Lehmer MINSTD），禁用语言内置 random。
"""
from __future__ import annotations
import json
import os
import sys
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import na_00_spec_check as base # noqa: E402 复用 load_spec / eval_cond / apply_effects
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
BASELINE = os.path.join(OUT, "ne_14_3_baseline.json")
N = 2000
SEED = 20261016
STEP_BUDGET = 200
DIST_DRIFT_PERMILLE_MAX = 5
def init_state(spec: dict) -> dict:
aff = {v: next(c for c in spec["characters"] if c["id"] == v)["affinity"]["init"]
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 canonical(s: dict) -> str:
aff = ",".join(f"{k}={s['aff'][k]}" for k in sorted(s["aff"]))
fl = ",".join(k for k in sorted(s["flags"]) if s["flags"][k])
it = ",".join(k for k in sorted(s["items"]) if s["items"][k])
return f"{s['node']}|{aff}|{fl}|{it}|{'>'.join(s['path'])}"
def digest(s: dict) -> int:
h = 2166136261
for ch in canonical(s):
h = ((h ^ ord(ch)) * 16777619) & 0xFFFFFFFF
return h
def measure(spec: dict, n: int = N, seed: int = SEED) -> dict:
"""跑一批同种子路径，返回 (摘要, 分布, 残差) 三元组。"""
by = {x["id"]: x for x in spec["dialogue"]}
start = spec["meta"]["start"]
rng = seed
def rnd(mod: int) -> int:
nonlocal rng
rng = (rng * 48271) % 2147483647
return rng % mod
dist = {e["name"]: 0 for e in spec["endings"]}
aff_bad = 0
zero_opt = 0
digests = []
for _ in range(n):
s = init_state(spec)
while True:
cur = by.get(s["node"])
if cur is None or cur.get("ending"):
if cur is not None:
dist[cur["ending"]] += 1
digests.append(digest(s))
break
opts = [c for c in (cur.get("choices") or []) if base.eval_cond(c.get("when"), s)]
if not opts:
zero_opt += 1
break
c = opts[rnd(len(opts))]
ns = base.apply_effects(c.get("effects") or {}, {"aff": s["aff"], "flags": s["flags"], "items": s["items"]})
s = {"node": c["goto"], "aff": ns["aff"], "flags": ns["flags"], "items": ns["items"],
"path": [*s["path"], c["goto"]]}
for v in spec["state"]["vars"]:
if not (0 <= s["aff"][v] <= 100):
aff_bad += 1
summary = 2166136261
for d in digests:
for ch in str(d):
summary = ((summary ^ ord(ch)) * 16777619) & 0xFFFFFFFF
static_dead = sum(1 for x in spec["dialogue"] if not (x.get("choices") or []) and not x.get("ending"))
missing = sum(1 for e in spec["endings"] if dist[e["name"]] == 0)
residual = aff_bad + static_dead + zero_opt + missing
return {"summary_digest": summary, "ending_distribution": dist,
"residual": residual, "n": n, "seed": seed,
"residual_parts": {"affinity_out_of_range": aff_bad, "static_dead_ends": static_dead,
"zero_option_states": zero_opt, "missing_endings": missing}}
def drift_permille(base_dist: dict, cur_dist: dict, n: int) -> int:
worst = 0
for k in base_dist:
d = abs(cur_dist.get(k, 0) - base_dist[k]) * 1000 // max(1, n)
if d > worst:
worst = d
return worst
def compare(base_metrics: dict, cur: dict) -> list:
"""返回告警列表（软告警，不影响 exit code）。"""
alarms = []
if cur["summary_digest"] != base_metrics["summary_digest"]:
alarms.append(("summary_mismatch", f"基线 {base_metrics['summary_digest']} → 现 {cur['summary_digest']}"))
drift = drift_permille(base_metrics["ending_distribution"], cur["ending_distribution"], cur["n"])
if drift > DIST_DRIFT_PERMILLE_MAX:
alarms.append(("distribution_drift", f"最大占比偏离 {drift}‰（阈值 {DIST_DRIFT_PERMILLE_MAX}‰）"))
if cur["residual"] != 0:
alarms.append(("residual_nonzero", f"残差 {cur['residual']} {cur['residual_parts']}"))
return alarms
def mutate_add(spec: dict, choice_id: str, key: str, delta: int) -> dict:
clone = json.loads(json.dumps(spec))
for x in clone["dialogue"]:
for c in (x.get("choices") or []):
if c["id"] == choice_id:
c.setdefault("effects", {}).setdefault("add", {})[key] += delta
return clone
def mutate_when(spec: dict, choice_id: str, new_when) -> dict:
clone = json.loads(json.dumps(spec))
for x in clone["dialogue"]:
for c in (x.get("choices") or []):
if c["id"] == choice_id:
c["when"] = new_when
return clone
def mutate_init(spec: dict, char_id: str, val: int) -> dict:
clone = json.loads(json.dumps(spec))
for c in clone["characters"]:
if c["id"] == char_id:
c["affinity"]["init"] = val
return clone
def main() -> int:
spec = base.load_spec()
cur = measure(spec)
# ---- 建立或读取基线 ----
if not os.path.exists(BASELINE):
os.makedirs(OUT, exist_ok=True)
with open(BASELINE, "w", encoding="utf-8") as fh:
json.dump(cur, fh, ensure_ascii=False, indent=2)
fh.write("\n")
print(f"基线不存在 → 已建立 {os.path.relpath(BASELINE, ROOT)}（{os.path.getsize(BASELINE)} 字节）")
base_metrics = cur
else:
with open(BASELINE, encoding="utf-8") as fh:
base_metrics = json.load(fh)
print(f"基线 {os.path.relpath(BASELINE, ROOT)}（{os.path.getsize(BASELINE)} 字节，n={base_metrics['n']}，seed={base_metrics['seed']}）")
# ---- 正常比对 ----
alarms = compare(base_metrics, cur)
print(f"当前：摘要 {cur['summary_digest']} · 分布 {cur['ending_distribution']} · 残差 {cur['residual']}")
if alarms:
for kind, detail in alarms:
print(f"WARN {kind}：{detail}")
else:
print("比对通过：摘要一致、分布无漂移、残差 0")
# ---- 三组注入负例：证明三类告警会开火（内存内，不改 config） ----
demos = [
("摘要失配", mutate_add(spec, "c_pier_1b", "hui", 1)),
("分布漂移", mutate_when(spec, "c_dq_2b", None)),
("残差非零", mutate_init(spec, "yan", 150)),
]
fired = 0
for label, mutated in demos:
m = measure(mutated)
al = compare(cur, m)
kinds = [k for k, _ in al]
print(f"负例[{label}] 摘要 {m['summary_digest']} · 残差 {m['residual']} → 告警 {kinds}")
for kind, detail in al:
print(f"WARN {kind}：{detail}")
if al:
fired += 1
print(f"三组负例触发告警 {fired}/3（软告警：全部 WARN，exit 0）")
if fired < 3:
print("FAIL 有负例未触发任何告警，告警器形同虚设")
return 1
print("OK ne_14_3_regression：基线可比对、三类告警均可触发（软告警不硬失败）")
return 0
if __name__ == "__main__":
sys.exit(main())
本机实跑 /opt/homebrew/bin/python3 sim/ne_14_3_regression.py（基线已存在）：
基线 out/ne_14_3_baseline.json（351 字节，n=2000，seed=20261016）
当前：摘要 4205648192 · 分布 {'灯守': 117, '离港': 1032, '真相': 21, '缄默': 304, '沉没': 526} · 残差 0
比对通过：摘要一致、分布无漂移、残差 0
负例[摘要失配] 摘要 3795856790 · 残差 0 → 告警 ['summary_mismatch']
WARN summary_mismatch：基线 4205648192 → 现 3795856790
负例[分布漂移] 摘要 734158025 · 残差 0 → 告警 ['summary_mismatch', 'distribution_drift']
WARN summary_mismatch：基线 4205648192 → 现 734158025
WARN distribution_drift：最大占比偏离 305‰（阈值 5‰）
负例[残差非零] 摘要 1543395875 · 残差 17227 → 告警 ['summary_mismatch', 'distribution_drift', 'residual_nonzero']
WARN summary_mismatch：基线 4205648192 → 现 1543395875
WARN distribution_drift：最大占比偏离 32‰（阈值 5‰）
WARN residual_nonzero：残差 17227 {'affinity_out_of_range': 17227, 'static_dead_ends': 0, 'zero_option_states': 0, 'missing_endings': 0}
三组负例触发告警 3/3（软告警：全部 WARN，exit 0）
OK ne_14_3_regression：基线可比对、三类告警均可触发（软告警不硬失败）
正常比对通过：摘要 4205648192 一致、分布无漂移、残差 0。三组负例各触发告警：①改一个 add 值（2→3）→ summary_mismatch；②把真相结局的门 c_dq_2b 改成无条件 → distribution_drift 305‰；③把 yan 初值改成 150 → residual_nonzero 17227。三组全部 exit 0——软告警不硬失败。
基线文件
基线文件是纯 JSON，体积 351 字节，落在 out/ne_14_3_baseline.json：
{
"summary_digest": 4205648192,
"ending_distribution": {
"灯守": 117,
"离港": 1032,
"真相": 21,
"缄默": 304,
"沉没": 526
},
"residual": 0,
"n": 2000,
"seed": 20261016,
"residual_parts": {
"affinity_out_of_range": 0,
"static_dead_ends": 0,
"zero_option_states": 0,
"missing_endings": 0
}
}
改一个 add 值就会让 summary_digest 从 4205648192 跳到 3795856790——摘要失配是最灵敏的探针。但它只告诉你「变了」，不告诉你「哪里变」。要定位，看分布漂移：把真相结局的门去掉后，distribution_drift 报到 305‰，一眼就知道偏的是真相那条线。要判断「规格是不是真的坏了」，看残差：把 yan 初值设成 150 后，残差跳到 17227（全部来自 affinity_out_of_range），其余三项仍为 0——残差把「哪一类不变量被破坏」也一并报了出来。
一个细节值得写清：分布漂移的阈值必须高于采样噪声。本节固定 2000 条同种子路径，同种子同规格两次跑出的分布逐项相同（摘要也相同），所以基线比对是零噪声的；阈值 5‰ 只用来挡「改一处导致的小抖动」，不挡「换种子重采样」——后者本来就该重算基线。要更稳就把 N 调大，代价是跑得更慢。
告警的强度分级
| 层 | 行为 | 为什么 |
|---|---|---|
| 负例测试 | 必须 硬 fail（exit 1） | 注入已知破坏却无声，是告警器失效，绝不能放过 |
| 基线比对 | 软告警（WARN，exit 0） | 分布变化可能是改动的目的，硬失败会挡住正常迭代 |
这条分级与全书一致：能被证明是错的（负例）硬失败，只是变了（基线）软告警。把两者倒过来，要么让告警器永远沉默，要么让每次正常改动都被拦下。
自检表
| 检查项 | 合格标准 |
|---|---|
| 基线三量 | 摘要 / 分布 / 残差都落盘，粒度互补 |
| 零噪声 | 同种子同规格两次比对逐项相同 |
| 三类告警可触发 | 摘要失配 / 分布漂移 / 残差非零各有一个注入负例 |
| 软硬分级 | 基线比对 WARN + exit 0，负例硬 fail |
| 真源不动 | 注入在克隆体上做，story_spec.json 零改动 |
基线让「改动可见」，告警让「改动可判」。第 14 章到此把内核的可测试性这条线走完：属性测试管性质、模糊测试管随机输入、回归基线管改动。下一章换一个维度——规格变大之后，求值与遍历的开销会怎么长。
阅读导航：上一节：14.2 随机路径的模糊测试 · 下一节：15.1 规格规模的增长 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。