#!/usr/bin/env python3 """Does the standalone's generator land where the oracle's did? Lane Z left two calibrated pairs -- a pre-turn save, a post-turn save, and a word count verified from the file bytes alone (`twists = 0`, so the number does not route through anyone's twist implementation). This tool runs the standalone on the pre-turn save with `--commit-rng` and compares three things against the post-turn save: * the 624-word state block, byte for byte; * `left`, the count of words still unread in the block; * the resulting absolute position, which is what a word count actually is. The verdict is binary and the residual is stated in words, never as a percentage. A residual of N means the standalone models N fewer words than the turn really spends, and lane Z's per-call-site ledger says which sites those are. tools/rng_oracle_check.py # both calibrated pairs tools/rng_oracle_check.py --binary PATH # a sots_turn built elsewhere tools/rng_oracle_check.py --pair IN OUT WORDS # an ad-hoc pair with a known cost """ import argparse import json import os import shutil import subprocess import sys import tempfile RE_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")) PAIR_DIR = os.path.join(RE_ROOT, "verify", "results", "shim", "tailrng") sys.path.insert(0, os.path.join(RE_ROOT, "verify", "save-reader")) import save_reader as sr # noqa: E402 N, M = 624, 397 # (pre-turn, post-turn, words the game spent, what it is) PAIRS = [ ("z2-endturn.sav", "z2-autosave.sav", 20, "ref-turn2, turn 4 -> 5"), ("z-t6-endturn.sav", "z-t6-autosave.sav", 18, "ref-turn2, turn 5 -> 6"), ] DEFAULT_BINARIES = [ os.path.expanduser("~/sots-engine-wt-yield/build-host/src/app/sots_turn"), os.path.expanduser("~/sots-engine/build-host/src/app/sots_turn"), "/srv/re-lab/build/sots-engine-y/src/app/sots_turn", ] def twist(mt): mt = list(mt) for kk in range(N - M): y = (mt[kk] & 0x80000000) | (mt[kk + 1] & 0x7FFFFFFF) mt[kk] = mt[kk + M] ^ (y >> 1) ^ (0x9908B0DF if y & 1 else 0) for kk in range(N - M, N - 1): y = (mt[kk] & 0x80000000) | (mt[kk + 1] & 0x7FFFFFFF) mt[kk] = mt[kk + (M - N)] ^ (y >> 1) ^ (0x9908B0DF if y & 1 else 0) y = (mt[N - 1] & 0x80000000) | (mt[0] & 0x7FFFFFFF) mt[N - 1] = mt[M - 1] ^ (y >> 1) ^ (0x9908B0DF if y & 1 else 0) return mt def _find_rng(node, out): if getattr(node, "name", None) == "RNG": out.append(node) for c in getattr(node, "children", []) or []: _find_rng(c, out) def rng_state(path): """(mt[624], left) out of a save's generator frame.""" res = sr.read_save(path) hits = [] _find_rng(res.tree, hits) if not hits: raise SystemExit("no generator frame in " + path) n = hits[0] raw = n.raw or b"".join(c.raw for c in (getattr(n, "children", []) or []) if c.raw) if not raw: raise SystemExit("generator frame in %s carries no bytes" % path) for off in range(0, len(raw) - 2500 + 1): mt = [int.from_bytes(raw[off + 4 * i:off + 4 * i + 4], "little") for i in range(N)] left = int.from_bytes(raw[off + 2496:off + 2500], "little", signed=True) if 0 <= left <= N: return mt, left raise SystemExit("cannot parse the generator blob of %s (%d B)" % (path, len(raw))) def words_between(a, b, max_twists=64): """Words consumed moving from state a to state b, exact across block boundaries.""" (mta, la), (mtb, lb) = a, b cur, tw = mta, 0 while tw <= max_twists and cur != mtb: cur = twist(cur) tw += 1 if cur != mtb: return None, None return 624 * tw + (la - lb), tw def find_binary(explicit): if explicit: return explicit if os.path.exists(explicit) else None for p in DEFAULT_BINARIES: if os.path.exists(p): return p return shutil.which("sots_turn") def run_pair(binary, src, oracle, expected, note, workdir): out_sav = os.path.join(workdir, "post-" + os.path.basename(src)) metric = os.path.join(workdir, "metric.json") cmd = [binary, src, "--out", out_sav, "--metric", metric, "--commit-rng"] proc = subprocess.run(cmd, capture_output=True, text=True) row = {"input": os.path.basename(src), "oracle": os.path.basename(oracle), "note": note, "oracleWords": expected, "exit": proc.returncode} if proc.returncode != 0: row["error"] = proc.stderr.strip()[:2000] return row before = rng_state(src) after_oracle = rng_state(oracle) after_ours = rng_state(out_sav) oracle_words, oracle_twists = words_between(before, after_oracle) our_words, our_twists = words_between(before, after_ours) row.update({ "oracleWordsFromFiles": oracle_words, "oracleTwists": oracle_twists, "ourWords": our_words, "ourTwists": our_twists, "blockIdentical": after_ours[0] == after_oracle[0], "leftOurs": after_ours[1], "leftOracle": after_oracle[1], "stateMatches": after_ours[0] == after_oracle[0] and after_ours[1] == after_oracle[1], }) row["residual"] = None if our_words is None or oracle_words is None else oracle_words - our_words if os.path.exists(metric): with open(metric) as f: m = json.load(f) row["standaloneRngWords"] = m.get("run", {}).get("rngWords") row["standaloneRngCommitted"] = m.get("run", {}).get("rngCommitted") row["unaccounted"] = m.get("run", {}).get("rngUnaccounted", []) return row def render(rows): out = [] out.append("# does the standalone's generator land on the oracle's?\n") for r in rows: out.append("## %s -> %s (%s)" % (r["input"], r["oracle"], r["note"])) if r.get("error"): out.append(" standalone failed: " + r["error"].splitlines()[0]) out.append("") continue out.append(" oracle spent %s word(s) (%s twist(s), from the files)" % (r["oracleWordsFromFiles"], r["oracleTwists"])) out.append(" standalone %s word(s) (metric says %s)" % (r["ourWords"], r.get("standaloneRngWords"))) out.append(" residual %s word(s) NOT modelled" % r["residual"]) out.append(" state block identical: %s left: ours %s, oracle %s" % ("yes" if r["blockIdentical"] else "no", r["leftOurs"], r["leftOracle"])) out.append(" GENERATOR STATE MATCHES ORACLE: %s" % ("YES" if r["stateMatches"] else "no")) for u in r.get("unaccounted", []): out.append(" - unaccounted: " + u) out.append("") matched = sum(1 for r in rows if r.get("stateMatches")) out.append("verdict: %d of %d calibrated pair(s) match the oracle's generator state exactly" % (matched, len(rows))) return "\n".join(out) + "\n" def main(argv=None): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--binary") ap.add_argument("--pair", nargs=3, metavar=("INPUT", "ORACLE", "WORDS")) ap.add_argument("--json", help="write the machine-readable result here") args = ap.parse_args(argv) binary = find_binary(args.binary) if not binary: print("rng_oracle_check: sots_turn not found; build the host preset first", file=sys.stderr) return 2 if args.pair: pairs = [(args.pair[0], args.pair[1], int(args.pair[2]), "ad-hoc")] else: pairs = [(os.path.join(PAIR_DIR, a), os.path.join(PAIR_DIR, b), w, n) for a, b, w, n in PAIRS] rows = [] with tempfile.TemporaryDirectory(prefix="rngoracle-") as wd: for src, oracle, words, note in pairs: if not (os.path.exists(src) and os.path.exists(oracle)): print("missing calibrated pair: %s / %s" % (src, oracle), file=sys.stderr) return 2 rows.append(run_pair(binary, src, oracle, words, note, wd)) text = render(rows) print(text, end="") if args.json: with open(args.json, "w") as f: json.dump({"binary": binary, "pairs": rows}, f, indent=2) # Exit 0 whether or not it matches: this is a measurement, not a gate. A non-zero exit # is reserved for the tool being unable to measure. return 0 if __name__ == "__main__": sys.exit(main())