11.3 快照的向前兼容
服务端叙事分支状态架构 里,规格升级通常伴随一次数据迁移脚本,趁低峰期跑一遍、把线上状态重写进新表。单机存档没有那个窗口——玩家手里散着成千上万份旧存档,你发一版新规格,旧的必须还能读。这一节讲的不是「升级时怎么改数据」,而是「新代码读到旧存档时怎么兜住」。
规格只会往前长
规格演化有四种典型形态,且几乎总是往「多」的方向长:
| 演化 | 例子 | 对旧存档的影响 |
|---|---|---|
| 新增 flag | 加一个 knows_signal_legend | 旧存档里没有这个键 |
| 新增物品 | 加一件 signal_flag | 旧存档里没有这件物品 |
| 新增好感度变量 | 加一个角色 lin | 旧存档里没有 lin 的好感度 |
| 新增节点 | 加一个 d_new_1 对话节点 | 旧存档的节点 id 仍然有效 |
迁移函数的三条判据
兼容性靠一个迁移函数兜底,它把「旧存档 + 新规格」映射成「新规格下的完整状态」。判据只有三条,但每条都必须显式定义:
- 缺失字段 → 补默认值。新规格里有、旧存档里没有的键,按类型补默认:flag 补
0、item 补0、好感度变量补该角色的affinity.init。 - 多余字段 → 丢弃并报告。旧存档里有、新规格里没有的键(规格删过字段),丢弃,但要记进
dropped列表,不能悄悄扔。 - 类型不符 → 拒绝。值类型不对(比如 flag 该是
0/1却写成字符串"yes"),直接抛错,不猜测、不强转。
三条判据合起来是一个偏序:能补就补,补不了就丢(但报告),丢都不能丢(类型错)就拒绝。关键是不存在第四种「默默接受」。
S 层:兼容性矩阵
下面这段是 S 层脚本 sim/ne_11_3_forward_compat.py。它由旧规格深拷贝后在内存里加字段造出新规格——不修改 config/story_spec.json——然后把第 10 步的旧存档读进去,跑出兼容性矩阵与三条判据。
#!/usr/bin/env python3
# file: sim/ne_11_3_forward_compat.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_11_3_forward_compat.py
"""卷二 11.3 快照的向前兼容(S 层)。
规格演化后(新增 flag / 新增 item / 新增好感度变量 / 新增节点),旧存档必须仍能读。
把旧规格的存档读进一份**新规格**,给出兼容性矩阵与补默认值规则:
新增 flag → 补 0;新增 item → 补 0;新增 var → 补角色 affinity.init;新增节点 → 无影响。
迁移函数三条判据:缺失补默认值、多余字段丢弃并报告、类型不符拒绝。
新规格由旧规格深拷贝后在内存里加字段得到——**不修改 config/story_spec.json**。
"""
from __future__ import annotations
import copy
import json
import os
import sys
from ne_11_common import OUT, canonical, clamp, load_spec, save_bytes, walk_path
NEW_FLAG, NEW_ITEM, NEW_VAR, NEW_NODE = "knows_signal_legend", "signal_flag", "lin", "d_new_1"
def _int(v, where: str) -> int:
if isinstance(v, bool) or not isinstance(v, int):
raise ValueError(f"type mismatch at {where}: {v!r}")
return v
def migrate(st: dict, new_spec: dict) -> tuple:
"""把旧存档读进新规格:缺失补默认值、多余丢弃并报告、类型不符拒绝。"""
filled, dropped = [], []
out = {"node": st["node"], "aff": {}, "flags": {}, "items": {}, "path": list(st["path"])}
for v in sorted(new_spec["state"]["vars"]):
if v in st["aff"]:
out["aff"][v] = clamp(_int(st["aff"][v], f"aff.{v}"))
else:
init = next(c for c in new_spec["characters"] if c["id"] == v)["affinity"]["init"]
out["aff"][v] = init
filled.append(f"aff:{v}={init}")
dropped += [f"aff:{v}" for v in sorted(set(st["aff"]) - set(new_spec["state"]["vars"]))]
for seg in ("flags", "items"):
for k in sorted(new_spec["state"][seg]):
if k in st[seg]:
out[seg][k] = 1 if _int(st[seg][k], f"{seg}.{k}") else 0
else:
out[seg][k] = 0
filled.append(f"{seg[:-1]}:{k}=0")
dropped += [f"{seg[:-1]}:{k}" for k in sorted(set(st[seg]) - set(new_spec["state"][seg]))]
if st["node"] not in {n["id"] for n in new_spec["dialogue"]}:
raise ValueError(f"unknown node: {st['node']}")
return out, {"filled": filled, "dropped": dropped}
def project(st: dict, spec: dict) -> str:
"""把状态投影回某规格的字段域(跨版本比较用)。"""
aff = ",".join(f"{k}={st['aff'][k]}" for k in sorted(spec["state"]["vars"]))
fl = ",".join(k for k in sorted(spec["state"]["flags"]) if st["flags"].get(k))
it = ",".join(k for k in sorted(spec["state"]["items"]) if st["items"].get(k))
return f"{st['node']}|{aff}|{fl}|{it}|{'>'.join(st['path'])}"
def _extend(spec: dict, seg: str, name: str) -> dict:
n = copy.deepcopy(spec)
n["state"][seg] = sorted(n["state"][seg] + [name])
return n
def add_flag(spec: dict) -> dict:
return _extend(spec, "flags", NEW_FLAG)
def add_item(spec: dict) -> dict:
return _extend(spec, "items", NEW_ITEM)
def add_var(spec: dict) -> dict:
return _extend(spec, "vars", NEW_VAR)
def add_node(spec: dict) -> dict:
n = copy.deepcopy(spec)
scene = next(x for x in n["dialogue"] if x["id"] == "e_lit")["scene"]
n["dialogue"].append({"id": NEW_NODE, "scene": scene, "speaker": "hai", "line": "新节点。",
"line_when": None,
"choices": [{"id": "c_new_1a", "text": "继续", "when": None,
"effects": {}, "goto": "e_lit"}]})
return n
def main() -> int:
old = load_spec()
_, states = walk_path(old)
save = json.loads(save_bytes(states[10]).decode("utf-8"))["state"]
print(f"spec={old['meta']['title']} 旧规格:vars {len(old['state']['vars'])} "
f"flags {len(old['state']['flags'])} items {len(old['state']['items'])} "
f"nodes {len(old['dialogue'])}")
print(f"旧存档(第 10 步):node={save['node']} aff={save['aff']}")
scenarios = [("新增 flag", add_flag), ("新增 item", add_item),
("新增 var", add_var), ("新增节点", add_node)]
print("兼容性矩阵(旧存档 × 新规格):")
print(" 情形 新规格规模 读档结果 补默认值 丢弃")
all_ok = True
for name, fn in scenarios:
new = fn(old)
try:
_, rep = migrate(save, new)
ok = True
except ValueError as e:
ok, rep = False, {"filled": [], "dropped": []}
print(f" {name} 失败:{e}")
all_ok = all_ok and ok
size = (f"v{len(new['state']['vars'])}/f{len(new['state']['flags'])}"
f"/i{len(new['state']['items'])}/n{len(new['dialogue'])}")
print(f" {name:<8} {size:<20} {'成功' if ok else '拒绝':<8} "
f"{(rep['filled'][0] if rep['filled'] else '—'):<18} {rep['dropped']}")
print("判据三例(缺失补默认 / 多余丢弃 / 类型不符拒绝):")
combined = add_node(add_var(add_item(add_flag(old))))
st, rep = migrate(save, combined)
print(f" 缺失补默认:filled={rep['filled']} dropped={rep['dropped']}")
extra = copy.deepcopy(save)
extra["flags"]["legacy_flag"] = 1
extra["items"]["legacy_item"] = 1
_, rep2 = migrate(extra, combined)
print(f" 多余丢弃:dropped={rep2['dropped']} filled={rep2['filled']}")
bad = copy.deepcopy(save)
bad["flags"][NEW_FLAG] = "yes"
try:
migrate(bad, combined)
print(" 类型不符拒绝:未拒绝(FAIL)")
return 1
except ValueError as e:
print(f" 类型不符拒绝:拒绝,ValueError: {e}")
same = project(st, old) == project(save, old)
print(f"迁移未改动旧字段(投影回旧字段域与旧存档逐位相同):{same}")
ok = all_ok and same and len(rep["filled"]) == 3 and rep2["dropped"] == ["flag:legacy_flag", "item:legacy_item"]
if not ok:
print("FAIL 向前兼容矩阵或迁移判据不成立")
return 1
os.makedirs(OUT, exist_ok=True)
with open(os.path.join(OUT, "ne_11_3_compat.json"), "w", encoding="utf-8") as fh:
json.dump({"scenarios": [n for n, _ in scenarios], "combined_filled": rep["filled"],
"dropped": rep2["dropped"], "projection_same": same}, fh,
ensure_ascii=False, indent=2)
fh.write("\n")
print("OK ne_11_3_forward_compat:旧存档读新规格全部成功,迁移判据成立")
return 0
if __name__ == "__main__":
sys.exit(main())
本机实跑 /opt/homebrew/bin/python3 sim/ne_11_3_forward_compat.py:
spec=雾港夜航 旧规格:vars 4 flags 18 items 9 nodes 48
旧存档(第 10 步):node=d_lb_1 aff={'hui': 60, 'mu': 55, 'shen': 50, 'yan': 20}
兼容性矩阵(旧存档 × 新规格):
情形 新规格规模 读档结果 补默认值 丢弃
新增 flag v4/f19/i9/n48 成功 flag:knows_signal_legend=0 []
新增 item v4/f18/i10/n48 成功 item:signal_flag=0 []
新增 var v5/f18/i9/n48 成功 aff:lin=40 []
新增节点 v4/f18/i9/n49 成功 — []
判据三例(缺失补默认 / 多余丢弃 / 类型不符拒绝):
缺失补默认:filled=['aff:lin=40', 'flag:knows_signal_legend=0', 'item:signal_flag=0'] dropped=[]
多余丢弃:dropped=['flag:legacy_flag', 'item:legacy_item'] filled=['aff:lin=40', 'flag:knows_signal_legend=0', 'item:signal_flag=0']
类型不符拒绝:拒绝,ValueError: type mismatch at flags.knows_signal_legend: 'yes'
迁移未改动旧字段(投影回旧字段域与旧存档逐位相同):True
OK ne_11_3_forward_compat:旧存档读新规格全部成功,迁移判据成立
判读:四种演化都读得进
兼容性矩阵四行都「成功」。注意每行的新规格规模:新增 flag 那行 f19(flag 从 18 变 19),新增 item 那行 i10,新增 var 那行 v5,新增节点那行 n49——规模确实变了,说明新规格真被构造出来了,不是空跑。
补默认值一列对应三种「缺字段」:flag:knows_signal_legend=0、item:signal_flag=0、aff:lin=40。其中 lin=40 是查了新规格里 lin 这个角色的 affinity.init 得到的——新增好感度变量补的是角色初始值,不是 0,这条最容易被写错。第四行「新增节点」补默认值是 —:节点是规格里的结构,不是存档里的字段,旧存档不涉及新节点,所以没有可补的字段。
判读:三条判据逐一验证
矩阵下面是三条判据的直接验证:
- 缺失补默认:四种新增同时施加,
filled恰好 3 条(aff:lin=40、flag:knows_signal_legend=0、item:signal_flag=0),dropped为空——因为旧存档没有多余字段。 - 多余丢弃:往旧存档里塞
legacy_flag和legacy_item,迁移后dropped恰好是这两条,filled仍是那 3 条。丢的东西被列了出来,这是可审计的关键。 - 类型不符拒绝:把
knows_signal_legend写成字符串"yes",迁移抛ValueError: type mismatch at flags.knows_signal_legend: 'yes'。不转、不猜、不忽略。
最后一行是本节最重要的一条断言:迁移未改动旧字段——把迁移后的新状态投影回旧字段域(只取旧规格有的 vars/flags/items),与旧存档逐位相同,True。意思是迁移只做了「补新字段」,没有顺手改任何一个旧字段的值。这条保证「读旧存档」不会悄悄篡改玩家已有的进度。
P 层:把兼容性写成测试
分析逻辑在 core/src/inkwell/ne_11_3_forward_compat.ts(migrate / project / addFlag / addItem / addVar / addNode),测试复用 engine.ts 的 applyChoice / canonical,不改动内核。
// file: core/test/ne_11_3_forward_compat.test.ts
// tier: P
// verify: node --test 'core/test/ne_11_3_forward_compat.test.ts'
// 11.3 快照的向前兼容:旧存档读新规格的兼容性矩阵与迁移判据。
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 { applyChoice } from "../src/inkwell/engine.ts";
import type { Spec } from "../src/inkwell/engine.ts";
import { saveBytes } from "../src/inkwell/ne_11_1_save_serialization.ts";
import { walkPath } from "../src/inkwell/ne_11_2_load_restore.ts";
import type { SaveState } from "../src/inkwell/ne_11_2_load_restore.ts";
import { addFlag, addItem, addNode, addVar, migrate, project } from "../src/inkwell/ne_11_3_forward_compat.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 { choices, states } = walkPath(spec);
const save = (JSON.parse(new TextDecoder().decode(saveBytes(states[10]))) as { state: SaveState }).state;
test("兼容性矩阵:旧存档读新规格(新增 flag/item/var/node)全部成功", () => {
const cases: [string, Spec, string][] = [
["新增 flag", addFlag(spec), "flag:knows_signal_legend=0"],
["新增 item", addItem(spec), "item:signal_flag=0"],
["新增 var", addVar(spec), "aff:lin=40"],
["新增节点", addNode(spec), ""],
];
for (const [name, newSpec, expect] of cases) {
const r = migrate(save, newSpec);
if (expect) assert.ok(r.filled.includes(expect), `${name} 应补默认值 ${expect}`);
assert.equal(r.dropped.length, 0, `${name} 不该丢弃字段`);
}
});
test("缺失补默认值:四种新增同时施加,filled 恰为 3 条", () => {
const combined = addNode(addVar(addItem(addFlag(spec))));
const r = migrate(save, combined);
assert.deepEqual(r.filled, ["aff:lin=40", "flag:knows_signal_legend=0", "item:signal_flag=0"]);
assert.deepEqual(r.dropped, []);
});
test("多余字段丢弃并报告", () => {
const combined = addNode(addVar(addItem(addFlag(spec))));
const extra: SaveState = { ...save, flags: { ...save.flags, legacy_flag: 1 }, items: { ...save.items, legacy_item: 1 } };
const r = migrate(extra, combined);
assert.deepEqual(r.dropped, ["flag:legacy_flag", "item:legacy_item"]);
});
test("类型不符拒绝:flag 值写成字符串,迁移抛错", () => {
const combined = addFlag(spec);
const bad: SaveState = { ...save, flags: { ...save.flags, knows_signal_legend: "yes" as unknown as number } };
assert.throws(() => migrate(bad, combined), /type mismatch at flags.knows_signal_legend/);
});
test("新规格下续跑,投影回旧字段与旧终局逐位相同", () => {
const combined = addNode(addVar(addItem(addFlag(spec))));
let s = migrate(save, combined).state;
for (const cid of choices.slice(10)) s = applyChoice(combined, s, cid);
assert.equal(project(s, spec), project(states[states.length - 1], spec));
console.log(`旧存档在新规格下续跑,投影回旧字段与旧终局逐位相同:true`);
});
本机实跑 node --test 'core/test/ne_11_3_forward_compat.test.ts':
旧存档在新规格下续跑,投影回旧字段与旧终局逐位相同:true
✔ 兼容性矩阵:旧存档读新规格(新增 flag/item/var/node)全部成功 (0.818959ms)
✔ 缺失补默认值:四种新增同时施加,filled 恰为 3 条 (0.453542ms)
✔ 多余字段丢弃并报告 (0.102417ms)
✔ 类型不符拒绝:flag 值写成字符串,迁移抛错 (0.28175ms)
✔ 新规格下续跑,投影回旧字段与旧终局逐位相同 (0.515833ms)
ℹ tests 5
ℹ suites 0
ℹ pass 5
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 227.810833
五条全过。Python 与 TS 两侧用同一套迁移判据,所以「四种新增全部成功」「filled 恰为 3 条」「类型不符抛错」在两侧给出同样结论。这是向前兼容的跨语言落点:不只看旧存档能不能读,还要看两侧对「该补什么、该丢什么、该拒绝什么」的判断是否一致。
自检表
| 检查项 | 合格标准 |
|---|---|
| 四种演化可读 | 新增 flag/item/var/node,旧存档全部读入成功 |
| 补默认值正确 | flag/item 补 0,好感度变量补角色 affinity.init |
| 丢弃可报告 | 多余字段进 dropped 列表,不静默丢弃 |
| 类型不符拒绝 | 值类型错误直接抛错,不转不猜 |
| 不改旧字段 | 迁移后投影回旧字段域,与旧存档逐位相同 |
阅读导航:上一节:11.2 读档与状态恢复 · 下一节:12.1 文本表与键 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。