#!/usr/bin/env python3 # -*- coding: utf-8 -*- """parity_anchor.py — 「逐位一致对拍」锚工具（架构重构前后逐位一致验收；原需求卡已按「吸收即死」退役，凭据见 git 提交史） 载体 = adventure_monte_carlo.gd 同种子批量完整局（SceneTree headless，自持 ConfigDB，零渲染）。 三变体钉死（种子面错开，覆盖 base / 同调账 / 商店-GP 账三类 T2/T3 重构面）： base : -- 20 71000 both min off off off attune : -- 20 72000 both min on off off shop : -- 20 73000 both min off on off 锚双面： A. RUNSUM 逐行块（stdout 局级摘要面，字符串列表原样入库 anchor.json——红灯时可打印首条差异行）； B. playtest JSONL 全事件流（剔除带墙钟的 session/diagnosis 两类行；文件名本身带时间戳不入比对面）。 比对口径 = 字节级逐位一致（同引擎二进制+同种子确定性，冻结前已实测两跑全同）。 原始面不入库（体量 MB 级）：红灯取证用 dump 模式对重构前后各落一份原始数据再 diff—— 旧提交随时可重生成同字节原始面（确定性保证），锚只存 sha256 清单。 用法: python tools/parity_anchor.py freeze # 冻结锚（重构前基线，写入 game/tests/evidence/parity/anchor.json） python tools/parity_anchor.py check # 对拍：重跑同参批，与锚逐位比对（全同退出 0，差异退出 1 并打印首差异） python tools/parity_anchor.py dump --out DIR # 只跑批落原始面到 DIR（红灯取证） 可选 --variants base,attune,shop / --quick（只跑 base） """ import argparse import hashlib import json import re import subprocess import sys import time from pathlib import Path from godot_user_dir import resolve_user_dir ROOT = Path(__file__).resolve().parent.parent GODOT = r"D:\code\godot\Godot_v4.7.2-stable_win64_console.exe" GAME_DIR = ROOT / "game" ANCHOR_PATH = ROOT / "game" / "tests" / "evidence" / "parity" / "anchor.json" USER_LOGS = resolve_user_dir(ROOT) / "logs" # 变体表：名字 → (seed0, attune, shop)。（N=20/策略 both/route min/feat off 全变体钉死。） VARIANTS = { "base": (71000, "off", "off"), "attune": (72000, "on", "off"), "shop": (73000, "off", "on"), } N_RUNS = 20 BATCH_RE = re.compile(r"^BATCH (\w+) N=(\d+) 终局计数=(.*) 日志=user://logs/(playtest_\d+_\S+\.jsonl)\s*$") TS_ROW_KINDS = ("session", "diagnosis") # 带墙钟 ts 的两类行——剔出比对面 def run_variant(name: str, seed0: int, attune: str, shop: str, timeout: int = 560): """跑一个变体批，返回 (runsum_lines, {policy: jsonl绝对路径})。""" before = set(p.name for p in USER_LOGS.glob("playtest_*.jsonl")) if USER_LOGS.is_dir() else set() cmd = [GODOT, "--headless", "--path", str(GAME_DIR), "--script", "res://sim/adventure_monte_carlo.gd", "--", str(N_RUNS), str(seed0), "both", "min", attune, shop, "off"] proc = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout) out = (proc.stdout or "") + (proc.stderr or "") runsum = [ln for ln in out.splitlines() if ln.startswith("RUNSUM ")] logs = {} for ln in out.splitlines(): m = BATCH_RE.match(ln) if m: logs[m.group(1)] = m.group(4) after = set(p.name for p in USER_LOGS.glob("playtest_*.jsonl")) if USER_LOGS.is_dir() else set() fresh = after - before resolved = {} for policy, fname in logs.items(): p = USER_LOGS / fname if p.is_file(): resolved[policy] = p else: cand = [f for f in fresh if f.startswith("playtest_%d_" % seed0)] if len(cand) == 1: resolved[policy] = USER_LOGS / cand[0] else: print("[parity] FAIL：变体 %s 的 %s 日志文件定位失败（fresh=%s）" % (name, policy, cand)) sys.exit(2) expected_policies = 2 if len(runsum) != N_RUNS * expected_policies or len(resolved) != expected_policies: print("[parity] FAIL：变体 %s 产出不完整（RUNSUM %d 行 / 日志 %d 份）" % (name, len(runsum), len(resolved))) print(out[-3000:]) sys.exit(2) return runsum, resolved def filter_jsonl(path: Path) -> bytes: """剔除带 ts 行，保留原始字节行序（bit-exact；不做任何再序列化）。""" kept = [] with open(path, "rb") as f: for raw in f.read().split(b"\n"): if not raw.strip(): continue head = raw[:48] if any(('{"kind":"%s"' % k).encode() in head for k in TS_ROW_KINDS): continue kept.append(raw) return b"\n".join(kept) + b"\n" def collect(variants, quick=False): if quick: variants = [v for v in variants if v == "base"] or ["base"] data = {} for v in variants: seed0, attune, shop = VARIANTS[v] t0 = time.time() print("[parity] 跑批 %s（N=%d×2 策略，seed0=%d，attune=%s，shop=%s）…" % (v, N_RUNS, seed0, attune, shop)) runsum, logs = run_variant(v, seed0, attune, shop) faces = {"runsum": runsum, "jsonl": {}} for policy in sorted(logs): body = filter_jsonl(logs[policy]) faces["jsonl"][policy] = { "sha256": hashlib.sha256(body).hexdigest(), "rows": body.count(b"\n"), } data[v] = faces print("[parity] %s 完成（%.0fs，RUNSUM %d 行，JSONL %s）" % (v, time.time() - t0, len(runsum), ", ".join("%s=%s" % (p, faces["jsonl"][p]["sha256"][:12]) for p in sorted(faces["jsonl"])))) return data def dump_variant_raw(variants, outdir: Path): outdir.mkdir(parents=True, exist_ok=True) for v in variants: seed0, attune, shop = VARIANTS[v] runsum, logs = run_variant(v, seed0, attune, shop) (outdir / ("%s_runsum.txt" % v)).write_text("\n".join(runsum) + "\n", encoding="utf-8", newline="\n") for policy, p in logs.items(): (outdir / ("%s_%s.jsonl" % (v, policy))).write_bytes(filter_jsonl(p)) print("[parity] dump %s → %s" % (v, outdir)) def main(): if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") ap = argparse.ArgumentParser(description="迭代27重构逐位一致对拍锚") ap.add_argument("mode", choices=["freeze", "check", "dump"]) ap.add_argument("--variants", default="base,attune,shop") ap.add_argument("--quick", action="store_true", help="只跑 base 变体") ap.add_argument("--out", default="", help="dump 模式输出目录") args = ap.parse_args() variants = [v.strip() for v in args.variants.split(",") if v.strip() in VARIANTS] if not variants: print("[parity] FAIL：无有效变体（可选 %s）" % list(VARIANTS)) return 2 if args.mode == "dump": dump_variant_raw(variants, Path(args.out if args.out else str(ROOT / "tools" / "parity_dump"))) return 0 data = collect(variants, quick=args.quick) if args.mode == "freeze": ANCHOR_PATH.parent.mkdir(parents=True, exist_ok=True) anchor = {"variants": {v: data[v] for v in variants}} ANCHOR_PATH.write_text(json.dumps(anchor, ensure_ascii=False, indent=1), encoding="utf-8", newline="\n") print("[parity] 锚已冻结 → %s（变体 %s）" % (ANCHOR_PATH, variants)) return 0 # check if not ANCHOR_PATH.is_file(): print("[parity] FAIL：锚不存在（先 freeze）→ %s" % ANCHOR_PATH) return 2 anchor = json.loads(ANCHOR_PATH.read_text(encoding="utf-8")) bad = 0 for v in variants: if v not in anchor["variants"]: print("[parity] FAIL：锚无变体 %s（锚含 %s）" % (v, list(anchor["variants"]))) bad += 1 continue old = anchor["variants"][v] new = data[v] if old["runsum"] != new["runsum"]: bad += 1 print("[parity] ✗ %s RUNSUM 漂移（%d 行 vs %d 行）" % (v, len(old["runsum"]), len(new["runsum"]))) for i, (a, b) in enumerate(zip(old["runsum"], new["runsum"])): if a != b: print("[parity] 首差异第 %d 行：" % i) print("[parity] 锚: %s" % a[:600]) print("[parity] 今: %s" % b[:600]) break for policy in sorted(old["jsonl"]): o, n = old["jsonl"][policy], new["jsonl"].get(policy) if n is None: bad += 1 print("[parity] ✗ %s/%s 本次缺日志" % (v, policy)) elif o["sha256"] != n["sha256"]: bad += 1 print("[parity] ✗ %s/%s JSONL 逐位漂移（锚 %s… 行数 %d vs 今 %s… 行数 %d）" % (v, policy, o["sha256"][:12], o["rows"], n["sha256"][:12], n["rows"])) if bad: print("[parity] 逐位一致对拍 ✗ 失败 %d 项——该波回炉（V1 红线）" % bad) return 1 print("[parity] 逐位一致对拍 ✓ 全同（变体 %s，RUNSUM+JSONL 双面逐位一致）" % variants) return 0 if __name__ == "__main__": sys.exit(main())