11.2 读档与状态恢复

读档后从存档点继续推进,必须与从头一次走完得到逐位相同的终局状态。本节沿 25 步路径在第 1、5、10、15、20、25 步分别存档读档,终局规范串全部相同;并给出负例:把存档里的好感度篡改成 999,宽松读档钳位到 100 并记录修复,严格读档直接拒绝,检测器抓得到。

11.2 读档与状态恢复

服务端叙事分支状态架构 关心的是断线重连与并发写——多个客户端同时改一份状态,得靠数据库事务和锁来保证一致。本节是单机读档:一份存档读回来,从存档点继续推进,得到的终局必须与「从头一次走完」逐位相同。这里没有并发,只有一条确定性路径和一次状态的往返。

读档要回答的两个问题

读档看起来只是「把 11.1 写的 JSON 读回来」,但要真正可用,得回答两个问题。

第一,「存档 → 读档 → 继续」是否等于「一次走完」。存档点在第 10 步,从第 10 步的状态继续应用剩下的 15 个选项,得到的终局必须与从头走 25 步完全相同——节点、好感度、flag、物品、路径全都要对上。如果存档漏了某个字段(比如忘了存物品),或者读档时把某个值算错了,这一步就会分叉。

第二,存档点选在哪里会不会影响结论。第 1 步存、第 10 步存、第 25 步存,读档后都应该走到同一个终局。这条性质保证「存档时机」不是玩家能利用的漏洞——不会出现「第 9 步存和读出来是 A 结局,第 10 步存和读出来是 B 结局」这种事。

S 层:存档点扫描

下面这段是 S 层脚本 sim/ne_11_2_load_restore.py。它沿 25 步路径,在六个位置存档读档,逐个续跑到终局并比对;最后给一个负例:篡改存档里的好感度,看检测器抓不抓得到。

#!/usr/bin/env python3
# file: sim/ne_11_2_load_restore.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_11_2_load_restore.py
"""卷二 11.2 读档与状态恢复(S 层)。

读档后从存档点继续推进,必须与「从头一次走完」得到**逐位相同**的终局状态:
  ① 「存档→读档→继续」与「一次走完」的终局 canonical 串与摘要相同;
  ② 存档点选择(第 1/5/10/15/20/25 步)不影响终局——结论都应相同;
  ③ 负例:篡改存档好感度(越界 999),读档时被钳位(留痕)或拒绝(抛错)。
共用内核见 sim/ne_11_common.py。
"""
from __future__ import annotations

import json
import os
import sys

from ne_11_common import (AFF_HI, AFF_LO, OUT, apply_effects, canonical, clamp,
                          fnv1a, load_spec, save_bytes, walk_path)

SAVE_POINTS = [1, 5, 10, 15, 20, 25]


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 continue_from(spec: dict, s: dict, choices: list) -> dict:
    """从状态 s 起,继续应用 choices(选项 id 必须都在各自节点的可用集合里)。"""
    cur = s
    for cid in choices:
        node = next(n for n in spec["dialogue"] if n["id"] == cur["node"])
        c = next(x for x in (node.get("choices") or []) if x["id"] == cid)
        cur = apply_effects(c.get("effects") or {}, cur, c["goto"])
    return cur


def load_validated(b: bytes) -> tuple:
    """宽松读档:越界钳位并记录 repairs;类型不符拒绝(抛 ValueError)。"""
    st = json.loads(b.decode("utf-8"))["state"]
    repairs, aff, flags, items = [], {}, {}, {}
    for k in sorted(st["aff"]):
        v = _int(st["aff"][k], f"aff.{k}")
        aff[k] = clamp(v)
        if aff[k] != v:
            repairs.append(f"aff:{k}:{v}->{aff[k]}")
    for k in sorted(st["flags"]):
        v = _int(st["flags"][k], f"flags.{k}")
        flags[k] = 1 if v else 0
        if v not in (0, 1):
            repairs.append(f"flag:{k}:{v}->{flags[k]}")
    for k in sorted(st["items"]):
        v = _int(st["items"][k], f"items.{k}")
        items[k] = 1 if v else 0
        if v not in (0, 1):
            repairs.append(f"item:{k}:{v}->{items[k]}")
    return {"node": st["node"], "aff": aff, "flags": flags, "items": items,
            "path": list(st["path"])}, repairs


def load_strict(b: bytes) -> dict:
    """严格读档:任何越界都直接拒绝。"""
    st = json.loads(b.decode("utf-8"))["state"]
    for k in sorted(st["aff"]):
        v = _int(st["aff"][k], f"aff.{k}")
        if not (AFF_LO <= v <= AFF_HI):
            raise ValueError(f"aff out of range: {k}={v}")
    return {"node": st["node"], "aff": dict(st["aff"]), "flags": dict(st["flags"]),
            "items": dict(st["items"]), "path": list(st["path"])}


def main() -> int:
    spec = load_spec()
    choices, states = walk_path(spec)
    full, dig = states[-1], None
    dig = fnv1a(canonical(full).encode())
    print(f"spec={spec['meta']['title']}  步数 {len(choices)}  终局 {full['node']}")
    print(f"一次走完终局规范串 {canonical(full)}")
    print(f"一次走完终局摘要 FNV-1a {dig}")

    print("① 存档→读档→继续 vs 一次走完(存档点取第 10 步):")
    loaded, _ = load_validated(save_bytes(states[10]))
    resumed = continue_from(spec, loaded, choices[10:])
    same = canonical(resumed) == canonical(full)
    print(f"  存档点规范串 {canonical(states[10])}")
    print(f"  读档后继续到终局 {canonical(resumed)}")
    print(f"  终局规范串相同 {same}  摘要相同 {fnv1a(canonical(resumed).encode()) == dig}")

    print("② 存档点选择的影响(在 25 步的不同位置存档,终局都应相同):")
    all_same = True
    for k in SAVE_POINTS:
        loaded, repairs = load_validated(save_bytes(states[k]))
        resumed = continue_from(spec, loaded, choices[k:])
        ok = canonical(resumed) == canonical(full) and not repairs
        all_same = all_same and ok
        print(f"  存档点 step {k:>2}  node={states[k]['node']:<10}  读档后终局 "
              f"{resumed['node']:<8}  与一次走完相同 {ok}")

    print("③ 负例:篡改存档好感度(hui=999),读档必须钳位或拒绝:")
    doc = json.loads(save_bytes(states[25]).decode("utf-8"))
    doc["state"]["aff"]["hui"] = 999
    tampered = json.dumps(doc, ensure_ascii=False, sort_keys=True,
                          separators=(",", ":")).encode("utf-8")
    print("  篡改后存档含 hui=999(越界 [0,100])")
    lenient, repairs = load_validated(tampered)
    print(f"  宽松读档:hui={lenient['aff']['hui']}  repairs={repairs}  "
          f"检测器抓到 {len(repairs) > 0}")
    caught = False
    try:
        load_strict(tampered)
    except ValueError as e:
        caught = True
        print(f"  严格读档:拒绝,抛 ValueError: {e}")
    print(f"  检测器(钳位或拒绝)抓得到:{bool(repairs) and caught}")

    ok = same and all_same and len(repairs) > 0 and caught and lenient["aff"]["hui"] == AFF_HI
    if not ok:
        print("FAIL 读档恢复与一次走完不一致,或篡改未被检出")
        return 1
    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ne_11_2_restore.json"), "w", encoding="utf-8") as fh:
        json.dump({"full": canonical(full), "full_digest": dig,
                   "save_points": SAVE_POINTS, "tamper_repairs": repairs}, fh,
                  ensure_ascii=False, indent=2)
        fh.write("\n")
    print("OK ne_11_2_load_restore:读档恢复逐位一致,篡改被检出")
    return 0


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

本机实跑 /opt/homebrew/bin/python3 sim/ne_11_2_load_restore.py:

spec=雾港夜航  步数 25  终局 e_lit
一次走完终局规范串 e_lit|hui=60,mu=65,shen=50,yan=65|cave_explored,found_yan_journal,knows_tide_rule,knows_yan_missing,met_hai,met_hui,met_mu,saw_lighthouse_light|journal,ledger_page,rope_knot,salt_letter|d_pier_1>d_pier_2>d_tavern_1>d_tavern_2>d_tavern_3>d_hub_a1>d_ch_1>d_ch_2>d_ch_3>d_hub_a2>d_lb_1>d_lb_2>d_lb_3>d_tc_1>d_tc_2>d_tc_3>d_ho_1>d_ho_2>d_ho_3>d_hub_b2>d_lt_1>d_lt_2>d_lt_3>d_dq_1>d_dq_2>e_lit
一次走完终局摘要 FNV-1a 4162796975
① 存档→读档→继续 vs 一次走完(存档点取第 10 步):
  存档点规范串 d_lb_1|hui=60,mu=55,shen=50,yan=20|knows_tide_rule,knows_yan_missing,met_hai,met_hui,met_mu,saw_lighthouse_light|ledger_page,rope_knot|d_pier_1>d_pier_2>d_tavern_1>d_tavern_2>d_tavern_3>d_hub_a1>d_ch_1>d_ch_2>d_ch_3>d_hub_a2>d_lb_1
  读档后继续到终局 e_lit|hui=60,mu=65,shen=50,yan=65|cave_explored,found_yan_journal,knows_tide_rule,knows_yan_missing,met_hai,met_hui,met_mu,saw_lighthouse_light|journal,ledger_page,rope_knot,salt_letter|d_pier_1>d_pier_2>d_tavern_1>d_tavern_2>d_tavern_3>d_hub_a1>d_ch_1>d_ch_2>d_ch_3>d_hub_a2>d_lb_1>d_lb_2>d_lb_3>d_tc_1>d_tc_2>d_tc_3>d_ho_1>d_ho_2>d_ho_3>d_hub_b2>d_lt_1>d_lt_2>d_lt_3>d_dq_1>d_dq_2>e_lit
  终局规范串相同 True  摘要相同 True
② 存档点选择的影响(在 25 步的不同位置存档,终局都应相同):
  存档点 step  1  node=d_pier_2    读档后终局 e_lit     与一次走完相同 True
  存档点 step  5  node=d_hub_a1    读档后终局 e_lit     与一次走完相同 True
  存档点 step 10  node=d_lb_1      读档后终局 e_lit     与一次走完相同 True
  存档点 step 15  node=d_tc_3      读档后终局 e_lit     与一次走完相同 True
  存档点 step 20  node=d_lt_1      读档后终局 e_lit     与一次走完相同 True
  存档点 step 25  node=e_lit       读档后终局 e_lit     与一次走完相同 True
③ 负例:篡改存档好感度(hui=999),读档必须钳位或拒绝:
  篡改后存档含 hui=999(越界 [0,100])
  宽松读档:hui=100  repairs=['aff:hui:999->100']  检测器抓到 True
  严格读档:拒绝,抛 ValueError: aff out of range: hui=999
  检测器(钳位或拒绝)抓得到:True
OK ne_11_2_load_restore:读档恢复逐位一致,篡改被检出

判读:逐位一致与存档点无关

第一部分给出的是核心结论:第 10 步存档,读回来继续走,终局规范串与一次走完逐位相同,摘要 4162796975 一致。存档点那行 d_lb_1|hui=60,mu=55,shen=50,yan=20|... 是第 10 步的中间状态——注意它和终局长得不一样:mu 是 55 不是 65,物品只有两件不是四件。读档的职责不是「让中间状态等于终局」,而是「让中间状态能正确推进到终局」。

第二部分是六个存档点:第 1、5、10、15、20、25 步分别存档读档,读档后终局全部是 e_lit,与一次走完相同。六个 True 说明存档时机不影响终局。第 25 步那行值得注意:它存的已经是终局状态,读档后「继续推进」没有剩余选项,终局就是它自己——这是边界情形,也应正确。

负例:篡改存档必须被检出

第三部分是负例。把存档里的 hui 改成 999(越界,合法范围是 0–100),两种读档策略给出两种正确反应:

策略行为结果
宽松读档钳位到边界并记录修复hui=100,repairs=['aff:hui:999->100']
严格读档越界即拒绝抛 ValueError: aff out of range: hui=999

两种都是「检测器抓得到」——前者把非法值拉回合法域并留痕,后者直接不接受。选哪种取决于产品:玩家手动改存档属于灰色地带,宽松读档更宽容;联机排行榜等场景下,严格读档更安全。关键不是选哪种,而是不能默默接受 999——如果读档既不钳位也不拒绝,那个越界值会一路流进后续的条件求值,让本该走不通的分支被打开。

P 层:把读档写成测试

分析逻辑在 core/src/inkwell/ne_11_2_load_restore.ts(walkPath / continueFrom / loadValidated / loadStrict),测试复用 engine.ts 的 applyChoice / canonical 与 11.1 的 saveBytes,不改动内核。

// file: core/test/ne_11_2_load_restore.test.ts
// tier: P
// verify: node --test 'core/test/ne_11_2_load_restore.test.ts'
// 11.2 读档与状态恢复:存档→读档→继续 与 一次走完 逐位相同;篡改被钳位或拒绝。

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 { canonical } 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 { continueFrom, loadStrict, loadValidated, walkPath } from "../src/inkwell/ne_11_2_load_restore.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 SAVE_POINTS = [1, 5, 10, 15, 20, 25];

test("存档→读档→继续 与 一次走完 得到逐位相同的终局", () => {
  const { choices, states } = walkPath(spec);
  const full = states[states.length - 1];
  const { state, repairs } = loadValidated(saveBytes(states[10]));
  assert.equal(repairs.length, 0);
  assert.equal(canonical(continueFrom(spec, state, choices.slice(10))), canonical(full));
});

test("存档点选择不影响终局:1/5/10/15/20/25 步存档结果都相同", () => {
  const { choices, states } = walkPath(spec);
  const full = canonical(states[states.length - 1]);
  for (const k of SAVE_POINTS) {
    const { state } = loadValidated(saveBytes(states[k]));
    assert.equal(canonical(continueFrom(spec, state, choices.slice(k))), full, `step ${k}`);
  }
});

test("负例:篡改 hui=999,宽松读档钳位到 100 并记录 repairs", () => {
  const { states } = walkPath(spec);
  const doc = JSON.parse(new TextDecoder().decode(saveBytes(states[10]))) as { state: { aff: Record<string, number> } };
  doc.state.aff.hui = 999;
  const blob = new TextEncoder().encode(JSON.stringify(doc));
  const { state, repairs } = loadValidated(blob);
  assert.equal(state.aff.hui, 100);
  assert.deepEqual(repairs, ["aff:hui:999->100"]);
  console.log(`篡改 hui=999:钳位到 ${state.aff.hui},repairs=${repairs.join(",")}`);
});

test("负例:篡改 hui=999,严格读档拒绝", () => {
  const { states } = walkPath(spec);
  const doc = JSON.parse(new TextDecoder().decode(saveBytes(states[10]))) as { state: { aff: Record<string, number> } };
  doc.state.aff.hui = 999;
  const blob = new TextEncoder().encode(JSON.stringify(doc));
  assert.throws(() => loadStrict(blob), /aff out of range: hui=999/);
});

本机实跑 node --test 'core/test/ne_11_2_load_restore.test.ts':

篡改 hui=999:钳位到 100,repairs=aff:hui:999->100
✔ 存档→读档→继续 与 一次走完 得到逐位相同的终局 (1.28475ms)
✔ 存档点选择不影响终局:1/5/10/15/20/25 步存档结果都相同 (2.091542ms)
✔ 负例:篡改 hui=999,宽松读档钳位到 100 并记录 repairs (1.703292ms)
✔ 负例:篡改 hui=999,严格读档拒绝 (0.461375ms)
ℹ tests 4
ℹ suites 0
ℹ pass 4
ℹ fail 0

四条全过,fail 0。Python 与 TS 两侧用同一套读档校验(钳位边界 0–100、repairs 逐条记录、严格模式抛错),所以「篡改 hui=999」在两侧给出同样的 aff:hui:999->100。这是跨语言对照在本节的落点:不只是状态摘要一致,连异常行为的判据也一致。

自检表

检查项合格标准
恢复逐位一致存档→读档→继续 与 一次走完 的终局规范串与摘要相同
存档点无关第 1/5/10/15/20/25 步存档,读档后终局都相同
篡改可检出越界值被钳位(留痕)或拒绝(抛错),不默默接受
跨语言一致两侧读档校验与异常判据一致

读档证明了「同一份状态读回来还是它」。但规格本身会变——加一个 flag、加一个物品、加一个好感度变量。下一节处理旧存档读新规格的向前兼容。

阅读导航:上一节:11.1 存档的序列化 · 下一节:11.3 快照的向前兼容 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「narrative」更多文章

  1. 15.3 规模告警
  2. 15.2 求值与遍历的开销
  3. 15.1 规格规模的增长