rune/scripts/check_sr_parity.py

#!/usr/bin/env python3
"""check_sr_parity.py — live parity for the system_registry literal-100%
residue sweep (dichotomy wave #2479, lane L1).

Channels (all must be 100%):
1. Live `_sr_env` parity: the Rust twin (FFI sr_env) vs the LIVE oracle
   `_sr_env()` under a sentinel RUNE_HOME rig — same roots in, identical
   env out.
2. Facade dual-pin: settings/registry/discover/scan scenarios through the
   LIVE delegation (RUNE_SYSTEM_REGISTRY_CORE=auto, SR_RUST_LIB) vs the
   oracle-pinned floor (RUNE_SYSTEM_REGISTRY_CORE=python) in separate
   subprocesses with a pinned RUNE_HOME sandbox.

Output: verification/sr_parity.log + stdout verdict. Exit 0 only on 100%.
"""
from __future__ import annotations

import ctypes
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]
CRATE = REPO / "src" / "rust" / "system_registry"

SCENARIO = r"""
import json, os, sys
from pathlib import Path
sys.path.insert(0, os.environ["SR_CHECK_SRC"])
home = os.environ["SR_SANDBOX_HOME"]
os.environ["RUNE_HOME"] = home
from rune import system_registry as sr

out = []
# settings round-trip + dev-mode default
s = sr.load_settings()
out.append(s)
s["mode"] = "prod"
s["verbose_logging"] = False  # keep stdout pure JSON (scan prints)
sr.save_settings(s)
out.append(sr.load_settings()["mode"])
out.append(sr.is_dev_mode())
# env roots under the sentinel home
env = sr._sr_env()
out.append({k: v.replace(home, "HOME") for k, v in env.items()})
# discovery over the sandboxed scripts dir
scripts = Path(env["scripts_dir"])
scripts.mkdir(parents=True, exist_ok=True)
(scripts / "alpha.py").write_text("print('a')")
(scripts / "beta.sh").write_text("echo b")
files = sr.discover_files()
out.append(sorted((f["name"], f["category"]) for f in files
                  if f["name"] in ("alpha.py", "beta.sh")))
# registry scan with pinned now + status lookup
now = "2026-10-15T00:00:00.123456+00:00"
changes = sr.scan_and_register()
# mtime-free projection: (file, type, hash when present)
out.append(sorted((c["file"], c["type"], c.get("hash", c.get("new_hash", ""))[:8])
                  for c in changes))
st = sr.get_file_status("alpha.py")
out.append({k: st[k] for k in ("category", "hash")} if st else None)
# diff after a touch
(scripts / "alpha.py").write_text("print('a2')")
out.append(sorted((c["file"], c.get("type", "modified")) for c in sr.diff_since_last_scan()))
print(json.dumps(out))
"""


def run_mode(mode: str, home: str) -> list:
    env = dict(os.environ)
    env["RUNE_SYSTEM_REGISTRY_CORE"] = mode
    env["SR_CHECK_SRC"] = str(REPO / "src")
    env["SR_SANDBOX_HOME"] = home
    env["SR_RUST_LIB"] = env.get("SR_RUST_LIB", str(CRATE / "target" / "release" / "libsystem_registry.so"))
    proc = subprocess.run([sys.executable, "-c", SCENARIO],
                          capture_output=True, text=True, env=env, timeout=300)
    if proc.returncode:
        raise RuntimeError(f"{mode} runner failed rc={proc.returncode}: {proc.stderr[-1500:]}")
    return json.loads(proc.stdout)


def load_lib() -> ctypes.CDLL:
    lib_path = Path(os.environ.get("SR_RUST_LIB", CRATE / "target" / "release" / "libsystem_registry.so"))
    if not lib_path.is_file():
        raise RuntimeError(f"system_registry .so missing: {lib_path}")
    lib = ctypes.CDLL(str(lib_path))
    fn = lib.sr_env
    fn.argtypes = [ctypes.c_void_p]
    fn.restype = ctypes.c_void_p
    scoring = ctypes.CDLL(str(REPO / "src" / "rust" / "capability_scoring" / "target" / "release" / "libcapability_scoring.so"))
    scoring.cs_free_string.argtypes = [ctypes.c_void_p]
    lib._free = scoring.cs_free_string
    lib._free.argtypes = [ctypes.c_void_p]
    return lib


def ffi_env(lib: ctypes.CDLL, payload: dict) -> dict:
    data = json.dumps(payload).encode()
    buf = ctypes.create_string_buffer(data)
    ptr = lib.sr_env(ctypes.cast(buf, ctypes.c_void_p))
    if not ptr:
        raise RuntimeError("sr_env returned null")
    out = ctypes.string_at(ptr).decode()
    lib._free(ptr)
    doc = json.loads(out)
    if not doc.get("ok"):
        raise RuntimeError(doc.get("err", "sr_env failed"))
    return doc["value"]


def live_env_parity(lib: ctypes.CDLL) -> tuple[int, list[str]]:
    """Rust _sr_env twin vs the live oracle under the sentinel rig."""
    home = tempfile.mkdtemp(prefix="sr_env_parity_")
    env = dict(os.environ)
    env["RUNE_HOME"] = home
    env["RUNE_CONFIG_DIR"] = home + "/config"
    env["RUNE_MEMORY_DIR"] = home + "/memory"
    env["RUNE_SYSTEM_REGISTRY_CORE"] = "python"
    env["SR_CHECK_SRC"] = str(REPO / "src")
    proc = subprocess.run(
        [sys.executable, "-c",
         'import json, os, sys; sys.path.insert(0, os.environ["SR_CHECK_SRC"]); '
         'from rune import system_registry as sr; print(json.dumps(sr._sr_env()))'],
        capture_output=True, text=True, env=env, timeout=120)
    if proc.returncode:
        raise RuntimeError(f"oracle _sr_env runner failed: {proc.stderr[-800:]}")
    oracle_env = json.loads(proc.stdout)
    got = ffi_env(lib, oracle_env)  # same five roots back through the twin
    if json.dumps(got, sort_keys=True) == json.dumps(oracle_env, sort_keys=True):
        return 1, []
    return 0, [f"_sr_env: {got!r} != {oracle_env!r}"]


def main() -> int:
    (REPO / "verification").mkdir(exist_ok=True)
    lines = []

    with tempfile.TemporaryDirectory(prefix="sr_parity_live_") as home_live, \
         tempfile.TemporaryDirectory(prefix="sr_parity_pin_") as home_pin:
        live = run_mode("auto", home_live)
        pinned = run_mode("python", home_pin)
    failures = []
    for i, (x, y) in enumerate(zip(live, pinned)):
        if json.dumps(x, sort_keys=True) != json.dumps(y, sort_keys=True):
            failures.append(f"scenario {i}: live={json.dumps(x)[:160]!r} pinned={json.dumps(y)[:160]!r}")
    lines.append(f"facade dual-pin scenarios: {len(live)} compared, {len(failures)} mismatches")

    lib = load_lib()
    e_ok, e_err = live_env_parity(lib)
    lines.append(f"live _sr_env rows: {e_ok} ok, {len(e_err)} mismatches")

    for f in failures + e_err:
        lines.append("  FAIL " + f[:240])

    verdict = not failures and not e_err
    lines.append("PARITY 100%" if verdict else "PARITY FAILURES")
    report = "\n".join(lines)
    (REPO / "verification" / "sr_parity.log").write_text(report + "\n")
    print(report)
    return 0 if verdict else 1


if __name__ == "__main__":
    raise SystemExit(main())