"""Grok: v0.3 adaptations of TRON test-review additions (2026-10-13), test-only. 1) SRT NaN-mask equality on coarse and fine. 2) Corner packets: all 8 unique corner+direction keys; four child packets + 0.25 weights. 3) coalesce_paths reconstructed test-side (removed from reference); hop2_valid semantics. 4) Diagnostic field delta vs v0.2. """ import os, sys, unittest, hashlib, platform, resource, importlib.util import numpy as np sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import test_cpu_reference as T ref = T.ref CX, CY, N, NF, LO, HI, H = T.CX, T.CY, T.N, T.NF, T.LO, T.HI, T.H def cls_of(a, b): if T.f_active(a, b): return "A" return "I" if T.f_inner(a, b) else "O" ENC = {"A": ref.ACTIVE, "I": ref.INNER_GHOST, "O": ref.OUTER_GHOST} def reconstruct_coalesce_paths(maps): """Independent path records formerly returned as diagnostics.coalesce_paths.""" rows = [tuple(int(v) for v in r) for r in maps.coalesce.tolist()] cells = np.zeros((len(rows), 4, 2, 2), dtype=np.int64) # [r,k] -> (intermed yx, origin yx) classes = np.zeros((len(rows), 4, 2), dtype=np.int64) for r, (q, ky, kx) in enumerate(rows): for k, (a, b) in enumerate(T.children(kx, ky)): # (x,y) children ym, xm = b - CY[q], a - CX[q] yo, xo = b - 2 * CY[q], a - 2 * CX[q] cells[r, k, 0] = (ym, xm) cells[r, k, 1] = (yo, xo) classes[r, k, 0] = ENC[cls_of(xm, ym)] classes[r, k, 1] = ENC[cls_of(xo, yo)] return {"rows": rows, "cells_yx": cells, "classes": classes, "valid": np.ones(len(rows), dtype=bool)} class TestSRTNanMask(unittest.TestCase): def test_srt_coarse_and_fine_nan_masks_equal(self): spec = ref.make_spec(collision="srt") maps = ref.build_maps(spec) st = T.perturbed_state(spec, 3) fc, ff = st.coarse_f.copy(), st.fine_f.copy() worst = 0.0 for _ in range(3): st, _ = ref.advance_coarse_once(spec, maps, st) fc, ff = T.independent_step(fc, ff, srt=True) np.testing.assert_array_equal(np.isnan(st.coarse_f), np.isnan(fc)) np.testing.assert_array_equal(np.isnan(st.fine_f), np.isnan(ff)) exp_fine_nan = np.array([[not T.f_active(a, b) for a in range(NF)] for b in range(NF)]) np.testing.assert_array_equal(np.isnan(st.fine_f), np.broadcast_to(exp_fine_nan, st.fine_f.shape)) self.assertTrue(np.isfinite(st.fine_f[:, ~exp_fine_nan]).all()) worst = max(worst, float(np.max(np.abs(np.nan_to_num(st.coarse_f - fc)))), float(np.max(np.abs(np.nan_to_num(st.fine_f - ff))))) print(f"\n[srt masks] coarse+fine NaN masks equal over 3 steps; max diff {worst:.3e}") self.assertLessEqual(worst, 1e-14) FLIPS = {"lower_left": (False, False), "lower_right": (True, False), "upper_left": (False, True), "upper_right": (True, True)} MX = (0, 2, 1, 3, 4, 8, 7, 6, 5) MY = (0, 1, 2, 4, 3, 7, 8, 5, 6) def mq(q, fx, fy): if fx: q = MX[q] if fy: q = MY[q] return q class TestCornerIndividualPackets(unittest.TestCase): def test_eight_unique_keys_and_child_packets(self): spec = ref.make_spec(collision="srt") maps = ref.build_maps(spec) keys = [link.key for link in maps.corner_links] self.assertEqual(len(keys), 8) self.assertEqual(len(set(keys)), 8) # expected keys from independent CORNER_PASS geometry expected = set() for tag, (fx, fy) in FLIPS.items(): for q0 in (7, 8): expected.add(f"{tag}_q{mq(q0, fx, fy)}") self.assertEqual(set(keys), expected) eps = 1e-3 exp_avg, g8 = T.TestSRTCornerOracle()._oracle(eps, True) mid8 = 4 * exp_avg - 3 * g8 for name, (fx, fy) in FLIPS.items(): st = ref.make_state(spec) fc = st.coarse_f.copy() sx, sy = (N - 1 - 4 if fx else 4), (N - 1 - 3 if fy else 3) for q, s in ((5, 1), (6, 1), (7, -1), (8, -1)): fc[mq(q, fx, fy), sy, sx] += s * eps out, d = ref.advance_coarse_once(spec, maps, ref.State(fc, st.fine_f, "arrival_f", 0)) self.assertEqual(len(d["corner_packets"]), 8) self.assertEqual(set(d["corner_packets"]), set(keys)) key = f"{name}_q{mq(8, fx, fy)}" cp = d["corner_packets"][key] kx, ky = (N - 1 - 3 if fx else 3), (N - 1 - 4 if fy else 4) q = mq(8, fx, fy) self.assertEqual(cp["direction"], q, key) self.assertEqual(tuple(cp["target_coarse_xy"]), (kx, ky), key) kids_xy = T.children(kx, ky) self.assertEqual({(y, x) for y, x in cp["children_fine_yx"]}, {(b, a) for a, b in kids_xy}, key) ax, ay = (NF - 1 - 3 if fx else 3), (NF - 1 - 5 if fy else 5) for (y, x), p in zip(cp["children_fine_yx"], cp["packets"]): xm, ym = x - CX[q], y - CY[q] expect = mid8 if (x, y) == (ax, ay) else g8 self.assertEqual(cls_of(xm, ym) == "A", (x, y) == (ax, ay), key) self.assertAlmostEqual(p, expect, places=15, msg=f"{key} child {(x, y)}") self.assertEqual(list(cp["weights"]), [0.25] * 4, key) self.assertAlmostEqual(sum(w * p for w, p in zip(cp["weights"], cp["packets"])), cp["coalesced"], places=16) self.assertAlmostEqual(cp["coalesced"], out.coarse_f[q, ky, kx], places=16) self.assertAlmostEqual(cp["coalesced"], exp_avg, places=15) print(f"\n[8 corner keys] {sorted(keys)}; packets 3xg8+1xmid8, weights 0.25, avg={exp_avg!r}") class TestRetainedDiagnostics(unittest.TestCase): @classmethod def setUpClass(cls): cls.spec = ref.make_spec(collision="srt") cls.maps = ref.build_maps(cls.spec) cls.out, cls.d = ref.advance_coarse_once(cls.spec, cls.maps, T.perturbed_state(cls.spec, 6)) def test_coalesce_paths_reconstructed_test_side(self): self.assertNotIn("coalesce_paths", self.d) cp = reconstruct_coalesce_paths(self.maps) rows = cp["rows"] slots = {(q, y, x) for y in range(N) for x in range(N) for q in range(9) if T.slot_kind(q, x, y)} self.assertEqual(len(rows), len(set(rows))) self.assertEqual(set(rows), slots) self.assertEqual(cp["cells_yx"].shape, (len(rows), 4, 2, 2)) self.assertTrue(np.asarray(cp["valid"]).all()) hist = {} for r, (q, ky, kx) in enumerate(rows): kids = {(b, a) for a, b in T.children(kx, ky)} seen = set() for k in range(4): (ym, xm), (yo, xo) = [tuple(int(v) for v in c) for c in cp["cells_yx"][r, k]] y, x = ym + CY[q], xm + CX[q] self.assertIn((y, x), kids) seen.add((y, x)) self.assertEqual((yo, xo), (y - 2 * CY[q], x - 2 * CX[q])) cm, co = cls_of(xm, ym), cls_of(xo, yo) self.assertEqual(int(cp["classes"][r, k, 0]), ENC[cm]) self.assertEqual(int(cp["classes"][r, k, 1]), ENC[co]) self.assertNotEqual(cm, "O") self.assertEqual(cls_of(x, y), "I") kind = T.slot_kind(q, kx, ky) hist[(kind, cm, co)] = hist.get((kind, cm, co), 0) + 1 self.assertEqual(seen, kids) print(f"\n[coalesce_paths test-side] {len(rows)} rows x 4; hist={dict(sorted(hist.items()))}") def test_hop2_mask_semantics(self): self.assertNotIn("fine_valid_after", self.d) v2 = np.asarray(self.d["hop2_valid"]) tgt = lambda a, b: cls_of(a, b) != "O" inb = lambda a, b: 0 <= a < NF and 0 <= b < NF exp1 = np.zeros((9, NF, NF), bool) exp2 = np.zeros((9, NF, NF), bool) for q in range(9): for b in range(NF): for a in range(NF): sa, sb = a - CX[q], b - CY[q] exp1[q, b, a] = tgt(a, b) and inb(sa, sb) for b in range(NF): for a in range(NF): sa, sb = a - CX[q], b - CY[q] exp2[q, b, a] = (tgt(a, b) and inb(sa, sb) and cls_of(sa, sb) != "O" and exp1[q, sb, sa]) np.testing.assert_array_equal(v2, exp2) act = np.array([[T.f_active(a, b) for a in range(NF)] for b in range(NF)]) inner = np.array([[cls_of(a, b) == "I" for a in range(NF)] for b in range(NF)]) self.assertTrue(v2[:, act].all()) self.assertFalse(v2[:, ~(act | inner)].any()) self.assertTrue(np.isnan(self.out.fine_f[:, ~act]).all()) n_valid_inner = int(v2[:, inner].sum()) self.assertGreater(n_valid_inner, 0) reads = self.d["coalesce_reads"] > 0 self.assertTrue(v2[reads].all()) print(f"\n[hop2_valid] matches independent derivation; valid inner={n_valid_inner}; " f"reads hop2-valid={int(reads.sum())}") class TestV02V03FieldDelta(unittest.TestCase): def test_diagnostic_field_delta(self): """Compare top-level diagnostic keys and budget keys between v0.2 and v0.3.""" here = os.path.dirname(os.path.abspath(__file__)) def load(name, path): spec = importlib.util.spec_from_file_location(name, path) mod = importlib.util.module_from_spec(spec) sys.modules[name] = mod spec.loader.exec_module(mod) return mod v2 = load("cpu_ref_v02", os.path.join(here, "cpu_reference_v0.2.py")) v3 = ref s2, s3 = v2.make_spec(collision="identity"), v3.make_spec(collision="identity") m2, m3 = v2.build_maps(s2), v3.build_maps(s3) st2, st3 = v2.make_state(s2), v3.make_state(s3) _, d2 = v2.advance_coarse_once(s2, m2, st2) _, d3 = v3.advance_coarse_once(s3, m3, st3) k2, k3 = set(d2), set(d3) removed = sorted(k2 - k3) added = sorted(k3 - k2) renamed = {"explosion_count", "coalesce_paths", "fine_valid_after"} self.assertTrue(set(removed) <= renamed | {"explosion_count"}) self.assertIn("hop2_valid", added) self.assertNotIn("coalesce_paths", d3) self.assertNotIn("explosion_count", d3) self.assertNotIn("fine_valid_after", d3) self.assertIn("closure_error", d2["budget"]) self.assertIn("bookkeeping_check", d3["budget"]) self.assertNotIn("closure_error", d3["budget"]) self.assertEqual(len(d2["corner_packets"]), 4) # v0.2 overwrite bug self.assertEqual(len(d3["corner_packets"]), 8) print(f"\n[v0.2→v0.3 fields] removed={removed} added={added}; " f"budget rename closure_error→bookkeeping_check; corner_packets 4→8") if __name__ == "__main__": print("reference:", T.REF_PATH, "API", ref.API_VERSION) for p in (T.REF_PATH, T.__file__, os.path.abspath(__file__)): print("sha256", hashlib.sha256(open(p, "rb").read()).hexdigest(), os.path.basename(p)) print("python", sys.version.split()[0], "numpy", np.__version__, platform.platform()) prog = unittest.main(exit=False, verbosity=2) print("peak_rss_MiB", resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024) sys.exit(0 if prog.result.wasSuccessful() else 1)