3.3 状态快照与差量

#!/usr/bin/env python3
# file: sim/ne_03_3_snapshot.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ne_03_3_snapshot.py
"""卷二 3.3 状态快照与差量（S 层）。

沿一条确定性路径（总是选第一个可用选项）走完：快照 = 完整 canonical 串；
差量 = 相邻快照间真正改变的字段集合。产出 out/ne_03_3_delta.csv（逐步骤差量表）。
并把「写而不读」的字段点出来——被 set/grant 却无任何条件读取，在差量里恒为
「只增不减的信息」（判据：声明未用=报告，不判 FAIL）。canonical 与 engine.ts
同算法；整数口径禁浮点。
"""
from __future__ import annotations

import csv
import json
import os
import sys

AFF_LO, AFF_HI = 0, 100
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
OUT = os.path.join(ROOT, "out")
CSV_PATH = os.path.join(OUT, "ne_03_3_delta.csv")


def clamp(v: int) -> int:
    return AFF_LO if v < AFF_LO else AFF_HI if v > AFF_HI else v


def cond_ok(c, s) -> bool:
    if c is None:
        return True
    k = c[0]
    if k == "flag":
        return s["flags"].get(c[1], 0) == c[2]
    if k == "item":
        return s["items"].get(c[1], 0) == c[2]
    if k == "aff":
        v, op, n = s["aff"].get(c[1], 0), c[2], c[3]
        return v >= n if op == ">=" else v < n if op == "<" else v == n if op == "==" else v != n
    if k == "and":
        return cond_ok(c[1], s) and cond_ok(c[2], s)
    if k == "or":
        return cond_ok(c[1], s) or cond_ok(c[2], s)
    return not cond_ok(c[1], s)


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 apply_effects(eff: dict, s: dict, goto: str) -> dict:
    aff, flags, items = dict(s["aff"]), dict(s["flags"]), dict(s["items"])
    for k in sorted((eff.get("set") or {})):
        flags[k] = eff["set"][k]
    for k in sorted((eff.get("add") or {})):
        aff[k] = clamp(aff.get(k, 0) + eff["add"][k])
    for k in sorted((eff.get("grant") or {})):
        items[k] = eff["grant"][k]
    for k in sorted((eff.get("revoke") or {})):
        items[k] = 0
    return {"node": goto, "aff": aff, "flags": flags, "items": items, "path": s["path"] + [goto]}


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 delta(prev: dict, nxt: dict) -> list:
    """两个快照之间真正改变的字段，逐条列出（有序）。"""
    out = []
    if prev["node"] != nxt["node"]:
        out.append(f"node:{prev['node']}->{nxt['node']}")
    for k in sorted(nxt["aff"]):
        if prev["aff"][k] != nxt["aff"][k]:
            out.append(f"aff:{k}:{prev['aff'][k]}->{nxt['aff'][k]}")
    for k in sorted(nxt["flags"]):
        if prev["flags"][k] != nxt["flags"][k]:
            out.append(f"flag:{k}:{prev['flags'][k]}->{nxt['flags'][k]}")
    for k in sorted(nxt["items"]):
        if prev["items"][k] != nxt["items"][k]:
            out.append(f"item:{k}:{prev['items'][k]}->{nxt['items'][k]}")
    return out


def walk(spec: dict) -> tuple:
    s, rows = init_state(spec), []
    for _ in range(64):
        cur = next(n for n in spec["dialogue"] if n["id"] == s["node"])
        if cur.get("ending"):
            break
        opts = [c for c in (cur.get("choices") or []) if cond_ok(c.get("when"), s)]
        if not opts:
            break
        nxt = apply_effects(opts[0].get("effects") or {}, s, opts[0]["goto"])
        rows.append((len(rows) + 1, opts[0]["id"], s["node"], nxt["node"], delta(s, nxt)))
        s = nxt
    return s, rows


def read_keys(spec: dict) -> tuple:
    """扫全 spec 的条件（对话 / 场景图 / 地点 / 结局），收集被读取的 flag / item。"""
    acc = {"flag": set(), "item": set()}

    def rec(c):
        if c is None:
            return
        k = c[0]
        if k in acc:
            acc[k].add(c[1])
        elif k in ("and", "or"):
            rec(c[1]); rec(c[2])
        elif k == "not":
            rec(c[1])

    conds = [n.get("line_when") for n in spec["dialogue"]]
    conds += [c.get("when") for n in spec["dialogue"] for c in (n.get("choices") or [])]
    conds += [e.get("when") for e in spec.get("scene_graph", {}).get("edges", [])]
    conds += [nd.get("enter_when") for nd in spec.get("scene_graph", {}).get("nodes", [])]
    conds += [loc.get("unlock_when") for loc in spec.get("world", {}).get("locations", [])]
    conds += [e.get("when") for e in spec.get("endings", [])]
    for c in conds:
        rec(c)
    return acc["flag"], acc["item"]


def main() -> int:
    with open(os.path.join(ROOT, "config/story_spec.json"), encoding="utf-8") as fh:
        spec = json.load(fh)
    s, rows = walk(spec)
    s2, _ = walk(spec)
    end = next(n for n in spec["dialogue"] if n["id"] == s["node"])
    print(f"spec={spec['meta']['title']}  策略=总是选第一个可用选项  步数 {len(rows)}  "
          f"终局 {s['node']}({end.get('ending', '?')})")
    print(f"终局规范串 {canonical(s)}")
    print(f"重跑一致（同一策略两次规范串相同）{canonical(s2) == canonical(s)}")
    print("逐步骤差量（列：step choice node_from node_to delta n_delta；此处打印前 4 步与末步）：")
    for st, cid, nf, nt, d in rows[:4]:
        print(f"  step {st:>2}  {cid:<12} {nf}->{nt}  {';'.join(d)}  (n={len(d)})")
    print(f"  …（共 {len(rows)} 步，完整表写入 out/ne_03_3_delta.csv）")
    st, cid, nf, nt, d = rows[-1]
    print(f"  step {st:>2}  {cid:<12} {nf}->{nt}  {';'.join(d)}  (n={len(d)})")

    rf, ri = read_keys(spec)
    wf = sorted({k for n in spec["dialogue"] for c in (n.get("choices") or [])
                 for k in ((c.get("effects") or {}).get("set") or {})})
    gi = sorted({k for n in spec["dialogue"] for c in (n.get("choices") or [])
                 for k in ((c.get("effects") or {}).get("grant") or {})})
    wrf = [k for k in wf if k not in rf]
    wri = [k for k in gi if k not in ri]
    print("写而不读（差量里恒为「只增不减的信息」）：")
    print(f"  flag {len(wrf)} 个：{wrf}（被 set 却无任何条件读取）")
    print(f"  item {len(wri)} 个：{wri}（被 grant 却无任何条件读取）")
    print("  判据：声明未用=报告，不判 FAIL")

    res = sum(1 for v in s["aff"].values() if not (AFF_LO <= v <= AFF_HI))
    print(f"I1 好感度越界 {res}")
    if res or canonical(s2) != canonical(s):
        return 1

    os.makedirs(OUT, exist_ok=True)
    with open(CSV_PATH, "w", encoding="utf-8", newline="") as fh:
        w = csv.writer(fh)
        w.writerow(["step", "choice", "node_from", "node_to", "delta", "n_delta"])
        for st, cid, nf, nt, d in rows:
            w.writerow([st, cid, nf, nt, ";".join(d), len(d)])
    print(f"wrote out/ne_03_3_delta.csv rows={len(rows)}")
    print("OK ne_03_3_snapshot：快照与差量一致，重跑逐位相同")
    return 0


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

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
重跑一致（同一策略两次规范串相同）True
逐步骤差量（列：step choice node_from node_to delta n_delta；此处打印前 4 步与末步）：
  step  1  c_pier_1a    d_pier_1->d_pier_2  node:d_pier_1->d_pier_2;flag:knows_yan_missing:0->1;flag:met_hai:0->1  (n=3)
  step  2  c_pier_2a    d_pier_2->d_tavern_1  node:d_pier_2->d_tavern_1;flag:knows_tide_rule:0->1  (n=2)
  step  3  c_tavern_1a  d_tavern_1->d_tavern_2  node:d_tavern_1->d_tavern_2;aff:hui:50->55;flag:met_hui:0->1  (n=3)
  step  4  c_tavern_2a  d_tavern_2->d_tavern_3  node:d_tavern_2->d_tavern_3;aff:hui:55->60  (n=2)
  …（共 25 步，完整表写入 out/ne_03_3_delta.csv）
  step 25  c_dq_2a      d_dq_2->e_lit  node:d_dq_2->e_lit  (n=1)
写而不读（差量里恒为「只增不减的信息」）：
  flag 7 个：['has_lantern', 'hui_trusts', 'met_hai', 'met_lin', 'met_mu', 'promised_hai', 'shen_trusts']（被 set 却无任何条件读取）
  item 1 个：['fog_glass']（被 grant 却无任何条件读取）
  判据：声明未用=报告，不判 FAIL
I1 好感度越界 0
wrote out/ne_03_3_delta.csv rows=25
OK ne_03_3_snapshot：快照与差量一致，重跑逐位相同

// file: core/test/ne_03_3_snapshot.test.ts
// tier: P
// verify: node --test 'core/test/ne_03_3_snapshot.test.ts'
// 3.3 状态快照与差量：快照 = canonical 串；差量 = 两快照间真正改变的字段。

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, initState } from "../src/inkwell/engine.ts";
import type { Spec } from "../src/inkwell/engine.ts";
import { delta, firstAvailablePath, snapshot, writtenNotRead } from "../src/inkwell/ne_03_3_snapshot.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;

test("确定性路径 25 步，终局 e_lit；快照即 canonical 串", () => {
  const steps = firstAvailablePath(spec);
  assert.equal(steps.length, 25);
  assert.equal(steps[steps.length - 1].next.node, "e_lit");
  assert.equal(snapshot(steps[0].prev), canonical(initState(spec)));
});

test("每一步差量都含节点改变；差量只列真正变化的字段", () => {
  const steps = firstAvailablePath(spec);
  for (const st of steps) {
    const d = delta(st.prev, st.next);
    assert.ok(d[0].startsWith("node:"), `step ${st.choice} 的差量应以 node 开头`);
    for (const k of Object.keys(st.next.aff)) {
      const changed = st.prev.aff[k] !== st.next.aff[k];
      const listed = d.some((x) => x.startsWith(`aff:${k}:`));
      assert.equal(listed, changed, `${k} 的差量列出与变化不一致`);
    }
  }
});

test("第 1 步 c_pier_1a 的差量：置位 knows_yan_missing 与 met_hai", () => {
  const st = firstAvailablePath(spec)[0];
  assert.equal(st.choice, "c_pier_1a");
  assert.deepEqual(delta(st.prev, st.next), [
    "node:d_pier_1->d_pier_2",
    "flag:knows_yan_missing:0->1",
    "flag:met_hai:0->1",
  ]);
});

test("写而不读：7 个 flag 与 1 个 item 被写却无任何条件读取", () => {
  const r = writtenNotRead(spec);
  assert.deepEqual(r.flags, ["has_lantern", "hui_trusts", "met_hai", "met_lin", "met_mu", "promised_hai", "shen_trusts"]);
  assert.deepEqual(r.items, ["fog_glass"]);
  console.log(`写而不读：flag ${r.flags.length} 个，item ${r.items.length} 个`);
});

test("重跑逐位相同：同一策略两次的终局规范串一致", () => {
  const a = firstAvailablePath(spec).at(-1)!.next;
  const b = firstAvailablePath(spec).at(-1)!.next;
  assert.equal(canonical(a), canonical(b));
});

写而不读：flag 7 个，item 1 个
✔ 确定性路径 25 步，终局 e_lit；快照即 canonical 串 (1.025209ms)
✔ 每一步差量都含节点改变；差量只列真正变化的字段 (0.425083ms)
✔ 第 1 步 c_pier_1a 的差量：置位 knows_yan_missing 与 met_hai (0.502291ms)
✔ 写而不读：7 个 flag 与 1 个 item 被写却无任何条件读取 (0.649708ms)
✔ 重跑逐位相同：同一策略两次的终局规范串一致 (0.204167ms)
ℹ tests 5
ℹ pass 5
ℹ fail 0