Nick
5605a1c0d4
Port the 7 facade-glue fns name-preserving into the existing Rust crate (_now, _raise_core_err, to_dict, from_dict, save, _ledger_state, _absorb) + 7 FFI twins; truth +20 residue rows (36 total, gen x2 byte-identical); battery replays all rows incl. file-byte save/_absorb; live parity 100% (scripts/check_ic_parity.py: 10/10 dual-pin facade, 36/36 FFI truth rows, 14/14 live residue); clippy -D warnings clean; Python suite 26/26 both pins. Audit: intent_continuity 6/13 PARTIAL -> 13/13 FULL RUST.
293 lines
12 KiB
Python
293 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""check_ic_parity.py — live parity for the intent_continuity literal-100%
|
|
residue sweep (dichotomy wave #2479, lane L1).
|
|
|
|
Three channels, all must be 100%:
|
|
1. Facade dual-pin scenarios: one rich ledger scenario through the LIVE
|
|
delegation (RUNE_IC_CORE=auto, Rust kernel via IC_RUST_LIB) and the
|
|
oracle-pinned floor (RUNE_IC_CORE=python) in SEPARATE clean subprocesses
|
|
— record states, saved file bytes, and crash class+messages compared.
|
|
2. Frozen-truth FFI replay: every row of frozen_ic_truth.json (kernel ops
|
|
AND residue rows) replayed through the .so FFI against the frozen
|
|
oracle expectations.
|
|
3. Live residue-fn parity: the Rust residue twins vs the LIVE oracle
|
|
functions on shared runtime-generated inputs (to_dict / from_dict /
|
|
save bytes / _ledger_state / _absorb / _raise_core_err / _now real
|
|
clock, same-second with tick-retry).
|
|
|
|
Output: verification/ic_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" / "intent_continuity"
|
|
TRUTH = CRATE / "frozen_ic_truth.json"
|
|
|
|
SCENARIO = r"""
|
|
import json, os, sys, tempfile
|
|
from pathlib import Path
|
|
sys.path.insert(0, os.environ["IC_CHECK_SRC"])
|
|
from rune import intent_continuity as ic
|
|
|
|
out = []
|
|
td = tempfile.mkdtemp(prefix="ic_parity_")
|
|
|
|
def catch(fn):
|
|
try:
|
|
return fn()
|
|
except Exception as exc:
|
|
return {"py_error_type": type(exc).__name__, "py_error_msg": str(exc)}
|
|
|
|
led = ic.IntentLedger(os.path.join(td, "a", "intents.json"))
|
|
led.create(id="t1", objective="ship PL1", classification="BLOCKER",
|
|
success_condition="transfer shown")
|
|
led.create(id="t2", objective="rewrite recall", classification="EXTENSION",
|
|
unresolved_obligations=["o1", "o2"])
|
|
led.create_detour(id="d1", objective="hotfix CI", classification="BLOCKER",
|
|
parent_intent="t1", return_when="CI green", return_to="t1")
|
|
out.append([r.to_dict() for r in led.records.values()])
|
|
out.append(Path(led.path).read_text())
|
|
led.resolve("t2")
|
|
out.append(led.records["t2"].status)
|
|
parent = led.resolve_detour("d1")
|
|
out.append([parent.id, parent.status, led.records["d1"].status])
|
|
out.append(catch(lambda: led.supersede("t2", "t2b", objective="v2",
|
|
classification="SUPERSEDING",
|
|
carry_obligations=False)))
|
|
out.append(catch(lambda: led.supersede("t2", "t2c", objective="v3",
|
|
classification="SUPERSEDING",
|
|
carry_obligations=True).to_dict()))
|
|
out.append(catch(lambda: led.supersede("missing", "x", objective="o",
|
|
classification="FIX")))
|
|
led2 = ic.IntentLedger(led.path) # reload round-trip
|
|
out.append([r.to_dict() for r in led2.records.values()])
|
|
out.append(catch(lambda: led2.create(id="bad", objective="o",
|
|
classification="BOGUS")))
|
|
out.append(Path(led2.path).read_text())
|
|
print(json.dumps(out))
|
|
"""
|
|
|
|
|
|
def run_mode(mode: str) -> list:
|
|
env = dict(os.environ)
|
|
env["RUNE_IC_CORE"] = mode
|
|
env["IC_CHECK_SRC"] = str(REPO / "src")
|
|
env["IC_RUST_LIB"] = env.get("IC_RUST_LIB", str(CRATE / "target" / "release" / "libintent_continuity.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("IC_RUST_LIB", CRATE / "target" / "release" / "libintent_continuity.so"))
|
|
if not lib_path.is_file():
|
|
raise RuntimeError(f"intent_continuity .so missing: {lib_path}")
|
|
lib = ctypes.CDLL(str(lib_path))
|
|
for sym in ("ic_create", "ic_create_detour", "ic_resolve", "ic_resolve_detour",
|
|
"ic_supersede", "ic_now", "ic_raise_core_err", "ic_to_dict",
|
|
"ic_from_dict", "ic_save", "ic_ledger_state", "ic_absorb"):
|
|
fn = getattr(lib, sym)
|
|
fn.argtypes = [ctypes.c_void_p]
|
|
fn.restype = ctypes.c_void_p
|
|
try:
|
|
lib.cs_free_string
|
|
lib._free = lib.cs_free_string
|
|
except AttributeError:
|
|
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_call(lib: ctypes.CDLL, sym: str, payload: dict):
|
|
data = json.dumps(payload).encode()
|
|
buf = ctypes.create_string_buffer(data)
|
|
ptr = getattr(lib, sym)(ctypes.cast(buf, ctypes.c_void_p))
|
|
if not ptr:
|
|
raise RuntimeError(f"{sym} returned null")
|
|
out = ctypes.string_at(ptr).decode()
|
|
lib._free(ptr)
|
|
return json.loads(out)
|
|
|
|
|
|
def expect(v): # unwrap doc envelope
|
|
if not v.get("ok"):
|
|
raise RuntimeError(v.get("err", "kernel failed"))
|
|
return v["value"]
|
|
|
|
|
|
def replay_truth_rows(lib: ctypes.CDLL) -> tuple[int, list[str]]:
|
|
rows = json.loads(TRUTH.read_text())["cases"]
|
|
ok, errors = 0, []
|
|
for i, case in enumerate(rows):
|
|
fn, args, expected = case["fn"], case["args"], case["expected"]
|
|
label = f"row {i}: {fn}"
|
|
try:
|
|
if fn.startswith("ic_"):
|
|
env = ffi_call(lib, fn, args)
|
|
# error rows freeze the {"err": ...} envelope shape
|
|
got = env["value"] if env.get("ok") else {"err": env.get("err")}
|
|
want = expected
|
|
elif fn == "_now":
|
|
got = expect(ffi_call(lib, "ic_now", {"now": args["now"]}))
|
|
want = expected["now"]
|
|
elif fn == "_raise_core_err":
|
|
got = expect(ffi_call(lib, "ic_raise_core_err", {"err": args["err"]}))
|
|
want = expected
|
|
elif fn in ("to_dict", "from_dict"):
|
|
got = expect(ffi_call(lib, "ic_" + fn, {"record": args["record"]}))
|
|
want = expected["record"]
|
|
elif fn == "save":
|
|
with tempfile.TemporaryDirectory() as td:
|
|
p = os.path.join(td, "sub", "dir", "intents.json")
|
|
ffi_call(lib, "ic_save", {"path": p, "ledger": args["ledger"]})
|
|
got = Path(p).read_text()
|
|
want = expected["file"]
|
|
elif fn == "_ledger_state":
|
|
got = expect(ffi_call(lib, "ic_ledger_state", {"ledger": args["ledger"]}))
|
|
want = expected
|
|
elif fn == "_absorb":
|
|
with tempfile.TemporaryDirectory() as td:
|
|
p = os.path.join(td, "absorb.json")
|
|
try:
|
|
v = expect(ffi_call(lib, "ic_absorb", {"core_result": args["core_result"], "path": p}))
|
|
got = {"ledger": v["ledger"], "file": Path(p).read_text()}
|
|
except RuntimeError as exc:
|
|
got = {"err": str(exc)}
|
|
want = expected
|
|
else:
|
|
errors.append(f"{label}: unknown fn")
|
|
continue
|
|
except Exception as exc: # noqa: BLE001 — crash channel
|
|
errors.append(f"{label}: crashed {exc!r}")
|
|
continue
|
|
if json.dumps(got, sort_keys=True) == json.dumps(want, sort_keys=True):
|
|
ok += 1
|
|
else:
|
|
errors.append(f"{label}: {json.dumps(got)[:120]!r} != {json.dumps(want)[:120]!r}")
|
|
return ok, errors
|
|
|
|
|
|
def live_residue_parity(lib: ctypes.CDLL) -> tuple[int, list[str]]:
|
|
"""Rust residue twins vs the LIVE oracle fns on runtime-generated inputs."""
|
|
sys.path.insert(0, str(REPO / "src"))
|
|
os.environ["RUNE_IC_CORE"] = "python" # observe the oracle floor directly
|
|
from rune import intent_continuity as ic
|
|
|
|
ok, errors = 0, []
|
|
|
|
def check(label, got, want):
|
|
nonlocal ok
|
|
if json.dumps(got, sort_keys=True) == json.dumps(want, sort_keys=True):
|
|
ok += 1
|
|
else:
|
|
errors.append(f"{label}: {json.dumps(got)[:140]!r} != {json.dumps(want)[:140]!r}")
|
|
|
|
td = tempfile.mkdtemp(prefix="ic_residue_")
|
|
led = ic.IntentLedger(os.path.join(td, "o", "intents.json"))
|
|
led.create(id="t1", objective="héllo — ünïcode → 测试", classification="FIX")
|
|
led.create(id="t2", objective="second", classification="EXTENSION",
|
|
unresolved_obligations=["ø1"])
|
|
led.create_detour(id="d1", objective="detour", classification="BLOCKER",
|
|
parent_intent="t1", return_when="w", return_to="t1")
|
|
|
|
# to_dict / from_dict
|
|
for rid in ("t1", "t2", "d1"):
|
|
d = led.records[rid].to_dict()
|
|
check(f"to_dict {rid}", expect(ffi_call(lib, "ic_to_dict", {"record": d})), d)
|
|
check(f"from_dict {rid}", expect(ffi_call(lib, "ic_from_dict", {"record": d})),
|
|
ic.IntentRecord.from_dict(d).to_dict())
|
|
partial = {"id": "p", "status": "ACTIVE", "classification": "FIX"}
|
|
check("from_dict partial", expect(ffi_call(lib, "ic_from_dict", {"record": partial})),
|
|
ic.IntentRecord.from_dict(partial).to_dict())
|
|
|
|
# save bytes
|
|
oracle_path = os.path.join(td, "o", "intents.json")
|
|
rust_path = os.path.join(td, "r", "sub", "intents.json")
|
|
led.save()
|
|
ffi_call(lib, "ic_save", {"path": rust_path, "ledger": led._ledger_state()})
|
|
check("save bytes", Path(rust_path).read_text(), Path(oracle_path).read_text())
|
|
|
|
# _ledger_state
|
|
check("_ledger_state", expect(ffi_call(lib, "ic_ledger_state", {"ledger": led._ledger_state()})),
|
|
led._ledger_state())
|
|
|
|
# _absorb (rebuild + persist)
|
|
core_result = {"ledger": led._ledger_state()}
|
|
p_oracle = os.path.join(td, "oa", "absorb.json")
|
|
p_rust = os.path.join(td, "ra", "absorb.json")
|
|
led_a = ic.IntentLedger(p_oracle)
|
|
led_a._absorb(core_result)
|
|
v = expect(ffi_call(lib, "ic_absorb", {"core_result": core_result, "path": p_rust}))
|
|
check("_absorb ledger", v["ledger"], led_a._ledger_state())
|
|
check("_absorb file", Path(p_rust).read_text(), Path(p_oracle).read_text())
|
|
|
|
# _raise_core_err
|
|
for msg in ("KeyError: t9", "cannot supersede t9: 3 unresolved obligations"):
|
|
try:
|
|
ic._raise_core_err(RuntimeError(msg))
|
|
oracle = {"class": "no-raise"}
|
|
except KeyError as e:
|
|
oracle = {"class": "KeyError", "message": e.args[0]}
|
|
except ValueError as e:
|
|
oracle = {"class": "ValueError", "message": str(e)}
|
|
check(f"_raise_core_err {msg[:20]}",
|
|
expect(ffi_call(lib, "ic_raise_core_err", {"err": msg})), oracle)
|
|
|
|
# _now — real clock, same-second with tick-retry
|
|
for _ in range(3):
|
|
py_now = ic._now()
|
|
rust_now = expect(ffi_call(lib, "ic_now", {}))
|
|
if py_now == rust_now or ic._now() == rust_now:
|
|
check("_now real clock", rust_now, rust_now)
|
|
break
|
|
else:
|
|
errors.append("_now real clock: never agreed within a second")
|
|
|
|
return ok, errors
|
|
|
|
|
|
def main() -> int:
|
|
(REPO / "verification").mkdir(exist_ok=True)
|
|
log_lines = []
|
|
|
|
live = run_mode("auto")
|
|
pinned = run_mode("python")
|
|
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)[:150]!r} pinned={json.dumps(y)[:150]!r}")
|
|
log_lines.append(f"facade scenarios: {len(live)} compared, {len(failures)} mismatches")
|
|
|
|
lib = load_lib()
|
|
t_ok, t_err = replay_truth_rows(lib)
|
|
log_lines.append(f"frozen-truth FFI rows: {t_ok}/{len(json.loads(TRUTH.read_text())['cases'])} ok")
|
|
|
|
r_ok, r_err = live_residue_parity(lib)
|
|
log_lines.append(f"live residue rows: {r_ok} ok, {len(r_err)} mismatches")
|
|
|
|
for f in failures + t_err + r_err:
|
|
log_lines.append(" FAIL " + f[:240])
|
|
|
|
verdict = not failures and not t_err and not r_err
|
|
log_lines.append("PARITY 100%" if verdict else "PARITY FAILURES")
|
|
report = "\n".join(log_lines)
|
|
(REPO / "verification" / "ic_parity.log").write_text(report + "\n")
|
|
print(report)
|
|
return 0 if verdict else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|