sots-re/verify/fpu-cw/trace_bitdiff.py

#!/usr/bin/env python3
"""Bit-exact comparison of two shim traces of the same turn, hook by hook.

Why not just diff the text: **the trace's own float rendering is not a stable comparison
surface when the x87 control word has been forced.** The emitter prints an f32 with `%.9g`,
and the CRT's digit generation is itself done in x87 arithmetic, so under `fpu_cw = 0x007f`
(24-bit) the *same* float32 renders as `1.5647336` instead of `1.56473362`. Comparing the text
of a 24-bit run against a 53-bit run reports dozens of "differences" that are all the printf,
not the simulation. Every float here is therefore re-quantised to its IEEE bit pattern before
comparison, and per-process noise (pointers, thread id, timestamps, call ids, the `fpu_cw`
field that is the independent variable) is scrubbed.

Usage: trace_bitdiff.py <a.jsonl[.gz]> <b.jsonl[.gz]> [hook ...]
"""
import gzip
import json
import re
import struct
import sys

PTR = re.compile(r"^0x[0-9a-f]{8}$")
# Only hooks that fire *inside* the turn pipeline: ServerPlayer::ComputeBudget also runs once
# per UI frame, so its call count depends on how long the session sat at a menu and is not
# comparable between runs. Name it explicitly on the command line if you want it anyway.
DEFAULT_HOOKS = (
    "Game::ServerSystem::ProcessTurn",
    "Game::StrategyServer::MoveFleet",
    "Game::TechTree::ProcessResearch",
    "Game::ServerPlayer::OnTechResearched",
)
NOISE = ("ts", "call_id", "thread")


def f32_bits(v):
    return struct.unpack("<I", struct.pack("<f", float(v)))[0]


def f64_bits(v):
    return struct.unpack("<Q", struct.pack("<d", float(v)))[0]


def norm(o):
    if isinstance(o, dict):
        t = o.get("t")
        if t == "ptr":
            return {"t": "ptr", "n": o.get("n")}
        if t == "f32":
            return {"t": "f32", "n": o.get("n"), "bits": f32_bits(o["v"])}
        if t == "f64":
            return {"t": "f64", "n": o.get("n"), "bits": f64_bits(o["v"])}
        return {k: norm(v) for k, v in o.items() if k not in NOISE}
    if isinstance(o, list):
        return [norm(x) for x in o]
    if isinstance(o, str) and PTR.match(o):
        return "<ptr>"
    return o


def _open(path):
    op = gzip.open if path.endswith(".gz") else open
    return op(path, "rt", encoding="utf-8", errors="replace")


def load(path, hook):
    out = []
    needle = '"hook":"%s"' % hook
    with _open(path) as f:
        for line in f:
            if needle not in line:
                continue
            rec = norm(json.loads(line))
            rec["args"] = [a for a in rec.get("args", []) if a.get("n") != "fpu_cw"]
            for side in (rec.get("side") or {}).values():
                if not isinstance(side, dict):
                    continue
                for snap in side.values():
                    if isinstance(snap, dict) and isinstance(snap.get("v"), dict):
                        snap["v"].pop("fpu_cw", None)
            out.append(rec)
    return out


def leafdiff(a, b, path=""):
    out = []
    if isinstance(a, dict) and isinstance(b, dict):
        for k in sorted(set(a) | set(b)):
            out += leafdiff(a.get(k), b.get(k), f"{path}.{k}")
    elif isinstance(a, list) and isinstance(b, list) and len(a) == len(b):
        for i, (x, y) in enumerate(zip(a, b)):
            out += leafdiff(x, y, f"{path}[{i}]")
    elif a != b:
        out.append((path, a, b))
    return out


def main():
    pa, pb = sys.argv[1], sys.argv[2]
    hooks = sys.argv[3:] or list(DEFAULT_HOOKS)
    print(f"A {pa}\nB {pb}")
    total = 0
    for hook in hooks:
        a, b = load(pa, hook), load(pb, hook)
        if not a and not b:
            continue
        if len(a) != len(b):
            print(f"{hook}: CALL COUNT DIFFERS {len(a)} vs {len(b)}")
            total += 1
            continue
        diffs = [(i, p, x, y) for i, (ra, rb) in enumerate(zip(a, b)) for p, x, y in leafdiff(ra, rb)]
        total += len(diffs)
        print(f"{hook}: {len(diffs)} leaf difference(s) over {len(a)} record(s)")
        for i, p, x, y in diffs:
            print(f"    rec[{i}] {p}: {x} -> {y}")
    print(f"TOTAL: {total}")
    return 1 if total else 0


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