sots-re/tools/rng_oracle_check.py

#!/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())