gplan-2/tools/parity_anchor.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""parity_anchor.py — 迭代27重构「逐位一致对拍」锚工具（代码架构技术债修复批〔迭代 27〕§7 V1 核心验收；该卡已按 doc-conventions §11「吸收即死」退役，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 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())