#!/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)