DKChan
fc7fee03bd
- tools/parity_anchor.py: 三变体(base/attune/shop)×40局 RUNSUM+JSONL 双面哈希对拍（V1 核心验收载体） - anchor.json: 重构前锚（种子面 71000/72000/73000 错开，覆盖 base/同调账/商店-GP 账三类重构面） - 探针迁移对照表: 11 挂载面逐一列成员依赖与迁移方向（卡片 T0.3）
203 lines
9.2 KiB
Python
203 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""parity_anchor.py — 迭代27重构「逐位一致对拍」锚工具（需求卡片-代码架构技术债修复批 §7 V1 核心验收）
|
||
|
||
载体 = 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 os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
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 = Path(os.environ.get("APPDATA", "")) / "Godot" / "app_userdata" / "DungeonSweeperVerticalSlice" / "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://smoke/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())
|