rune/scripts/check_construction_interface_parity.py

#!/usr/bin/env python3
"""Live parity: construction_interface Rust kernels vs the frozen oracle
truth (dichotomy wave, contract factory-gap-ci). Layers:

  1. FROZEN — the frozen truth corpora (frozen_ci_truth.json 85 cases +
     frozen_ci_truth_hash.json 5 shape-pinned cases, generated by
     scripts/gen_construction_interface_truth.py against the LIVE Python
     oracle through the scripts/ci_freeze_seams rig) replayed through
     the ci_replay ctypes FFI entry of the built cdylib, case by case.
  2. LIVE  — a fresh re-freeze of the oracle through the same rig in a
     subprocess; the regenerated corpora must be byte-identical to the
     committed files (the oracle is still the parity authority while the
     strangler runs).

Exit 0 only at 100% parity (85 verbatim + 5 shape-pinned + byte-identical
re-freeze).
"""
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" / "construction_interface"
LIB = CRATE / "target" / "release" / "libconstruction_interface.so"
FROZEN = CRATE / "frozen_ci_truth.json"
FROZEN_HASH = CRATE / "frozen_ci_truth_hash.json"

HASH_ID_PREFIXES = (
    "contract:", "regression:", "verification:",
    "composition:", "acquisition:",
)


def normalize_hash_ids(obj):
    if isinstance(obj, str):
        kind, _, digits = obj.partition(":")
        if (f"{kind}:" in HASH_ID_PREFIXES and digits
                and len(digits) <= 8 and digits.isdigit()):
            return f"{kind}:#"
        return obj
    if isinstance(obj, list):
        return [normalize_hash_ids(v) for v in obj]
    if isinstance(obj, dict):
        return {k: normalize_hash_ids(v) for k, v in obj.items()}
    return obj


def replay(lib, case):
    payload = json.dumps(case, separators=(",", ":")).encode()
    ptr = lib.ci_replay(payload, len(payload))
    if not ptr:
        raise RuntimeError("ci_replay returned NULL")
    text = ctypes.c_char_p(ptr).value.decode()
    lib.ci_free_string(ptr)
    return json.loads(text)


def frozen_layer():
    lib = ctypes.CDLL(str(LIB))
    lib.ci_replay.argtypes = [ctypes.c_char_p, ctypes.c_size_t]
    lib.ci_replay.restype = ctypes.c_void_p
    lib.ci_free_string.argtypes = [ctypes.c_void_p]
    lib.ci_free_string.restype = None
    ok = 0
    cases = json.loads(FROZEN.read_text())
    for case in cases:
        got = replay(lib, case)
        assert got == case["expected"], (
            f"FROZEN divergence: {case['fn']} "
            f"{case['args'][0] if case['args'] else '?'}")
        ok += 1
    hash_ok = 0
    for case in json.loads(FROZEN_HASH.read_text()):
        got = normalize_hash_ids(replay(lib, case))
        want = normalize_hash_ids(case["expected"])
        assert got == want, (
            f"HASH divergence: {case['fn']} "
            f"{case['args'][0] if case['args'] else '?'}")
        hash_ok += 1
    print(f"FROZEN: PARITY OK {ok}/{len(cases)} verbatim "
          f"+ {hash_ok} hash-shape cases")
    return ok, hash_ok


def main() -> int:
    committed_main = FROZEN.read_bytes()
    committed_hash = FROZEN_HASH.read_bytes()
    n_main, n_hash = frozen_layer()

    with tempfile.TemporaryDirectory(prefix="ci_gap_parity_") as _scratch:
        # the re-freeze must run at the CANONICAL rig home — the mini-doc
        # paths are embedded in the frozen truth, so a different
        # CI_FREEZE_HOME changes the bytes without changing behavior
        home = "/tmp/ci_gap_freeze_home"
        subprocess.run(["rm", "-rf", home], check=True)
        env = dict(
            os.environ,
            CI_FREEZE_HOME=home,
            PYTHONPATH="scripts/ci_freeze_seams:src",
            CS_RUST_LIB="/mnt/workdrive/rune/src/rust/capability_scoring/"
                         "target/release/libcapability_scoring.so",
        )
        r = subprocess.run(
            [sys.executable, "scripts/gen_construction_interface_truth.py"],
            cwd=str(REPO), env=env, capture_output=True, text=True, timeout=600)
        if r.returncode != 0:
            print(r.stdout[-2000:])
            print(r.stderr[-2000:])
            print("LIVE: re-freeze FAILED")
            return 1
        same_main = FROZEN.read_bytes() == committed_main
        same_hash = FROZEN_HASH.read_bytes() == committed_hash
    if not (same_main and same_hash):
        print(f"LIVE: re-freeze NOT byte-identical "
              f"(main={same_main} hash={same_hash})")
        return 1
    print(f"LIVE: fresh re-freeze byte-identical "
          f"({n_main}+{n_hash} cases, oracle authority held)")
    print("PARITY OK construction_interface")
    return 0


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