rune/scripts/check_construction_pattern_parity.py

#!/usr/bin/env python3
"""check_construction_pattern_parity.py — live parity: construction_pattern
facade under BOTH core pins vs the frozen oracle truth (dichotomy wave).

Per pin (RUNE_CP_CORE=python and unset → Rust via cp_loader) it replays
every frozen case (frozen_cp_truth.json, 96 cases) through the LIVE
facade. The class surface (construct/revise) does not sit behind the
_core() seam — under the kernel pin those replay through cp_loader's
construct/revise (the cp_construct/cp_revise externs), proving the .so
accepts the same inputs and answers byte-identically.

Usage: python3 scripts/check_construction_pattern_parity.py
Exit 0 iff every case matches its frozen expected in BOTH pins.
"""
from __future__ import annotations

import copy
import json
import os
import sys
from pathlib import Path
from types import SimpleNamespace

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

import rune.construction_pattern as cp  # noqa: E402
from rune import cp_loader  # noqa: E402

TRUTH = (
    Path(__file__).resolve().parents[1]
    / "src" / "rust" / "construction_pattern" / "src" / "frozen_cp_truth.json"
)

FAILS = []
CHECKS = [0]


def call(pin: str, fn, *args, **kwargs):
    old = os.environ.get("RUNE_CP_CORE")
    os.environ["RUNE_CP_CORE"] = pin
    try:
        return fn(*args, **kwargs)
    finally:
        if old is None:
            os.environ.pop("RUNE_CP_CORE", None)
        else:
            os.environ["RUNE_CP_CORE"] = old


def need_obj(d: dict):
    return SimpleNamespace(
        objective=d["objective"],
        affected_surface=list(d["affected_surface"]),
        constraints=dict(d["constraints"]),
    )


def replay(op: str, inp: dict):
    """The oracle-path call for one case (kernel path = same facade fns;
    construct/revise route through cp_loader under the kernel pin).
    Inputs are deep-copied per call: the class revise() appends to a
    revision_history list it shares with the caller (documented oracle
    quirk) — without the copy the oracle-pin run would corrupt the case
    for the kernel-pin run."""
    inp = copy.deepcopy(inp)
    if op == "construct":
        return cp.ConstructionPattern(**dict(inp["kwargs"])).to_dict()
    if op == "revise":
        pattern = cp.ConstructionPattern.from_dict(dict(inp["pattern"]))
        return pattern.revise(
            reason=inp["reason"],
            evidence=list(inp["evidence"]),
            changes=dict(inp["changes"]) if inp.get("changes") is not None else None,
        ).to_dict()
    if op == "propose":
        if isinstance(inp["confidence"], list):
            return [
                cp.record_pattern_proposal(
                    inp["pattern_id"],
                    provider=inp["provider"],
                    model=inp["model"],
                    evidence=list(inp["evidence"]),
                    confidence=b,
                )
                for b in inp["confidence"]
            ]
        return cp.record_pattern_proposal(
            inp["pattern_id"],
            provider=inp["provider"],
            model=inp["model"],
            evidence=list(inp["evidence"]),
            confidence=inp["confidence"],
        )
    if op == "evaluate":
        pattern = cp.ConstructionPattern.from_dict(dict(inp["pattern"]))
        return cp.evaluate_applicability(
            pattern, need_obj(inp["need"]), dict(inp["surface_result"]), dict(inp["authorities"])
        )
    if op == "derive":
        pattern = cp.ConstructionPattern.from_dict(dict(inp["pattern"]))
        return cp.derive_pattern_obligations(
            pattern,
            need_obj(inp["need"]),
            dict(inp["surface_result"]),
            dict(inp["authorities"]),
            project_exceptions=list(inp.get("project_exceptions") or []),
        )
    if op == "combine":
        return cp.combine_pattern_obligations(list(inp["derivations"]))
    if op == "path_match":
        return cp._path_matches(inp["signal"], inp["path"])
    if op == "entity_match":
        return cp._entity_matches(inp["signal"], inp["entity_id"])
    if op == "need_dict":
        return cp._need_dict(need_obj(inp["need"]))
    if op == "surface_text":
        return sorted(cp._surface_text(need_obj(inp["need"]), dict(inp["surface_result"])))
    if op == "when_hit":
        return list(
            cp._obligation_when_hit(
                dict(inp["obligation"]),
                need_obj(inp["need"]),
                dict(inp["surface_result"]),
                dict(inp["authorities"]),
            )
        )
    raise ValueError(f"unknown op {op}")


def replay_kernel_class_ops(op: str, inp: dict):
    """construct/revise through the live kernel externs (cp_loader)."""
    inp = copy.deepcopy(inp)
    if op == "construct":
        return cp_loader.construct(dict(inp["kwargs"]))
    if op == "revise":
        payload = {
            "pattern": dict(inp["pattern"]),
            "reason": inp["reason"],
            "evidence": list(inp["evidence"]),
        }
        if inp.get("changes") is not None:
            payload["changes"] = dict(inp["changes"])
        return cp_loader.revise(**payload)
    raise ValueError(f"not a class op {op}")


def check(name: str, got, want):
    CHECKS[0] += 1
    g = json.dumps(got, sort_keys=True, default=str)
    w = json.dumps(want, sort_keys=True, default=str)
    if g != w:
        FAILS.append(name)
        print(f"FAIL {name}\n  got:    {g[:220]}\n  want:   {w[:220]}")


def run_case(c: dict, pin: str):
    name = f"[{pin}] {c['name']}"
    op, inp = c["op"], c["input"]
    want = c["expected"]
    try:
        if pin == "kernel" and op in ("construct", "revise"):
            got = call("auto", replay_kernel_class_ops, op, inp)
            check(name, got, want)
            return
        got = call(pins[pin], replay, op, inp)
        check(name, got, want)
    except ValueError as e:
        if isinstance(want, dict) and "error" in want:
            check(name, str(e), want["error"])
        else:
            FAILS.append(name)
            print(f"FAIL {name}: unexpected ValueError {e}")
    except RuntimeError as e:
        # kernel envelope errors (cp_loader) carry the oracle's exact text
        if isinstance(want, dict) and "error" in want:
            check(name, str(e), want["error"])
        else:
            FAILS.append(name)
            print(f"FAIL {name}: unexpected RuntimeError {e}")


pins = {"oracle": "python", "kernel": "auto"}

if __name__ == "__main__":
    if not cp_loader.available():
        print("kernel .so not built — parity is vacuous without it")
        sys.exit(1)
    cases = json.loads(TRUTH.read_text())["cases"]
    for c in cases:
        for pin in ("oracle", "kernel"):
            run_case(c, pin)
    print(f"parity: {CHECKS[0] - len(FAILS)}/{CHECKS[0]} checks pass (both pins)")
    sys.exit(1 if FAILS else 0)