rune/scripts/check_graph_projection_parity.py

#!/usr/bin/env python3
"""check_graph_projection_parity.py — live parity for the graph_projection
literal-100% residue wave: `_model_to_dicts`, `_now`, `write_all` (new)
plus the projectors they compose with (project_json/dot/mermaid).

Oracle: src/rune/graph_projection.py under RUNE_GP_CORE=python (pure
Python floor). Kernel: the built .so via gp_loader, clock pinned through
the gp_pin_epoch seam (i64::MIN sentinel un-pins). Every comparison is
byte-level: structural JSON equality for _model_to_dicts, exact strings
for _now, exact file bytes + path maps for write_all.
"""
import json
import os
import random
import shutil
import sys
import tempfile
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))

from rune import gp_loader as gp  # noqa: E402
from rune import graph_projection as oracle  # noqa: E402

UNPIN = -(2 ** 63)
FAILURES = []


from contextlib import contextmanager


@contextmanager
def py_pin():
    """Oracle-floor context: RUNE_GP_CORE=python for oracle calls only —
    kernel calls run with the pin off so gp_loader stays live."""
    old = os.environ.get("RUNE_GP_CORE")
    os.environ["RUNE_GP_CORE"] = "python"
    try:
        yield
    finally:
        if old is None:
            os.environ.pop("RUNE_GP_CORE", None)
        else:
            os.environ["RUNE_GP_CORE"] = old


def check(name, ok, detail=""):
    if not ok:
        FAILURES.append(f"{name}: {detail}")
        print(f"FAIL {name}: {detail}")


class M:
    """Attr-branch model (StructuralModel shape: .doc + .project_id)."""

    def __init__(self, doc, pid):
        self.doc = doc
        self.project_id = pid


def fixed(epoch):
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(epoch))


def parity_model_to_dicts():
    cases = [
        ({"entities": [1, 2], "relationships": [], "extra": "x"}, ""),  # dict: no attr -> ""
        ({"relationships": [{"type": "x"}]}, ""),
        ({}, ""),
        ({"entities": [], "relationships": [], "project_id": "k"}, ""),
    ]
    for doc, pid in cases:
        with py_pin():
            want = oracle._model_to_dicts(doc)
        got = gp.model_to_dicts(doc, pid)
        check(f"m2d-dict-{json.dumps(doc)[:24]}", got == want, f"{got} != {want}")
    # attr branch
    model = M({"entities": [{"id": "a"}], "relationships": []}, "proj-9")
    with py_pin():
        want = oracle._model_to_dicts(model)
    got = gp.model_to_dicts(model.doc, model.project_id)
    check("m2d-attr", got == want, f"{got} != {want}")
    # facade delegation end-to-end (live kernel under the hood)
    got2 = oracle._model_to_dicts(model)
    check("m2d-facade-delegates", got2 == want, f"{got2} != {want}")


def parity_now():
    for epoch in (0, -1, 951782400, 1760483570, 4102444800):
        want = fixed(epoch)
        gp.pin_epoch(epoch)
        got = gp.now()
        gp.pin_epoch(UNPIN)
        check(f"now-{epoch}", got == want, f"{got} != {want}")
    # unpinned: kernel wall clock == the oracle law at call time
    got = gp.now()
    want = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    check("now-wall", got in {want, fixed(int(time.time()) + 1)}, f"{got} not current")


def parity_projectors(model_obj, epoch):
    fixed_now = fixed(epoch)
    orig = oracle._now
    oracle._now = lambda: fixed_now
    try:
        gp.pin_epoch(epoch)
        with py_pin():
            pj_o = oracle.project_json(model_obj)
        pj_k = gp._call("gp_project", {"model": oracle._model_to_dicts(model_obj), "format": "json"})
        check("project_json", pj_k == pj_o, f"{json.dumps(pj_k)[:120]} != {json.dumps(pj_o)[:120]}")
        with py_pin():
            dot_o = oracle.project_dot(model_obj)
        dot_k = gp._call("gp_project", {"model": oracle._model_to_dicts(model_obj), "format": "dot"})
        check("project_dot", dot_k == dot_o, f"{dot_k[:80]} != {dot_o[:80]}")
        with py_pin():
            mm_o = oracle.project_mermaid(model_obj)
        mm_k = gp._call("gp_project", {"model": oracle._model_to_dicts(model_obj), "format": "mermaid"})
        check("project_mermaid", mm_k == mm_o, f"{mm_k[:80]} != {mm_o[:80]}")
    finally:
        gp.pin_epoch(UNPIN)
        oracle._now = orig


def parity_write_all(model_obj, epoch, tag):
    fixed_now = fixed(epoch)
    orig = oracle._now
    oracle._now = lambda: fixed_now
    try:
        da = tempfile.mkdtemp(prefix=f"gp-par-o-{tag}-")
        db = tempfile.mkdtemp(prefix=f"gp-par-k-{tag}-")
        with py_pin():
            got_o = oracle.write_all(model_obj, da)
        gp.pin_epoch(epoch)
        got_k = gp.write_all(oracle._model_to_dicts(model_obj), db)
        gp.pin_epoch(UNPIN)
        check(f"write_all-paths-{tag}",
              {k: v.replace(da, "D") for k, v in got_o.items()}
              == {k: v.replace(db, "D") for k, v in got_k.items()},
              f"{got_k} vs {got_o}")
        for f in ("json", "dot", "mermaid"):
            a = Path(got_o[f]).read_bytes()
            b = Path(got_k[f]).read_bytes()
            check(f"write_all-{f}-bytes-{tag}", a == b,
                  f"{a[:100]!r} != {b[:100]!r}")
    finally:
        gp.pin_epoch(UNPIN)
        oracle._now = orig
        shutil.rmtree(da, ignore_errors=True)
        shutil.rmtree(db, ignore_errors=True)


def fuzz(n=60, seed=20261015):
    rng = random.Random(seed)
    kinds = ["domain", "component", "module"]
    for i in range(n):
        ents, rels = [], []
        for j in range(rng.randint(0, 5)):
            quote = '"' if rng.random() < .3 else ""
            eid = f"{rng.choice(kinds)}:n{j}-ö{quote}"
            ents.append({"id": eid, "kind": rng.choice(kinds),
                         **({"identities": {"name": f'N{j} ø"{i}"'}}
                            if rng.random() < .5 else {}),
                         **({"ext": {"source": "s", "other": "x"}}
                            if rng.random() < .4 else {})})
        for j in range(rng.randint(0, 4)):
            if not ents:
                break
            rels.append({"type": rng.choice(["depends_on", "belongs_to"]),
                         "from": rng.choice(ents)["id"],
                         "to": rng.choice(ents)["id"],
                         **({"status": "root"} if rng.random() < .3 else {})})
        model = M({"entities": ents, "relationships": rels}, f"fz-{i}")
        epoch = rng.randint(0, 2 ** 33)
        parity_projectors(model, epoch)
        parity_write_all(model, epoch, f"fuzz{i}")


def main():
    if not gp.available():
        print("FATAL: kernel not built (cargo build --release)")
        return 2
    parity_model_to_dicts()
    parity_now()
    sample = M({
        "entities": [
            {"id": "domain:cøré", "kind": "domain",
             "identities": {"name": 'Cøré "Q" ünïcode'}, "ext": {"source": "m"}},
            {"id": "component:alpha", "kind": "component"},
        ],
        "relationships": [
            {"type": "belongs_to", "from": "component:alpha",
             "to": "domain:cøré", "status": "root"},
            {"type": "depends_on", "from": "component:alpha", "to": "domain:cøré"},
        ],
    }, "proj-x")
    parity_projectors(sample, 951782400)
    parity_write_all(sample, 951782400, "sample")
    empty = M({"entities": [], "relationships": []}, "")
    parity_write_all(empty, 0, "empty")
    fuzz()
    print(f"parity: failures={len(FAILURES)}")
    return 1 if FAILURES else 0


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