rune/scripts/check_graph_memory_parity.py

#!/usr/bin/env python3
"""Live parity: graph_memory Rust kernels vs Python oracle (dichotomy wave,
gap-close pass). Layers:

  1. FROZEN — the gap truth corpus (src/rust/graph_memory/frozen_truth.json,
     generated by scripts/gen_graph_memory_truth.py under the
     scripts/gm_freeze_seams rig against the LIVE oracle): the kernel-owned
     rows (now_iso, strength_to_conf) replayed through the ctypes FFI,
     case by case, byte-exact.
  2. FUZZ — 60 seeded random op-sequences (seed 2479): the full facade
     (add_edge/resolve_entity/query/stats with their kernel decisions)
     executed with RUNE_GRAPH_MEMORY_CORE=python (oracle floor) and with
     the Rust core live, under a pinned clock; per-op results, final
     stats, and a query sweep must be identical.
  3. LIVE — dual-mode determinism: the same facade calls on the same
     hermetic store both pins; supersession chains preserved.

Exit 0 only at 100% parity on the gap-closing functions.
"""
import importlib
import json
import os
import random
import shutil
import sys
from datetime import datetime as _dt
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]
os.environ["RUNE_GRAPH_MEMORY_CORE"] = "python"  # oracle floor to start
SCRATCH = "/tmp/gm_parity"
os.environ["RUNE_MEMORY_DIR"] = SCRATCH

sys.path.insert(0, str(REPO / "src"))

import rune.gm_loader as gl  # noqa: E402
import rune.graph_memory as gm  # noqa: E402

if not gl.available():
    sys.exit("FATAL: Rust core not built — cargo build --release in src/rust/graph_memory")

FROZEN_NOW = "2026-10-13T12:00:00"

fails: list[str] = []
counts = {"frozen": 0, "fuzz": 0, "live": 0}


def fail(layer, name, detail):
    fails.append(f"[{layer}] {name}: {detail}")
    print(f"FAIL[{layer}] {name}: {detail}"[:400])


def j(x):
    return json.dumps(x, sort_keys=True, ensure_ascii=True)


def check(layer, name, want, got):
    counts[layer] += 1
    if j(want) != j(got):
        fail(layer, name, f"want={j(want)[:300]} got={j(got)[:300]}")


def pin(mode):
    if mode == "python":
        os.environ["RUNE_GRAPH_MEMORY_CORE"] = "python"
    else:
        os.environ.pop("RUNE_GRAPH_MEMORY_CORE", None)


def frozen_clock():
    base = _dt.fromisoformat(FROZEN_NOW)

    class FrozenDatetime(_dt):
        @classmethod
        def now(cls, tz=None):
            if tz is not None:
                return base.replace(tzinfo=tz)
            return base

    gm.datetime = FrozenDatetime


def fresh_db():
    shutil.rmtree(os.path.join(SCRATCH, "graph"), ignore_errors=True)
    return gm.connect()


TRUTH = json.loads(
    (REPO / "src" / "rust" / "graph_memory" / "frozen_truth.json").read_text())
CASES = TRUTH["cases"] if isinstance(TRUTH, dict) else TRUTH

# ── layer 1: FROZEN (kernel-owned rows through the FFI) ────────────────────

for row in CASES:
    fn = row["fn"]
    if fn == "now_iso":
        check("frozen", "ffi_now_iso", row["expected"],
              gl.now_iso("2026-10-13T12:00:00.123456+00:00"))
    elif fn == "strength_to_conf":
        (sv,) = row["args"]
        res = gl.strength_to_conf(sv)
        got = res["conf"] if res["ok"] else None
        check("frozen", f"ffi_strength_to_conf:{sv!r}", row["expected"], got)

# ── layers 2+3: FUZZ + LIVE (dual-mode facade on random op-sequences) ───────

NAMES = ["Ada Lovelace", "Babbage", "Analytical Engine", "Difference Engine",
         " punched card ", "Nightflyer", "", "babbage"]
PREDS = ["builds", "knows", "Works On", "frobnicate", "uses", "LIKES", "part_of"]
TYPES = ["person", "project", "alien", "tool", ""]
EVID = ["first notes", "revised notes", "  spaced  ", "x" * 500, ""]

rng = random.Random(2479)

def run_seq(seq, mode):
    pin(mode)
    importlib.reload(gm)
    frozen_clock()
    db = fresh_db()
    out = []
    for op in seq:
        kind = op[0]
        if kind == "add":
            _, subj, pred, obj, kw = op
            out.append(gm.add_edge(db, subj, pred, obj, **kw))
        elif kind == "resolve":
            _, name, nt = op
            out.append(gm.resolve_entity(db, name, nt))
        elif kind == "query":
            _, term, hops = op
            out.append(gm.query(db, term, hops=hops))
        elif kind == "stats":
            out.append(gm.stats(db))
    sweep = [gm.query(db, t, hops=h) for t in ("Ada Lovelace", "babbage",
                                               "Engine", "zzz") for h in (0, 1, 2)]
    out.append(gm.stats(db))
    db.close()
    return out, sweep

for trial in range(60):
    seq = []
    for _ in range(rng.randrange(4, 14)):
        r = rng.random()
        if r < 0.55:
            kw = {}
            if rng.random() < 0.5:
                kw["confidence"] = rng.choice([None, 0.5, 0.9, 0.99, 0.3])
            if rng.random() < 0.6 or "confidence" not in kw:
                kw["strength"] = rng.choice([None, 2, 3, 4, 5, "4", 3.5, "x"])
            if rng.random() < 0.7:
                kw["evidence"] = rng.choice(EVID)
            if rng.random() < 0.4:
                kw["source"] = rng.choice(["truth", "curator", ""])
            if rng.random() < 0.3:
                kw["subj_type"] = rng.choice(TYPES)
            if rng.random() < 0.3:
                kw["obj_type"] = rng.choice(TYPES)
            seq.append(("add", rng.choice(NAMES), rng.choice(PREDS),
                        rng.choice(NAMES), kw))
        elif r < 0.7:
            seq.append(("resolve", rng.choice(NAMES), rng.choice(TYPES)))
        elif r < 0.9:
            seq.append(("query", rng.choice(NAMES + ["Engine", "ada lovelace"]),
                        rng.choice([0, 1, 2, -1])))
        else:
            seq.append(("stats",))
    want_out, want_sweep = run_seq(seq, "python")
    got_out, got_sweep = run_seq(seq, "rust")
    for i, (w, g) in enumerate(zip(want_out, got_out)):
        check("fuzz", f"trial{trial}:op{i}", w, g)
    for i, (w, g) in enumerate(zip(want_sweep, got_sweep)):
        check("fuzz", f"trial{trial}:sweep{i}", w, g)
    if fails:
        break

# ── layer 3: LIVE — pinned-clock determinism + supersession chains ─────────

def chain_probe(mode):
    pin(mode)
    importlib.reload(gm)
    frozen_clock()
    db = fresh_db()
    gm.add_edge(db, "Ada", "uses", "Engine", confidence=0.6, evidence="v1", source="t")
    gm.add_edge(db, "Ada", "uses", "Engine", confidence=0.7, evidence="v2", source="t")
    gm.add_edge(db, "Ada", "uses", "Engine", confidence=0.7, evidence="v2", source="t")  # dup
    gm.add_edge(db, "Ada", "uses", "Engine", confidence=0.65, evidence="v2", source="t")  # weaker dup
    gm.add_edge(db, "Ada", "uses", "Engine", confidence=0.65, evidence="v3", source="t")  # weaker, NEW evidence -> supersede
    rows = [dict(r) for r in db.execute(
        "SELECT id, subj_id, pred, obj_id, confidence, evidence, status, superseded_by "
        "FROM edges ORDER BY id")]
    q = gm.query(db, "engine", hops=1)
    s = gm.stats(db)
    db.close()
    return rows, q, s

want_chain = chain_probe("python")
got_chain = chain_probe("rust")
for i in range(len(want_chain[0])):
    check("live", f"supersession_chain:edge{i}", want_chain[0][i], got_chain[0][i])
check("live", "supersession_chain:query", want_chain[1], got_chain[1])
check("live", "supersession_chain:stats", want_chain[2], got_chain[2])

print(f"parity: {sum(counts.values())} checks "
      f"(frozen {counts['frozen']}, fuzz {counts['fuzz']}, live {counts['live']}) "
      f"— {len(fails)} failures")
sys.exit(1 if fails else 0)