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.
230 lines
11 KiB
Python
230 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""gen_intent_continuity_truth.py — freeze the intent_continuity parity
|
|
oracle (dichotomy wave #2479 tranche).
|
|
|
|
Generator law: RUNE_IC_CORE=python pinned, the clock STUBBED (deterministic
|
|
per-op `now`), inputs deep-copied, every public transition/branch exercised
|
|
(create, create_detour incl. PARKED-parent no-op, resolve incl. missing and
|
|
already-resolved, resolve_detour incl. re-entry and un-parking, supersede
|
|
incl. obligations guard + KeyError). Output:
|
|
src/rust/intent_continuity/frozen_ic_truth.json — replayed VERBATIM by the
|
|
crate's cfg(test) battery.
|
|
"""
|
|
import copy
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
os.environ["RUNE_IC_CORE"] = "python"
|
|
|
|
import rune.intent_continuity as ic # noqa: E402
|
|
|
|
# deterministic clock: op k gets 2026-10-12T00:00:kZ
|
|
_tick = [0]
|
|
|
|
|
|
def _fake_strftime(_fmt, _t):
|
|
_tick[0] += 1
|
|
return f"2026-10-12T00:00:{_tick[0]:02d}Z"
|
|
|
|
|
|
ic.time = types.SimpleNamespace(strftime=_fake_strftime, gmtime=lambda: 0)
|
|
|
|
cases: list = []
|
|
|
|
|
|
def ledger_state(led: ic.IntentLedger) -> dict:
|
|
return {"intents": [r.to_dict() for r in led.records.values()]}
|
|
|
|
|
|
def record_op(led, fn, op_args, now, expected_record, before, after, err=None):
|
|
"""Freeze one op: kernel args from the pre-op state, expected from the
|
|
observed oracle outcome."""
|
|
args = {"ledger": copy.deepcopy(before), "now": now, **copy.deepcopy(op_args)}
|
|
if err is not None:
|
|
expected = {"err": err}
|
|
else:
|
|
expected = {"ledger": copy.deepcopy(after),
|
|
"record": copy.deepcopy(expected_record)}
|
|
cases.append({"fn": fn, "args": args, "expected": expected})
|
|
|
|
|
|
def op(led, fn, run, **kwargs):
|
|
"""run(**kwargs) performs the oracle op (e.g. led.create); kwargs are
|
|
recorded verbatim as kernel args."""
|
|
before = ledger_state(led)
|
|
now = f"2026-10-12T00:00:{_tick[0] + 1:02d}Z" # the now this op will get
|
|
try:
|
|
rec = run(**kwargs)
|
|
after = ledger_state(led)
|
|
record_op(led, fn, kwargs, now, rec.to_dict() if rec else None, before, after)
|
|
except (ValueError, KeyError) as e:
|
|
# kernel err strings match the oracle's exception text: the facade
|
|
# re-raises KeyError(e.args[0]) whose str is the quoted form, so the
|
|
# kernel carries the BARE id after the prefix
|
|
err = (f"KeyError: {e.args[0]}" if isinstance(e, KeyError) else str(e))
|
|
record_op(led, fn, kwargs, now, None, before, before, err=err)
|
|
return now
|
|
|
|
|
|
def main() -> None:
|
|
with tempfile.TemporaryDirectory() as td:
|
|
led = ic.IntentLedger(os.path.join(td, "intents.json"))
|
|
|
|
op(led, "ic_create", led.create,
|
|
id="t1", objective="ship the parser fix", classification="FIX")
|
|
op(led, "ic_create", led.create,
|
|
id="t2", objective="rewrite recall", classification="EXTENSION",
|
|
success_condition="A/B parity shown",
|
|
unresolved_obligations=["o1", "o2"])
|
|
op(led, "ic_create_detour", led.create_detour,
|
|
id="d1", objective="hotfix CI", classification="BLOCKER",
|
|
parent_intent="t1", return_when="CI green", return_to="t1")
|
|
op(led, "ic_create_detour", led.create_detour,
|
|
id="d2", objective="second detour", classification="LATER",
|
|
parent_intent="t1", # t1 already PARKED
|
|
return_when="later", return_to="t1")
|
|
op(led, "ic_resolve", led.resolve, intent_id="t2")
|
|
op(led, "ic_resolve", led.resolve, intent_id="t2") # already RESOLVED → None
|
|
op(led, "ic_resolve", led.resolve, intent_id="d1") # DETOUR → RESOLVED
|
|
op(led, "ic_resolve_detour", led.resolve_detour, detour_id="d1") # un-parks t1
|
|
op(led, "ic_resolve_detour", led.resolve_detour, detour_id="d1") # re-entry → parent
|
|
op(led, "ic_resolve", led.resolve, intent_id="missing")
|
|
op(led, "ic_resolve_detour", led.resolve_detour, detour_id="missing")
|
|
op(led, "ic_supersede", led.supersede,
|
|
old_id="t2", new_id="t2b", objective="rewrite recall v2",
|
|
classification="SUPERSEDING", carry_obligations=True)
|
|
op(led, "ic_supersede", led.supersede,
|
|
old_id="t2", new_id="t2c", objective="drop obligations",
|
|
classification="SUPERSEDING", carry_obligations=False)
|
|
op(led, "ic_supersede", led.supersede,
|
|
old_id="missing", new_id="x", objective="o", classification="FIX")
|
|
op(led, "ic_create", led.create,
|
|
id="bad", objective="o", classification="BOGUS")
|
|
op(led, "ic_create_detour", led.create_detour,
|
|
id="d3", objective="o", classification="FIX",
|
|
parent_intent="nope", return_when="w", return_to="r")
|
|
|
|
# ── residue surface (literal-100% sweep L1, 2026-10-15) ────────────
|
|
# The facade-glue twins: _now / _raise_core_err / to_dict / from_dict
|
|
# / save / _ledger_state / _absorb, observed on the live oracle.
|
|
|
|
# _now — the stubbed strftime IS the rig; the row pins the echo law
|
|
# (the rig value passed in is what comes out).
|
|
oracle_now = ic._now()
|
|
cases.append({"fn": "_now", "args": {"now": oracle_now},
|
|
"expected": {"now": oracle_now}})
|
|
|
|
# _raise_core_err — observable = exception class + raw message
|
|
# (facade re-raises KeyError(raw); str() quoting is downstream).
|
|
def observe_raise(msg):
|
|
try:
|
|
ic._raise_core_err(RuntimeError(msg))
|
|
return {"class": "no-raise", "message": msg}
|
|
except KeyError as e:
|
|
return {"class": "KeyError", "message": e.args[0]}
|
|
except ValueError as e:
|
|
return {"class": "ValueError", "message": str(e)}
|
|
|
|
for msg in ("KeyError: t2",
|
|
"cannot supersede t2: 2 unresolved obligations",
|
|
"invalid classification 'BOGUS'",
|
|
"KeyError: "):
|
|
cases.append({"fn": "_raise_core_err", "args": {"err": msg},
|
|
"expected": observe_raise(msg)})
|
|
|
|
# to_dict / from_dict — projection onto __slots__ order.
|
|
full = led.records["t2b"].to_dict() # has unresolved_obligations o1,o2
|
|
shuffled = {k: full[k] for k in reversed(list(full))}
|
|
cases.append({"fn": "to_dict", "args": {"record": shuffled},
|
|
"expected": {"record": full}})
|
|
cases.append({"fn": "from_dict", "args": {"record": shuffled},
|
|
"expected": {"record": ic.IntentRecord.from_dict(shuffled).to_dict()}})
|
|
partial = {"id": "x", "status": "ACTIVE", "classification": "FIX"}
|
|
cases.append({"fn": "from_dict", "args": {"record": partial},
|
|
"expected": {"record": ic.IntentRecord.from_dict(partial).to_dict()}})
|
|
with_extra = dict(full, junk="ignored", extra=[1, 2])
|
|
cases.append({"fn": "from_dict", "args": {"record": with_extra},
|
|
"expected": {"record": ic.IntentRecord.from_dict(with_extra).to_dict()}})
|
|
with_null = dict(full, success_condition=None)
|
|
cases.append({"fn": "from_dict", "args": {"record": with_null},
|
|
"expected": {"record": ic.IntentRecord.from_dict(with_null).to_dict()}})
|
|
uni = ic.IntentRecord(id="u1", objective="héllo — ünïcode → 测试",
|
|
classification="FIX",
|
|
unresolved_obligations=["ünïcode øblig"])
|
|
cases.append({"fn": "to_dict", "args": {"record": uni.to_dict()},
|
|
"expected": {"record": uni.to_dict()}})
|
|
|
|
# save — the FILE BYTES are the truth (paths stay per-run tempdirs).
|
|
led.path = Path(td) / "sub" / "dir" / "intents.json"
|
|
led.save()
|
|
cases.append({"fn": "save", "args": {"ledger": ledger_state(led)},
|
|
"expected": {"file": led.path.read_text()}})
|
|
led_empty = ic.IntentLedger(os.path.join(td, "e", "empty.json"))
|
|
led_empty.save()
|
|
cases.append({"fn": "save", "args": {"ledger": {"intents": []}},
|
|
"expected": {"file": led_empty.path.read_text()}})
|
|
led_uni = ic.IntentLedger(os.path.join(td, "u", "uni.json"))
|
|
led_uni.records["u1"] = uni
|
|
led_uni.save()
|
|
cases.append({"fn": "save", "args": {"ledger": ledger_state(led_uni)},
|
|
"expected": {"file": led_uni.path.read_text()}})
|
|
|
|
# _ledger_state — via _load so extras/defaults round-trip like a
|
|
# reloaded ledger (junk dropped, missing slots defaulted).
|
|
lpath = Path(td) / "reload.json"
|
|
lpath.write_text(json.dumps({"intents": [
|
|
{"junk": 1, "id": "a", "status": "ACTIVE", "classification": "FIX"},
|
|
{"id": "b"},
|
|
]}))
|
|
led3 = ic.IntentLedger(lpath)
|
|
cases.append({"fn": "_ledger_state",
|
|
"args": {"ledger": {"intents": [
|
|
{"junk": 1, "id": "a", "status": "ACTIVE", "classification": "FIX"},
|
|
{"id": "b"},
|
|
]}},
|
|
"expected": led3._ledger_state()})
|
|
|
|
# _absorb — rebuild + persist; observable = ledger + file bytes.
|
|
def absorb_row(core_result, err=None):
|
|
p = Path(td) / f"absorb{len(cases)}.json"
|
|
led4 = ic.IntentLedger(p)
|
|
try:
|
|
led4._absorb(copy.deepcopy(core_result))
|
|
exp = {"ledger": led4._ledger_state(), "file": p.read_text()}
|
|
except KeyError as e:
|
|
exp = {"err": f"KeyError: {e.args[0]}"}
|
|
cases.append({"fn": "_absorb",
|
|
"args": {"core_result": copy.deepcopy(core_result)},
|
|
"expected": exp})
|
|
|
|
absorb_row({"ledger": {"intents": [full, led.records["t1"].to_dict()]}})
|
|
absorb_row({"ledger": {"intents": []}})
|
|
absorb_row({}) # missing ledger key → KeyError('ledger')
|
|
absorb_row({"ledger": {"intents": [{"objective": "no id"}]}})
|
|
absorb_row({"ledger": {"intents": [ # duplicate id: first slot, last value
|
|
dict(full, id="dup", objective="first"),
|
|
dict(full, id="dup", objective="second"),
|
|
]}})
|
|
|
|
doc = {
|
|
"module": "intent_continuity",
|
|
"pin": "RUNE_IC_CORE=python",
|
|
"frozen_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"cases": cases,
|
|
}
|
|
out = ROOT / "src" / "rust" / "intent_continuity" / "frozen_ic_truth.json"
|
|
out.write_text(json.dumps(doc, indent=1) + "\n")
|
|
print(f"frozen {len(cases)} cases -> {out}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|