sots-re/verify/state-checksum/test_state_checksum.py
alex fdd0b72b7a state-checksum: whole-state diagnostic checksum harness (lane C)
The complement to the per-function compare harness. Instead of "did this
function's declared outputs match", it asks "is the entire simulation state
still identical" -- so no region-declaration mistake can hide from it.

Coverage is PROVED, not declared: the digest tree is re-serialised and compared
byte-for-byte against the inflated save on every run. When that reconstruction
reproduces the stream, the whole file is a function of the digest's inputs. A
run that cannot account for the file says so and exits non-zero. This is the
direct answer to B4's three hooks that printed "0 diverged" over an empty
region set.

It localises. The root is the fold of a per-subsystem / per-object tree with
named objects, so the known load->re-save delta reports as exactly five leaves
-- /Summary/Checksum and four /Sim/players/Player[...]/Status 4->0 -- naming the
two Singularity players by id where the raw byte diff could only say "1st of
two". One real End Turn reports as 108 fully attributed differences.

Float-parity policy is explicit and strict by default (STATE_CHECKSUM.md 3):
raw IEEE-754 bits; a `canonical` policy for signed zero and NaN payloads only;
and deliberately NO tolerant hashing mode, because quantisation moves the cliff
rather than removing it and destroys the roll-up. Tolerance lives in the differ
as --ulps, applied after localisation. Corpus census: 0 NaN, 0 -0.0, 0
subnormals across 4,474 float leaves, so the strict default costs nothing today
and a test fails the day that changes.

Validated on the real saves (verify/results/state-checksum/): 10 files, 4
distinct contents, all STABLE + COVERED; chain record/verify works on the real
turn1-3 saves. The VM-driven replay loop is designed (section 5) but UNRUN.

Section 3.5 names the one question the host side cannot settle -- whether the
turn pipeline depends on x87 intermediate precision -- and the experiment that
would: force fpu_cw to 0x027f / 0x127f / 0x137f across End Turn and checksum
the three autosaves.

Also recorded: Summary.Checksum is NOT a byte sum over the inflated stream nor
a sum over the int leaves (both ruled out), so nobody repeats those two.

38 tests; sots-engine untouched, clean_room_check.sh OK.
2026-09-08 03:46:34 -04:00

461 lines
19 KiB
Python

#!/usr/bin/env python3
"""Tests for state_checksum.py.
Synthetic tests always run. Real-save tests are skipped cleanly when no saves
are available: they look in $SOTS_SAVES_DIR first, then at the in-repo
verify/results/saves/. No save is ever written into either repo by this file.
uv run python3 -m unittest discover -s verify/state-checksum -v
"""
from __future__ import annotations
import gzip
import json
import math
import os
import struct
import sys
import tempfile
import unittest
_HERE = os.path.dirname(os.path.abspath(__file__))
_VERIFY = os.path.dirname(_HERE)
for _p in (_HERE, os.path.join(_VERIFY, "save-reader")):
if _p not in sys.path:
sys.path.insert(0, _p)
import save_reader as sr # noqa: E402
import save_writer_stub as sw # noqa: E402
import state_checksum as sc # noqa: E402
# --- helpers ------------------------------------------------------------------
def tiny(status_a=4, status_b=4, checksum=-1000, extra_ids=(),
fval_a=1.5, fval_b=2.5) -> bytes:
"""A miniature save with the shapes the tool special-cases.
The `Sim` prefix fields (`KeyPath`..`NMnx`) are emitted because the reader
types `PlayerIDs` positionally off the Sim shape; without them the tag is
guessed and a 4-byte int is indistinguishable from a 2-byte string at the
same item size (SAVE_FORMAT.md section 2).
"""
w = sw.SaveWriter("joint")
w.begin("Summary")
w.string("GameName", "T")
w.int("Turn", 2)
w.int("Checksum", checksum)
w.end()
w.begin("Sim")
w.string("KeyPath", "")
w.int("NMSz", 16)
w.int("NMLc", 0)
w.int("NMnx", 109)
ids = (16, 32) + tuple(extra_ids)
w.int("PlayerIDs", len(ids))
for i in ids:
w.int(".", i)
w.int("NumPlrs", 2)
for pid, st, fv in ((16, status_a, fval_a), (32, status_b, fval_b)):
w.int("PlayerID", pid)
w.begin("Player")
w.string("PlryName", f"p{pid}")
w.int("Status", st)
w.float("IdealSuit", fv)
w.end()
w.end()
return w.bytes()
def find_saves() -> list[str]:
d = os.environ.get("SOTS_SAVES_DIR") or os.path.join(_VERIFY, "results", "saves")
if not os.path.isdir(d):
return []
return sorted(os.path.join(d, f) for f in os.listdir(d) if f.endswith(".sav"))
REAL = find_saves()
needs_saves = unittest.skipUnless(REAL, "no saves in $SOTS_SAVES_DIR or verify/results/saves")
_CACHE: dict = {}
def real_ck(path: str, **kw):
"""Memoised checksum_save -- a real save costs ~6 s to parse."""
key = (path, tuple(sorted(kw.items())))
if key not in _CACHE:
_CACHE[key] = sc.checksum_save(path, **kw)
return _CACHE[key]
# --- coverage: the property that makes the digest evidence --------------------
class CoverageTest(unittest.TestCase):
def test_reconstruction_is_exact_on_a_synthetic_save(self):
data = tiny()
ck = sc.checksum_bytes(data)
self.assertTrue(ck.coverage["ok"])
self.assertEqual(ck.coverage["rebuiltBytes"], len(data))
def test_reconstruction_is_exact_on_the_schema_fixture(self):
data, _ = sw.build_fixture()
ck = sc.checksum_bytes(data)
self.assertTrue(ck.coverage["ok"], f"first diff at {ck.coverage['firstDiff']}")
def test_reconstruction_handles_gzip_and_bare_streams_alike(self):
data = tiny()
a = sc.checksum_bytes(data)
b = sc.checksum_bytes(gzip.compress(data, mtime=0))
self.assertEqual(a.digest, b.digest)
self.assertTrue(b.coverage["ok"])
def test_a_broken_reconstruction_is_reported_not_swallowed(self):
res = sr.read_bytes(tiny())
# corrupt the parse, not the file: a leaf whose raw no longer matches
for n in res.tree.children[0].children:
if n.name == "Turn":
n.raw = b"\xff\xff\xff\xff"
cov = sc.audit_coverage(res.tree, res.inflated)
self.assertFalse(cov["ok"])
self.assertIsNotNone(cov["firstDiff"])
# --- determinism and sensitivity ---------------------------------------------
class DigestTest(unittest.TestCase):
def test_same_bytes_same_digest(self):
data = tiny()
self.assertEqual(sc.checksum_bytes(data).digest, sc.checksum_bytes(data).digest)
def test_every_value_leaf_is_load_bearing(self):
"""Flip each leaf in turn; the root must move every single time.
This is the anti-'empty region set' test: a leaf the digest does not
actually consume would pass silently under any other check.
"""
base = sc.checksum_bytes(tiny())
seen = 0
for name, kw in (("status", dict(status_a=5)), ("checksum", dict(checksum=-1001)),
("float", dict(fval_a=1.5000001))):
with self.subTest(name):
other = sc.checksum_bytes(tiny(**kw))
self.assertNotEqual(base.digest, other.digest)
seen += 1
self.assertEqual(seen, 3)
def test_a_one_bit_float_change_moves_the_root(self):
a = sc.checksum_bytes(tiny(fval_a=1.5))
(bits,) = struct.unpack("<I", struct.pack("<f", 1.5))
(nudged,) = struct.unpack("<f", struct.pack("<I", bits + 1))
b = sc.checksum_bytes(tiny(fval_a=nudged))
self.assertNotEqual(a.digest, b.digest)
d = sc.diff(a.root, b.root)
self.assertEqual(len(d), 1, [repr(e) for e in d])
self.assertTrue(d[0].path.endswith("/IdealSuit"))
def test_a_float_leaf_diff_carries_its_ulp_distance(self):
"""Built from reader nodes directly: the fixture's positional type
inference would otherwise decide whether the leaf is a float at all."""
def leaf(v):
raw = struct.pack("<f", v)
n = sr.Node("F", "float", v, 0, 12, raw=raw)
return sc._build(n, "/F", "F", "bits", (), {})
(bits,) = struct.unpack("<I", struct.pack("<f", 1.5))
(nudged,) = struct.unpack("<f", struct.pack("<I", bits + 3))
d = sc.diff(leaf(1.5), leaf(nudged))
self.assertEqual(len(d), 1)
self.assertEqual(d[0].ulps, 3.0)
def test_the_digest_depends_on_the_inferred_kind_not_only_the_bytes(self):
"""Documented caveat: same 4 bytes, different reader typing, different
digest. Hence READER_FINGERPRINT is recorded with every chain."""
raw = struct.pack("<f", 1.5)
as_f = sc._build(sr.Node("F", "float", 1.5, 0, 12, raw=raw), "/F", "F", "bits", (), {})
as_i = sc._build(sr.Node("F", "int", 1069547520, 0, 12, raw=raw), "/F", "F", "bits", (), {})
self.assertNotEqual(as_f.digest, as_i.digest)
self.assertNotEqual(sc.READER_FINGERPRINT, "unknown")
def test_policy_is_domain_separated_into_the_root(self):
"""A strict root and a lenient root must never be confusable."""
data = tiny()
self.assertNotEqual(sc.checksum_bytes(data, floats="bits").digest,
sc.checksum_bytes(data, floats="canonical").digest)
# --- localisation -------------------------------------------------------------
class LocalisationTest(unittest.TestCase):
def test_diff_names_the_object_not_just_the_hash(self):
a = sc.checksum_bytes(tiny(status_a=4, status_b=4))
b = sc.checksum_bytes(tiny(status_a=0, status_b=0))
d = sc.diff(a.root, b.root)
paths = sorted(e.path for e in d)
self.assertEqual(paths, ['/Sim/players/Player[16 "p16"]/Status',
'/Sim/players/Player[32 "p32"]/Status'])
self.assertEqual([(e.a, e.b) for e in d], [(4, 0), (4, 0)])
def test_identical_trees_report_nothing(self):
a = sc.checksum_bytes(tiny())
b = sc.checksum_bytes(tiny())
self.assertEqual(sc.diff(a.root, b.root), [])
def test_an_inline_id_list_reports_as_one_list_not_a_shift_cascade(self):
a = sc.checksum_bytes(tiny())
b = sc.checksum_bytes(tiny(extra_ids=(48,)))
d = sc.diff(a.root, b.root)
lists = [e for e in d if e.kind == "list"]
self.assertEqual(len(lists), 1, [repr(e) for e in d])
self.assertEqual(lists[0].path, "/Sim/PlayerIDs[]")
self.assertEqual(lists[0].b, [48])
self.assertEqual(lists[0].a, [])
def test_sibling_indices_are_per_tag_so_insertions_do_not_renumber(self):
a = sc.checksum_bytes(tiny())
b = sc.checksum_bytes(tiny(extra_ids=(48,)))
# the players group keeps its labels even though ids were inserted before it
pa = {c.label for c in a.root.find("/Sim").find("/Sim/players").children}
pb = {c.label for c in b.root.find("/Sim").find("/Sim/players").children}
self.assertEqual(pa, pb)
def test_object_identity_folds_the_id_in(self):
"""Two objects with the same body but different ids must differ."""
w = sc.checksum_bytes(tiny())
p16 = w.root.find('/Sim/players/Player[16 "p16"]')
self.assertIsNotNone(p16)
self.assertEqual(p16.label, 'Player[16 "p16"]')
# --- float policy -------------------------------------------------------------
class FloatPolicyTest(unittest.TestCase):
NEG0 = struct.pack("<I", 0x80000000)
POS0 = struct.pack("<I", 0x00000000)
SNAN = struct.pack("<I", 0x7F800001)
QNAN = struct.pack("<I", 0x7FC00000)
FLT_MAX = struct.pack("<I", 0x7F7FFFFF)
def test_bits_policy_separates_signed_zero_and_nan_payloads(self):
self.assertNotEqual(sc.float_key(self.NEG0, "bits"), sc.float_key(self.POS0, "bits"))
self.assertNotEqual(sc.float_key(self.SNAN, "bits"), sc.float_key(self.QNAN, "bits"))
def test_canonical_policy_merges_them(self):
self.assertEqual(sc.float_key(self.NEG0, "canonical"), sc.float_key(self.POS0, "canonical"))
self.assertEqual(sc.float_key(self.SNAN, "canonical"), sc.float_key(self.QNAN, "canonical"))
def test_canonical_leaves_every_ordinary_value_alone(self):
for v in (0.0, 1.0, -1.0, 1e-30, 3.4028234663852886e38, 11.106206893920898):
raw = struct.pack("<f", v)
self.assertEqual(sc.float_key(raw, "canonical"), raw, v)
def test_flt_max_is_not_infinity(self):
"""events.md: the default EvPos is FLT_MAX (0x7f7fffff), not inf."""
(v,) = struct.unpack("<f", self.FLT_MAX)
self.assertFalse(math.isinf(v))
self.assertEqual(sc.float_key(self.FLT_MAX, "canonical"), self.FLT_MAX)
def test_ulps_apart(self):
one = struct.pack("<f", 1.0)
(b,) = struct.unpack("<I", one)
self.assertEqual(sc.ulps_apart(one, one), 0.0)
self.assertEqual(sc.ulps_apart(one, struct.pack("<I", b + 1)), 1.0)
self.assertEqual(sc.ulps_apart(one, struct.pack("<I", b + 5)), 5.0)
# crossing zero is continuous under the ordinal map
self.assertEqual(sc.ulps_apart(self.POS0, self.NEG0), 0.0)
self.assertEqual(sc.ulps_apart(one, struct.pack("<f", float("inf"))), math.inf)
self.assertEqual(sc.ulps_apart(self.QNAN, self.SNAN), 0.0)
def test_tolerance_is_not_available_as_a_hashing_policy(self):
"""Deliberate: a tolerant hash is a contradiction (see STATE_CHECKSUM.md 3.4)."""
self.assertEqual(set(sc.FLOAT_POLICIES), {"bits", "canonical"})
with self.assertRaises(ValueError):
sc.float_key(struct.pack("<f", 1.0), "tol:2")
# --- masking ------------------------------------------------------------------
class MaskTest(unittest.TestCase):
def test_resave_mask_absorbs_exactly_the_documented_delta(self):
a = sc.checksum_bytes(tiny(status_a=4, status_b=4, checksum=-1000), mask="resave")
b = sc.checksum_bytes(tiny(status_a=0, status_b=0, checksum=-1016), mask="resave")
self.assertEqual(a.digest, b.digest)
def test_mask_reports_what_it_hit(self):
ck = sc.checksum_bytes(tiny(), mask="resave")
self.assertEqual(ck.mask_hits, {"Status": 2, "Checksum": 1})
def test_mask_is_path_scoped(self):
"""A same-named tag outside the rule's path must NOT be masked."""
w = sw.SaveWriter("joint")
w.begin("Sim")
w.int("Status", 4) # not under /Sim/players/Player[...]
w.end()
ck = sc.checksum_bytes(w.bytes(), mask="resave")
self.assertEqual(ck.mask_hits, {})
other = sc.checksum_bytes(w.bytes().replace(struct.pack("<i", 4),
struct.pack("<i", 0)), mask="resave")
self.assertNotEqual(ck.digest, other.digest)
def test_no_mask_by_default(self):
a = sc.checksum_bytes(tiny(status_a=4))
b = sc.checksum_bytes(tiny(status_a=0))
self.assertNotEqual(a.digest, b.digest)
self.assertEqual(a.mask_hits, {})
# --- chain --------------------------------------------------------------------
class ChainTest(unittest.TestCase):
def _write(self, data: bytes) -> str:
f = tempfile.NamedTemporaryFile(suffix=".sav", delete=False)
f.write(gzip.compress(data, mtime=0))
f.close()
self.addCleanup(os.unlink, f.name)
return f.name
def test_record_then_verify_matches(self):
files = [self._write(tiny(checksum=-1000)), self._write(tiny(checksum=-2000))]
chain = sc.build_chain(files)
self.assertTrue(all(t["coverage"] for t in chain["turns"]))
ok, msgs = sc.verify_chain(chain, files)
self.assertTrue(ok, msgs)
def test_verify_stops_at_the_first_divergent_turn(self):
files = [self._write(tiny(checksum=-1000)), self._write(tiny(checksum=-2000))]
chain = sc.build_chain(files)
bad = [files[0], self._write(tiny(checksum=-2001))]
ok, msgs = sc.verify_chain(chain, bad)
self.assertFalse(ok)
self.assertIn("MATCH", msgs[0])
self.assertIn("DIVERGE", msgs[1])
self.assertTrue(any("/Summary" in m for m in msgs))
def test_chain_records_the_policy_it_was_built_under(self):
files = [self._write(tiny())]
chain = sc.build_chain(files, floats="canonical", mask="resave")
self.assertEqual(chain["policy"]["floats"], "canonical")
self.assertEqual(chain["policy"]["mask"], "resave")
self.assertEqual(chain["policy"]["readerFingerprint"], sc.READER_FINGERPRINT)
json.dumps(chain) # must stay serialisable
def test_verifying_under_a_different_policy_is_refused(self):
files = [self._write(tiny())]
chain = sc.build_chain(files, floats="bits")
ok, msgs = sc.verify_chain(chain, files, floats="canonical")
self.assertFalse(ok)
self.assertTrue(any("not comparable" in m for m in msgs), msgs)
def test_a_stale_reader_fingerprint_is_called_out(self):
files = [self._write(tiny())]
chain = sc.build_chain(files)
chain["policy"]["readerFingerprint"] = "0" * 16
_, msgs = sc.verify_chain(chain, files)
self.assertTrue(any("save_reader" in m for m in msgs), msgs)
# --- real saves ---------------------------------------------------------------
@needs_saves
class RealSaveTest(unittest.TestCase):
def test_coverage_is_proved_on_every_available_save(self):
for p in REAL:
with self.subTest(os.path.basename(p)):
ck = real_ck(p)
self.assertTrue(ck.coverage["ok"],
f"reconstruction diverged at {ck.coverage['firstDiff']}")
self.assertEqual(ck.coverage["rebuiltBytes"], ck.coverage["inflatedBytes"])
def test_identical_bytes_give_identical_digests(self):
by_hash: dict = {}
for p in REAL:
with open(p, "rb") as f:
data = f.read()
by_hash.setdefault(data, []).append(p)
for data, paths in by_hash.items():
if len(paths) < 2:
continue
digs = {real_ck(p).digest for p in paths}
self.assertEqual(len(digs), 1, paths)
def test_repeated_runs_are_stable(self):
p = REAL[0]
self.assertEqual(real_ck(p).digest, real_ck(p).digest)
def test_reader_is_clean_on_every_save(self):
for p in REAL:
with self.subTest(os.path.basename(p)):
ck = real_ck(p, audit=False)
self.assertEqual(ck.res.count("error"), 0)
self.assertEqual(ck.res.count("warn"), 0)
def test_canonical_float_policy_is_a_noop_on_this_corpus(self):
"""No save holds a NaN, a -0.0 or an infinity, so 'canonical' changes
nothing below the root. Recorded so a future save that breaks it fails
here rather than silently."""
for p in REAL:
with self.subTest(os.path.basename(p)):
a = real_ck(p, audit=False, floats="bits")
b = real_ck(p, audit=False, floats="canonical")
self.assertEqual(sc.diff(a.root, b.root), [])
def test_known_resave_delta_localises_to_five_named_leaves(self):
"""determinism-oracle.md: loading a post-turn autosave and re-saving
changes exactly Player.Status (4 -> 0) on the four turn-participating
players, plus the derived Summary.Checksum. The tree must name them."""
pairs = []
for a in REAL:
for b in REAL:
if a >= b:
continue
ca, cb = real_ck(a, audit=False), real_ck(b, audit=False)
d = sc.diff(ca.root, cb.root)
if d and all(e.kind == "value" for e in d) and \
{e.path.rsplit("/", 1)[-1] for e in d} == {"Status", "Checksum"}:
pairs.append((a, b, d))
if not pairs:
self.skipTest("no re-save pair among the available saves")
for a, b, d in pairs:
self.assertEqual(len(d), 5, [repr(e) for e in d])
statuses = [e for e in d if e.path.endswith("/Status")]
self.assertEqual(len(statuses), 4)
for e in statuses:
self.assertIn("/Sim/players/Player[", e.path)
self.assertEqual((e.a, e.b), (4, 0))
chk = [e for e in d if e.path == "/Summary/Checksum"]
self.assertEqual(len(chk), 1)
self.assertEqual(chk[0].b - chk[0].a, -16) # 4 x (4 -> 0)
def test_resave_mask_makes_that_pair_identical(self):
found = False
for a in REAL:
for b in REAL:
if a >= b:
continue
if real_ck(a, audit=False, mask="resave").digest == \
real_ck(b, audit=False, mask="resave").digest and \
real_ck(a, audit=False).digest != \
real_ck(b, audit=False).digest:
found = True
if not found:
self.skipTest("no re-save pair among the available saves")
self.assertTrue(found)
def test_a_real_turn_transition_localises_to_named_objects(self):
by_turn = {}
for p in REAL:
ck = real_ck(p, audit=False)
t = (ck.res.typed.get("summary") or {}).get("Turn")
by_turn.setdefault(t, ck)
if not {2, 3} <= set(by_turn):
self.skipTest("need a turn-2 and a turn-3 save")
d = sc.diff(by_turn[2].root, by_turn[3].root, limit=500)
self.assertGreater(len(d), 10)
self.assertTrue(any(e.path == "/Summary/Turn" for e in d))
# every difference must be attributed to a path, never to a bare hash
for e in d:
self.assertTrue(e.path.startswith("/"), repr(e))
named = [e for e in d if "/Sim/players/Player[" in e.path]
self.assertTrue(named, "a turn transition must move at least one player")
if __name__ == "__main__":
unittest.main(verbosity=2)