509 lines
18 KiB
Python
509 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""gen_mutation_guards_truth.py — freeze/replay parity truth for
|
|
mutation_guards (dichotomy wave, #2479 tranche).
|
|
|
|
Freeze mode (default): exercises the LIVE Python oracle
|
|
(src/rune/mutation_guards.py — classify_mutation and guard_mutation, the
|
|
module's two public functions) across every branch and records
|
|
(kernel_input, expected_kernel_output) pairs into
|
|
src/rust/mutation_guards/src/frozen_mg_truth.json.
|
|
|
|
Replay mode (--replay): re-runs the SAME scenarios with the Rust core
|
|
delegation ACTIVE (RUNE_MG_CORE unpinned) and asserts both kernel-level
|
|
(loader direct) and facade-level (oracle delegates) parity against the
|
|
frozen file.
|
|
|
|
Poisoned-truth law: freeze pins RUNE_MG_CORE=python BEFORE any import.
|
|
Mutation law: case inputs are deep-copied BEFORE they reach the oracle.
|
|
|
|
Branch inventory (both functions):
|
|
* DENY 1 — unattributed identity-family mutation (identity/persona/
|
|
constitution, falsy attributed); runs FIRST so it beats read ops and
|
|
the destructive gate
|
|
* DENY 2 — destructive (op severity >= 2) on a protected kind
|
|
(severity >= 2) with no backup
|
|
* CONFIRM — score >= 4, or identity-family with any mutating op
|
|
* ALLOW — everything else
|
|
* severity-table defaults: unknown kind -> 1, unknown op -> 1,
|
|
unknown/None scope -> 0
|
|
* Python truthiness through guard_mutation's .get passthrough: the
|
|
default applies only when the KEY is missing — present-but-null
|
|
attributed is falsy (unattributed), falsy 0/[]/"yes"-truthy pins the
|
|
quirk; the f-string renders a null kind/operation as "None"
|
|
"""
|
|
import copy
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
TRUTH_PATH = os.path.join(REPO, "src", "rust", "mutation_guards",
|
|
"src", "frozen_mg_truth.json")
|
|
REPLAY = "--replay" in sys.argv
|
|
|
|
if not REPLAY:
|
|
os.environ["RUNE_MG_CORE"] = "python" # poisoned-truth law (freeze side)
|
|
else:
|
|
os.environ.pop("RUNE_MG_CORE", None) # live-parity law (facade delegates)
|
|
|
|
sys.path.insert(0, os.path.join(REPO, "src"))
|
|
sys.path.insert(0, os.path.join(REPO, "src", "rune"))
|
|
|
|
import mutation_guards as oracle_mod # noqa: E402
|
|
|
|
if REPLAY:
|
|
from rune import mu_loader # noqa: E402 (kernel-level replay side)
|
|
if not mu_loader.available():
|
|
print("REPLAY ABORT: kernel not built —", mu_loader.LIB_PATH)
|
|
sys.exit(2)
|
|
|
|
|
|
def facade_classify(inp):
|
|
return oracle_mod.classify_mutation(
|
|
copy.deepcopy(inp["target_kind"]),
|
|
copy.deepcopy(inp["operation"]),
|
|
copy.deepcopy(inp["scope"]),
|
|
copy.deepcopy(inp["attributed"]),
|
|
copy.deepcopy(inp["backed_up"]),
|
|
)
|
|
|
|
|
|
def kernel_classify(inp):
|
|
return mu_loader.classify(**copy.deepcopy(inp))
|
|
|
|
|
|
def facade_guard(inp):
|
|
return oracle_mod.guard_mutation(copy.deepcopy(inp["mutation"]))
|
|
|
|
|
|
def kernel_guard(inp):
|
|
return mu_loader.guard(**copy.deepcopy(inp))
|
|
|
|
|
|
# ── classify_mutation ────────────────────────────────────────────────────────
|
|
|
|
def cl_case(rec, name, target_kind, operation, scope="single",
|
|
attributed=True, backed_up=False):
|
|
inp = {"target_kind": target_kind, "operation": operation,
|
|
"scope": scope, "attributed": attributed, "backed_up": backed_up}
|
|
got = facade_classify(copy.deepcopy(inp))
|
|
rec(name, inp, got, got)
|
|
|
|
|
|
# DENY 1 — unattributed identity family (runs FIRST)
|
|
def c_den1_identity(rec):
|
|
cl_case(rec, "cl-deny1-unattributed-identity-overwrite",
|
|
"identity", "overwrite", "single", False, False)
|
|
|
|
|
|
def c_den1_persona_append_whole(rec):
|
|
# append (op severity 1) would CONFIRM on persona; unattributed wins
|
|
cl_case(rec, "cl-deny1-unattributed-persona-append-whole",
|
|
"persona", "append", "whole", False, False)
|
|
|
|
|
|
def c_den1_constitution_falsy_zero(rec):
|
|
# attributed=0 is falsy (Python truthiness) → unattributed
|
|
cl_case(rec, "cl-deny1-constitution-attributed-zero",
|
|
"constitution", "append", "single", 0, False)
|
|
|
|
|
|
def c_den1_beats_read(rec):
|
|
# even a READ is denied when unattributed — precedence proof
|
|
cl_case(rec, "cl-deny1-identity-read-still-denied",
|
|
"identity", "read", "single", False, False)
|
|
|
|
|
|
def c_den1_null_attributed(rec):
|
|
# explicit None attributed → falsy → DENY
|
|
cl_case(rec, "cl-deny1-identity-attributed-null",
|
|
"identity", "append", "single", None, False)
|
|
|
|
|
|
# DENY 2 — destructive without a snapshot
|
|
def c_den2_constitution_delete_whole(rec):
|
|
cl_case(rec, "cl-deny2-constitution-delete-whole",
|
|
"constitution", "delete", "whole", True, False)
|
|
|
|
|
|
def c_den2_skill_delete_multi(rec):
|
|
cl_case(rec, "cl-deny2-skill-delete-multi",
|
|
"skill", "delete", "multi", True, False)
|
|
|
|
|
|
def c_den2_belief_overwrite_whole(rec):
|
|
# op severity 2 + kind severity 2 boundary: still DENY
|
|
cl_case(rec, "cl-deny2-belief-overwrite-whole-boundary",
|
|
"belief", "overwrite", "whole", True, False)
|
|
|
|
|
|
def c_den2_backed_up_zero(rec):
|
|
# backed_up=0 is falsy → unbacked → DENY
|
|
cl_case(rec, "cl-deny2-skill-overwrite-whole-backed-zero",
|
|
"skill", "overwrite", "whole", True, 0)
|
|
|
|
|
|
# CONFIRM — score >= 4 or identity-family mutating op
|
|
def c_conf_backed_skill_overwrite(rec):
|
|
cl_case(rec, "cl-confirm-backed-skill-overwrite-whole",
|
|
"skill", "overwrite", "whole", True, True)
|
|
|
|
|
|
def c_conf_identity_append_attributed(rec):
|
|
# kind severity 3 + op severity 1 → CONFIRM even at score 4
|
|
cl_case(rec, "cl-confirm-identity-append-attributed",
|
|
"identity", "append", "single", True, False)
|
|
|
|
|
|
def c_conf_identity_read_whole(rec):
|
|
# read (op 0) escapes the identity rule; score 3+0+2=5 ≥ 4 → CONFIRM
|
|
cl_case(rec, "cl-confirm-identity-read-whole-score",
|
|
"identity", "read", "whole", True, False)
|
|
|
|
|
|
def c_conf_operating_delete_single(rec):
|
|
# score exactly 4 — the CONFIRM boundary
|
|
cl_case(rec, "cl-confirm-operating-delete-single-score-4",
|
|
"operating", "delete", "single", True, False)
|
|
|
|
|
|
def c_conf_fact_overwrite_whole(rec):
|
|
# op 2 + kind 1: no DENY (kind < 2), score 5 → CONFIRM
|
|
cl_case(rec, "cl-confirm-fact-overwrite-whole",
|
|
"fact", "overwrite", "whole", True, False)
|
|
|
|
|
|
def c_conf_backed_truthy_string(rec):
|
|
# backed_up="yes" is truthy → skips DENY 2 → CONFIRM
|
|
cl_case(rec, "cl-confirm-skill-delete-multi-backed-truthy",
|
|
"skill", "delete", "multi", True, "yes")
|
|
|
|
|
|
def c_conf_constitution_delete_backed(rec):
|
|
cl_case(rec, "cl-confirm-constitution-delete-whole-backed",
|
|
"constitution", "delete", "whole", True, True)
|
|
|
|
|
|
def c_conf_persona_overwrite_backed(rec):
|
|
cl_case(rec, "cl-confirm-persona-overwrite-whole-backed",
|
|
"persona", "overwrite", "whole", True, True)
|
|
|
|
|
|
def c_conf_archive_delete_whole(rec):
|
|
# archive (kind 0) still CONFIRMs at score 5 — score rule beats kind
|
|
cl_case(rec, "cl-confirm-archive-delete-whole-score",
|
|
"archive", "delete", "whole", True, False)
|
|
|
|
|
|
def c_conf_unknown_kind_delete(rec):
|
|
# unknown kind defaults to severity 1: 1+3+0=4 → CONFIRM
|
|
cl_case(rec, "cl-confirm-unknown-kind-delete",
|
|
"widget", "delete", "single", True, False)
|
|
|
|
|
|
# ALLOW — everything else
|
|
def c_allow_fact_append_single(rec):
|
|
cl_case(rec, "cl-allow-fact-append-single-defaults",
|
|
"fact", "append", "single", True, False)
|
|
|
|
|
|
def c_allow_fact_append_multi(rec):
|
|
# score 3 — directly below the CONFIRM boundary
|
|
cl_case(rec, "cl-allow-fact-append-multi-score-3",
|
|
"fact", "append", "multi", True, False)
|
|
|
|
|
|
def c_allow_identity_read_single(rec):
|
|
# identity + read + single: score 3, identity rule needs op ≥ 1
|
|
cl_case(rec, "cl-allow-identity-read-single",
|
|
"identity", "read", "single", True, False)
|
|
|
|
|
|
def c_allow_archive_delete_single(rec):
|
|
# kind severity 0: even delete is cheap on archives
|
|
cl_case(rec, "cl-allow-archive-delete-single",
|
|
"archive", "delete", "single", True, False)
|
|
|
|
|
|
def c_allow_archive_read_whole(rec):
|
|
cl_case(rec, "cl-allow-archive-read-whole",
|
|
"archive", "read", "whole", True, False)
|
|
|
|
|
|
def c_allow_fact_read_whole(rec):
|
|
cl_case(rec, "cl-allow-fact-read-whole",
|
|
"fact", "read", "whole", True, False)
|
|
|
|
|
|
def c_allow_unknown_operation(rec):
|
|
# unknown op defaults to severity 1: 1+1+1=3 → ALLOW
|
|
cl_case(rec, "cl-allow-unknown-operation",
|
|
"fact", "nuke", "multi", True, False)
|
|
|
|
|
|
def c_allow_unknown_scope(rec):
|
|
# unknown scope defaults to severity 0
|
|
cl_case(rec, "cl-allow-unknown-scope",
|
|
"fact", "append", "galaxy", True, False)
|
|
|
|
|
|
def c_allow_null_kind(rec):
|
|
# None kind: severity default + f-string renders "None"
|
|
cl_case(rec, "cl-allow-null-kind-renders-None",
|
|
None, "append", "single", True, False)
|
|
|
|
|
|
def c_allow_null_operation(rec):
|
|
cl_case(rec, "cl-allow-null-operation-renders-None",
|
|
"fact", None, "single", True, False)
|
|
|
|
|
|
def c_allow_null_scope(rec):
|
|
# None scope: severity 0 (dict lookup miss)
|
|
cl_case(rec, "cl-allow-null-scope",
|
|
"fact", "append", None, True, False)
|
|
|
|
|
|
# ── guard_mutation ───────────────────────────────────────────────────────────
|
|
|
|
def gu_case(rec, name, mutation):
|
|
inp = {"mutation": mutation}
|
|
got = facade_guard(copy.deepcopy(inp))
|
|
rec(name, inp, got, got)
|
|
|
|
|
|
def g_empty_dict_defaults(rec):
|
|
gu_case(rec, "gu-empty-dict-all-defaults", {})
|
|
|
|
|
|
def g_fact_append_defaults(rec):
|
|
gu_case(rec, "gu-fact-append-allow", {"target_kind": "fact",
|
|
"operation": "append"})
|
|
|
|
|
|
def g_unattributed_identity(rec):
|
|
gu_case(rec, "gu-unattributed-identity-deny",
|
|
{"target_kind": "identity", "operation": "overwrite",
|
|
"attributed": False})
|
|
|
|
|
|
def g_unbacked_constitution_delete(rec):
|
|
gu_case(rec, "gu-unbacked-constitution-delete-deny",
|
|
{"target_kind": "constitution", "operation": "delete",
|
|
"scope": "whole", "attributed": True, "backed_up": False})
|
|
|
|
|
|
def g_backed_skill_confirms(rec):
|
|
gu_case(rec, "gu-backed-skill-overwrite-confirm",
|
|
{"target_kind": "skill", "operation": "overwrite",
|
|
"scope": "whole", "attributed": True, "backed_up": True})
|
|
|
|
|
|
def g_null_attributed_present(rec):
|
|
# present-but-null attributed is FALSY — the default True applies
|
|
# only to a MISSING key
|
|
gu_case(rec, "gu-attributed-null-falsy-deny",
|
|
{"target_kind": "identity", "attributed": None})
|
|
|
|
|
|
def g_empty_list_attributed(rec):
|
|
gu_case(rec, "gu-attributed-empty-list-falsy-deny",
|
|
{"target_kind": "persona", "operation": "delete",
|
|
"attributed": []})
|
|
|
|
|
|
def g_backed_up_zero(rec):
|
|
gu_case(rec, "gu-backed-up-zero-falsy-deny",
|
|
{"target_kind": "skill", "operation": "overwrite",
|
|
"scope": "whole", "backed_up": 0})
|
|
|
|
|
|
def g_backed_up_truthy_string(rec):
|
|
gu_case(rec, "gu-backed-up-truthy-string-confirm",
|
|
{"target_kind": "skill", "operation": "overwrite",
|
|
"scope": "whole", "backed_up": "yes"})
|
|
|
|
|
|
def g_read_allows(rec):
|
|
gu_case(rec, "gu-read-allows", {"target_kind": "fact",
|
|
"operation": "read"})
|
|
|
|
|
|
def g_extra_keys_ignored(rec):
|
|
gu_case(rec, "gu-extra-keys-ignored",
|
|
{"target_kind": "fact", "operation": "append",
|
|
"note": "x", "weight": 9})
|
|
|
|
|
|
def g_null_target_kind(rec):
|
|
# present-null target_kind beats the "fact" default (get quirk);
|
|
# severity defaults to 1, f-string renders "None"
|
|
gu_case(rec, "gu-null-target-kind-renders-None",
|
|
{"target_kind": None})
|
|
|
|
|
|
def g_empty_operation_string(rec):
|
|
# empty op string: severity default 1, f-string renders ""
|
|
gu_case(rec, "gu-empty-operation-string",
|
|
{"target_kind": "fact", "operation": ""})
|
|
|
|
|
|
# ── _mg_core (the loader-decision residue) ────────────────────────────────────
|
|
|
|
def _probe_once(rec, name, env):
|
|
"""Freeze the oracle's _mg_core() decision under one env state.
|
|
|
|
The env flip + restore brackets the probe so the poisoned-truth pin
|
|
(python) holds for every facade case captured around it. `env=None`
|
|
probes the unset state (os.environ.get default 'auto'). The auto/unset
|
|
rows record the mu_loader module name when the .so is built — the
|
|
replay env guarantees it is."""
|
|
old = os.environ.get("RUNE_MG_CORE")
|
|
if env is None:
|
|
os.environ.pop("RUNE_MG_CORE", None)
|
|
else:
|
|
os.environ["RUNE_MG_CORE"] = env
|
|
try:
|
|
core = oracle_mod._mg_core()
|
|
expected = {"core": None if core is None
|
|
else getattr(core, "__name__", str(core))}
|
|
rec(name, {"env": env}, dict(expected), expected)
|
|
finally:
|
|
if old is None:
|
|
os.environ.pop("RUNE_MG_CORE", None)
|
|
else:
|
|
os.environ["RUNE_MG_CORE"] = old
|
|
|
|
|
|
def p_python_pin(rec):
|
|
_probe_once(rec, "probe-python-pin", "python")
|
|
|
|
|
|
def p_auto(rec):
|
|
_probe_once(rec, "probe-auto", "auto")
|
|
|
|
|
|
def p_unset(rec):
|
|
_probe_once(rec, "probe-unset", None)
|
|
|
|
|
|
# ── driver ───────────────────────────────────────────────────────────────────
|
|
|
|
CLASSIFY = [
|
|
c_den1_identity, c_den1_persona_append_whole,
|
|
c_den1_constitution_falsy_zero, c_den1_beats_read,
|
|
c_den1_null_attributed, c_den2_constitution_delete_whole,
|
|
c_den2_skill_delete_multi, c_den2_belief_overwrite_whole,
|
|
c_den2_backed_up_zero, c_conf_backed_skill_overwrite,
|
|
c_conf_identity_append_attributed, c_conf_identity_read_whole,
|
|
c_conf_operating_delete_single, c_conf_fact_overwrite_whole,
|
|
c_conf_backed_truthy_string, c_conf_constitution_delete_backed,
|
|
c_conf_persona_overwrite_backed, c_conf_archive_delete_whole,
|
|
c_conf_unknown_kind_delete, c_allow_fact_append_single,
|
|
c_allow_fact_append_multi, c_allow_identity_read_single,
|
|
c_allow_archive_delete_single, c_allow_archive_read_whole,
|
|
c_allow_fact_read_whole, c_allow_unknown_operation,
|
|
c_allow_unknown_scope, c_allow_null_kind, c_allow_null_operation,
|
|
c_allow_null_scope,
|
|
]
|
|
|
|
GUARD = [
|
|
g_empty_dict_defaults, g_fact_append_defaults,
|
|
g_unattributed_identity, g_unbacked_constitution_delete,
|
|
g_backed_skill_confirms, g_null_attributed_present,
|
|
g_empty_list_attributed, g_backed_up_zero,
|
|
g_backed_up_truthy_string, g_read_allows, g_extra_keys_ignored,
|
|
g_null_target_kind, g_empty_operation_string,
|
|
]
|
|
|
|
GROUPS = {
|
|
"classify": (CLASSIFY, True, "mu_classify", facade_classify,
|
|
kernel_classify),
|
|
"guard": (GUARD, True, "mu_guard", facade_guard, kernel_guard),
|
|
# the loader probe is facade-only (its kernel counterpart is the
|
|
# in-crate _mg_core fn, not an FFI symbol — replaying it via the
|
|
# loader would be circular)
|
|
"core_probe": ([p_python_pin, p_auto, p_unset], True, None, None, None),
|
|
}
|
|
|
|
|
|
def canon(x) -> str:
|
|
"""Content canonicalization (key-order independent)."""
|
|
return json.dumps(json.loads(json.dumps(x, default=str)), sort_keys=True)
|
|
|
|
|
|
def main():
|
|
frozen = None
|
|
if REPLAY:
|
|
with open(TRUTH_PATH) as f:
|
|
frozen = json.load(f)["cases"]
|
|
|
|
groups = {}
|
|
for group, (cases, facade_replayable, _sym, fcall, kcall) in GROUPS.items():
|
|
recs = []
|
|
|
|
def rec(name, inp, oracle_ret, expected, _r=recs, _g=group):
|
|
entry = {"name": name, "input": inp,
|
|
"return": oracle_ret, "expected": expected}
|
|
if REPLAY and facade_replayable:
|
|
old = next(c for c in frozen[_g] if c["name"] == name)
|
|
for key in ("input", "return", "expected"):
|
|
got, want = entry[key], old[key]
|
|
if canon(got) != canon(want):
|
|
print(f"FACADE DIVERGE [{_g}] {name} {key}")
|
|
print(" got: ", json.dumps(got, sort_keys=True, default=str)[:600])
|
|
print(" want:", json.dumps(want, sort_keys=True, default=str)[:600])
|
|
_r.append({"name": name, "diverged": key})
|
|
return
|
|
_r.append(entry)
|
|
|
|
for c in cases:
|
|
c(rec)
|
|
groups[group] = recs
|
|
|
|
total = sum(len(v) for v in groups.values())
|
|
if REPLAY:
|
|
facade_groups = [g for g, (_, r, _, _, _) in GROUPS.items() if r]
|
|
facade_total = sum(len(groups[g]) for g in facade_groups)
|
|
facade_failures = sum(1 for g in facade_groups for c in groups[g]
|
|
if "diverged" in c)
|
|
|
|
failures = 0
|
|
checked = 0
|
|
for group, cases in frozen.items():
|
|
sym_fns = {"classify": kernel_classify, "guard": kernel_guard}
|
|
if group not in sym_fns:
|
|
continue # core_probe: metadata group, facade-replayed above
|
|
for case in cases:
|
|
checked += 1
|
|
got = sym_fns[group](case["input"])
|
|
want = case["expected"]
|
|
if canon(got) != canon(want):
|
|
failures += 1
|
|
print(f"KERNEL DIVERGE [{group}] {case['name']}")
|
|
print(" got: ", json.dumps(got, sort_keys=True, default=str)[:600])
|
|
print(" want:", json.dumps(want, sort_keys=True, default=str)[:600])
|
|
print(f"replay: facade {facade_total - facade_failures}/{facade_total}, "
|
|
f"kernel {checked - failures}/{checked}")
|
|
return 1 if (failures or facade_failures) else 0
|
|
|
|
doc = {
|
|
"frozen": datetime.now(timezone.utc).isoformat(),
|
|
"oracle": "src/rune/mutation_guards.py",
|
|
"pin": "RUNE_MG_CORE=python",
|
|
"note": "Frozen via scripts/gen_mutation_guards_truth.py against "
|
|
"the live Python oracle. Battery embeds this VERBATIM.",
|
|
"cases": groups,
|
|
}
|
|
os.makedirs(os.path.dirname(TRUTH_PATH), exist_ok=True)
|
|
with open(TRUTH_PATH, "w") as f:
|
|
json.dump(doc, f, indent=1)
|
|
f.write("\n")
|
|
print(f"frozen {total} cases → {os.path.relpath(TRUTH_PATH, REPO)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|