Files
rune/scripts/gen_self_modify_truth.py
Nick 1af27dc5d4 feat(self_modify): port to FULL Rust — dichotomy wave
Full name-preserving port of src/rune/self_modify.py to
src/rust/self_modify (RUST verdict, #2479 dichotomy tranche — 7/7 fns
incl. the _run subprocess seam, dichotomy audit match clean):
- _run: the subprocess seam is kernel-owned — Command spawn with piped
  reader threads, poll-based timeout→kill (TimeoutExpired raise-parity
  via the envelope), strict-UTF-8 text=True, universal-newline
  translation, signal exit codes
- propose_change: rev-parse gate ('unknown base' branch), mkdtemp
  0700 'selfmod-' + 8 chars of Python's tempfile alphabet, worktree
  add with err.strip()[:200] char-parity failure branch, no-clock `at` stamp as an argument
- run_gates: default pytest argv pinned, falsy/empty/null cmd falls
  back, (stdout+stderr)[-2000:] CHARACTER tail parity
- compare_baseline: full Python truthiness lattice on passed (missing
  key, null, 0/1 coercion), the three verdict reasons verbatim
- promote_or_discard: lazy rollback_ref read (KeyError fires only
  after merge/remove side effects, like the oracle), merge-fail
  err[:200] (no strip — oracle quirk preserved), discard ignores both
  git failures
- rollback: bad-ref err[:200] verbatim; monitor: .get(key, "")
  passes ANY present value through verbatim (int/null included)
- raise parity at the TYPE level: smod_loader._raise_like_oracle
  re-raises FileNotFoundError/PermissionError/TimeoutExpired/
  KeyError/TypeError from the envelope err encodings
- smod_loader mirrors broker_loader (doc-envelope FFI, cs_free_string,
  SMOD_RUST_LIB override; RUNE_SM_CORE belongs to shard_manager so the
  pin is RUNE_SMOD_CORE); facade delegation with python-oracle
  fallback preserved
- 32 frozen truth rows vs the LIVE oracle (clock frozen, tempfile
  pinned, deterministic git fixtures with literal seed SHA; setup as
  data + PATH-stubbed default gate command), deterministic across two
  generator runs; battery replays VERBATIM (cargo test 8/8) and live
  replay is 32/32 at BOTH facade and kernel level; pytest 6/6 both
  through the Rust core and with RUNE_SMOD_CORE=python
2026-10-12 02:50:56 +01:00

523 lines
21 KiB
Python

#!/usr/bin/env python3
"""gen_self_modify_truth.py — freeze/replay parity truth for self_modify
(dichotomy wave, #2479 tranche).
Freeze mode (default): exercises EVERY public function of the LIVE Python
oracle (src/rune/self_modify.py) across every branch and records
(kernel_input, expected_kernel_output) pairs into
src/rust/self_modify/src/frozen_smod_truth.json.
Replay mode (--replay): re-runs the SAME scenarios with the Rust core
delegation active and asserts both kernel-level (loader direct) and
facade-level (oracle functions delegate) parity against the frozen file.
Poisoned-truth law: freeze pins RUNE_SMOD_CORE=python BEFORE any import
(the RUNE_SM_CORE pin already belongs to shard_manager's core on master).
No-clock law: the wall clock is frozen by patching the `time` symbol in
self_modify, so every propose record carries the frozen `at` stamp and
replay is deterministic regardless of run date. The oracle's internal
tempfile.mkdtemp is likewise pinned to a case-local path.
Deterministic-fixture law: git repos are built with fixed identity and
fixed author/committer dates, so the seed commit SHA is identical across
generator runs, replay runs and the Rust battery — the SHA is frozen
LITERALLY, no placeholder. Machine-specific paths (repo dir, worktree dir,
stub dir) are frozen as @REPO@ / @WORKTREE@ / @STUBDIR@ placeholders and
substituted at replay time. Git-dependent scenarios declare their setup as
data ("worktree:<branch>", "commit_in_worktree", "dirty_repo", plus the
"stub" flag for the PATH-stubbed default gate command) so the generator,
the Python replay and the Rust battery run the SAME recipe.
Raise-parity rows freeze {"raises": "<oracle exception>"}: the Python
facade must raise the same exception; the Rust core (envelope law) must
surface an error (the loader raises RuntimeError — documented divergence
in exception TYPE only).
"""
import copy
import json
import os
import stat
import subprocess
import sys
import tempfile
import time
from datetime import datetime, timezone
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TRUTH_PATH = os.path.join(
REPO, "src", "rust", "self_modify", "src", "frozen_smod_truth.json"
)
REPLAY = "--replay" in sys.argv
os.environ["RUNE_SMOD_CORE"] = "python" # poisoned-truth law (freeze mode)
os.environ["RUNE_MEMORY_CORE"] = "python"
sys.path.insert(0, os.path.join(REPO, "src"))
sys.path.insert(0, os.path.join(REPO, "src", "rune"))
import self_modify as oracle_mod # noqa: E402
if REPLAY:
del os.environ["RUNE_SMOD_CORE"] # facade must delegate to the Rust core
import smod_loader # noqa: E402 (kernel-level replay side)
FROZEN_AT = "2026-10-11T12:00:00Z"
FROZEN_EPOCH = int(datetime(2026, 10, 11, 12, 0, 0, tzinfo=timezone.utc).timestamp())
GIT_ENV_FIXED = {
"GIT_AUTHOR_DATE": "2026-10-11T12:00:00+00:00",
"GIT_COMMITTER_DATE": "2026-10-11T12:00:00+00:00",
}
MISSING_REPO = "/nonexistent-selfmod-truth/repo"
class FrozenTime:
"""time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) → FROZEN_AT."""
@staticmethod
def gmtime():
return time.gmtime(FROZEN_EPOCH)
@staticmethod
def strftime(fmt, t=None):
return time.strftime(fmt, t if t is not None else FrozenTime.gmtime())
oracle_mod.time = FrozenTime
class FrozenTempfile:
"""tempfile.mkdtemp(prefix="selfmod-") → a case-local, known path."""
ROOT = None
N = 0
@staticmethod
def mkdtemp(prefix=""):
FrozenTempfile.N += 1
p = os.path.join(FrozenTempfile.ROOT, f"{prefix}{FrozenTempfile.N:02d}")
os.makedirs(p)
return p
oracle_mod.tempfile = FrozenTempfile
# ── deterministic fixture + setup interpreter (shared recipe) ───────────────
def _git(root, *argv):
env = dict(os.environ, **GIT_ENV_FIXED)
return subprocess.run(["git", "-C", root, *argv], capture_output=True,
text=True, env=env)
def make_fixture(case_root):
"""The battery recipe — fixed identity, fixed dates, seed commit.
The resulting master SHA is machine-independent (frozen literally)."""
repo = os.path.join(case_root, "repo")
os.makedirs(repo)
_git(repo, "init", "-q", "-b", "master")
_git(repo, "config", "user.email", "t@t")
_git(repo, "config", "user.name", "t")
_git(repo, "config", "commit.gpgsign", "false")
with open(os.path.join(repo, "f.txt"), "w") as f:
f.write("v1\n")
_git(repo, "add", ".")
_git(repo, "commit", "-qm", "seed")
return repo
def base_sha(repo):
return _git(repo, "rev-parse", "master").stdout.strip()
def run_setup(repo, steps):
"""Declarative setup ops, executed identically by freeze, replay and
the Rust battery. Returns the worktree path (or None)."""
wt = None
for step in steps:
op, _, arg = step.partition(":")
if op == "worktree":
wt = os.path.join(os.path.dirname(repo), f"wt-{arg}")
r = _git(repo, "worktree", "add", wt, "-b", arg, "master")
assert r.returncode == 0, r.stderr
elif op == "commit_in_worktree":
with open(os.path.join(wt, "f.txt"), "w") as f:
f.write("v2\n")
r = _git(wt, "commit", "-qam", "c")
assert r.returncode == 0, r.stderr
elif op == "dirty_repo":
with open(os.path.join(repo, "f.txt"), "w") as f:
f.write("v9\n")
else:
raise ValueError(f"unknown setup op: {step}")
return wt
def make_stub_dir(case_root):
"""PATH stub: a python3 that deterministically 'runs the gate suite'
(the oracle's default gate command resolves through PATH)."""
stub = os.path.join(case_root, "stub")
os.makedirs(stub)
path = os.path.join(stub, "python3")
with open(path, "w") as f:
f.write("#!/bin/sh\necho stub-gates-green\nexit 0\n")
os.chmod(path, os.stat(path).st_mode | stat.S_IXUSR)
return stub
# ── scenario cases ───────────────────────────────────────────────────────────
class Case:
def __init__(self, name, group, steps, inp, stub=False):
self.name = name
self.group = group
self.steps = steps
self.inp = inp # (repo, wt, stub) → kernel input dict
self.stub = stub
def propose_cases():
return [
Case("propose-ok", "propose_change", [],
lambda r, w, s: {"repo": "@REPO@", "branch": "cand-ok",
"base": "master", "now": FROZEN_AT}),
Case("propose-base-head", "propose_change", [],
lambda r, w, s: {"repo": "@REPO@", "branch": "cand-head",
"base": "HEAD", "now": FROZEN_AT}),
Case("propose-unknown-base", "propose_change", [],
lambda r, w, s: {"repo": "@REPO@", "branch": "cand-bad",
"base": "no-such-ref", "now": FROZEN_AT}),
Case("propose-dup-branch", "propose_change", ["worktree:dup"],
lambda r, w, s: {"repo": "@REPO@", "branch": "dup",
"base": "master", "now": FROZEN_AT}),
Case("propose-missing-repo", "propose_change", [],
lambda r, w, s: {"repo": MISSING_REPO, "branch": "cand-gone",
"base": "master", "now": FROZEN_AT}),
]
def gates_cases():
return [
Case("gates-pass", "run_gates", [],
lambda r, w, s: {"worktree": "@REPO@",
"test_cmd": ["echo", "gates-green"]}),
Case("gates-fail-stderr", "run_gates", [],
lambda r, w, s: {"worktree": "@REPO@",
"test_cmd": ["sh", "-c", "echo boom 1>&2; exit 1"]}),
Case("gates-tail-2000", "run_gates", [],
lambda r, w, s: {"worktree": "@REPO@",
"test_cmd": ["sh", "-c", "seq 1 3000"]}),
Case("gates-out-then-err", "run_gates", [],
lambda r, w, s: {"worktree": "@REPO@",
"test_cmd": ["sh", "-c",
"echo o1; echo e1 1>&2; echo o2; exit 7"]}),
Case("gates-default-cmd", "run_gates", [],
lambda r, w, s: {"worktree": "@REPO@", "test_cmd": None},
stub=True),
]
def compare_cases():
def row(name, cand, base):
return Case("cb-" + name, "compare_baseline", [],
lambda r, w, s, c=cand, b=base:
{"candidate_result": c, "baseline_result": b})
return [
row("candidate-red", {"passed": False}, {"passed": True}),
row("candidate-red-baseline-red", {"passed": False}, {"passed": False}),
row("baseline-red", {"passed": True}, {"passed": False}),
row("both-green", {"passed": True}, {"passed": True}),
row("missing-passed-key", {}, {"passed": True}),
row("truthy-coercion", {"passed": 1}, {"passed": 0}),
row("explicit-null-passed", {"passed": None}, {"passed": True}),
]
def _proposal(branch, repo):
return {"proposed": True, "branch": branch, "worktree": "@WORKTREE@",
"base": "master", "rollback_ref": base_sha(repo), "at": FROZEN_AT}
def pod_cases():
return [
Case("pod-discard", "promote_or_discard", ["worktree:discard-1"],
lambda r, w, s: {"repo": "@REPO@", "proposal": _proposal("discard-1", r),
"verdict": {"promote": False, "reason": "test"}}),
Case("pod-promote-up-to-date", "promote_or_discard", ["worktree:podd-2"],
lambda r, w, s: {"repo": "@REPO@", "proposal": _proposal("podd-2", r),
"verdict": {"promote": True, "reason": "test"}}),
Case("pod-promote-real-merge", "promote_or_discard",
["worktree:podd-3", "commit_in_worktree"],
lambda r, w, s: {"repo": "@REPO@", "proposal": _proposal("podd-3", r),
"verdict": {"promote": True, "reason": "test"}}),
Case("pod-merge-fail", "promote_or_discard",
["worktree:podd-4", "commit_in_worktree", "dirty_repo"],
lambda r, w, s: {"repo": "@REPO@", "proposal": _proposal("podd-4", r),
"verdict": {"promote": True, "reason": "test"}}),
Case("pod-discard-ghost", "promote_or_discard", [],
lambda r, w, s: {"repo": "@REPO@", "proposal": _proposal("ghost-branch", r),
"verdict": {"promote": False, "reason": "test"}}),
Case("pod-missing-branch-key", "promote_or_discard", [],
lambda r, w, s: {"repo": "@REPO@",
"proposal": {k: v for k, v in _proposal("keyless", r).items()
if k != "branch"},
"verdict": {"promote": True}}),
]
def rollback_cases():
return [
Case("rb-ok-after-dirty", "rollback", ["dirty_repo"],
lambda r, w, s: {"repo": "@REPO@", "rollback_ref": base_sha(r)}),
Case("rb-ref-master", "rollback", ["dirty_repo"],
lambda r, w, s: {"repo": "@REPO@", "rollback_ref": "master"}),
Case("rb-bad-ref", "rollback", [],
lambda r, w, s: {"repo": "@REPO@", "rollback_ref": "no-such-ref"}),
Case("rb-missing-repo", "rollback", [],
lambda r, w, s: {"repo": MISSING_REPO, "rollback_ref": "master"}),
]
def monitor_cases():
def row(name, proposal):
return Case("mon-" + name, "monitor", [],
lambda r, w, s, p=proposal: {"proposal": p})
return [
row("full", {"branch": "b", "rollback_ref": "abc123"}),
row("empty", {}),
row("partial", {"branch": "only-b"}),
row("extra-keys", {"branch": "b", "rollback_ref": "r", "extra": 1}),
row("nonstring-branch", {"branch": 7, "rollback_ref": None}),
]
GROUPS = [
("propose_change", propose_cases()),
("run_gates", gates_cases()),
("compare_baseline", compare_cases()),
("promote_or_discard", pod_cases()),
("rollback", rollback_cases()),
("monitor", monitor_cases()),
]
FACADE_CALL = {
"propose_change": lambda i: oracle_mod.propose_change(
i["repo"], i["branch"], i["base"]),
"run_gates": lambda i: oracle_mod.run_gates(i["worktree"], i["test_cmd"]),
"compare_baseline": lambda i: oracle_mod.compare_baseline(
i["candidate_result"], i["baseline_result"]),
"promote_or_discard": lambda i: oracle_mod.promote_or_discard(
i["repo"], i["proposal"], i["verdict"]),
"rollback": lambda i: oracle_mod.rollback(i["repo"], i["rollback_ref"]),
"monitor": lambda i: oracle_mod.monitor(i["proposal"]),
}
KERNEL_CALL = {
"propose_change": lambda i: smod_loader.propose_change(
i["repo"], i["branch"], i["base"], i["now"]),
"run_gates": lambda i: smod_loader.run_gates(i["worktree"], i["test_cmd"]),
"compare_baseline": lambda i: smod_loader.compare_baseline(
i["candidate_result"], i["baseline_result"]),
"promote_or_discard": lambda i: smod_loader.promote_or_discard(
i["repo"], i["proposal"], i["verdict"]),
"rollback": lambda i: smod_loader.rollback(i["repo"], i["rollback_ref"]),
"monitor": lambda i: smod_loader.monitor(i["proposal"]),
}
# ── placeholder plumbing ─────────────────────────────────────────────────────
def substitute(value, repo, wt, stub):
"""Machine paths → frozen placeholders (kernel input side)."""
s = json.dumps(value)
for actual, ph in ((repo, "@REPO@"), (wt, "@WORKTREE@"), (stub, "@STUBDIR@")):
if actual:
s = s.replace(actual, ph)
return json.loads(s)
def de_substitute(value, repo, wt, stub):
"""Frozen placeholders → live paths (call side)."""
s = json.dumps(value)
for actual, ph in ((repo, "@REPO@"), (wt, "@WORKTREE@"), (stub, "@STUBDIR@")):
if actual:
s = s.replace(ph, actual)
return json.loads(s)
def normalize(out, repo, wt, stub):
"""Actual output paths → placeholders (kernel output side). The
worktree a propose call created internally is captured from the
output itself — neither side can know it in advance."""
reps = [t for t in ((wt, "@WORKTREE@") if wt else None,
(repo, "@REPO@") if repo else None,
(stub, "@STUBDIR@") if stub else None) if t]
if isinstance(out, dict) and isinstance(out.get("worktree"), str):
reps.append((out["worktree"], "@WORKTREE@"))
s = json.dumps(out, sort_keys=True)
for actual, ph in reps:
if actual and actual != "/":
s = s.replace(actual, ph)
return json.loads(s)
def _catch(fn, *a):
try:
return fn(*a)
except FileNotFoundError:
return {"__raised__": "FileNotFoundError"}
except KeyError as e:
return {"__raised__": f"KeyError: {e}"}
except RuntimeError:
return {"__raised__": "RuntimeError"}
class _PathEnv:
"""Set PATH to the stub dir for the duration of one call."""
def __init__(self, stub):
self.stub = stub
def __enter__(self):
self.old = os.environ.get("PATH")
os.environ["PATH"] = self.stub + os.pathsep + (self.old or "")
def __exit__(self, *exc):
if self.old is not None:
os.environ["PATH"] = self.old
return False
def run_scenario(case, tmp_root, call):
"""One scenario, identical for freeze and both replay tiers:
fresh fixture → setup → substituted input → call → normalized output."""
case_root = tempfile.mkdtemp(prefix=f"smod_{case.name}_", dir=tmp_root)
repo = make_fixture(case_root)
FrozenTempfile.ROOT = case_root
FrozenTempfile.N = 0
wt = run_setup(repo, case.steps)
stub = make_stub_dir(case_root) if case.stub else None
live_wt = wt or os.path.join(case_root, "wt-input")
live_stub = stub or os.path.join(case_root, "stub-unused")
inp = substitute(case.inp(repo, wt, stub), repo, wt, stub)
inp_copy = copy.deepcopy(inp) # deep-copy law: inputs frozen before calls
live = de_substitute(inp_copy, repo, live_wt, live_stub)
if case.stub:
with _PathEnv(live_stub):
out = _catch(call, live)
else:
out = _catch(call, live)
if isinstance(out, dict) and out.get("__raised__"):
return inp, {"raises": out["__raised__"]}
return inp, normalize(out, repo, live_wt, stub)
def main():
frozen = None
if REPLAY:
with open(TRUTH_PATH) as f:
frozen = json.load(f)["cases"]
tmp_root = tempfile.mkdtemp(prefix="smod_truth_")
groups = {}
failures = 0
total = 0
for group, cases in GROUPS:
recs = []
def rec(name, inp, expected, case=None, _r=recs, _g=group):
entry = {"name": name, "input": inp, "expected": expected}
if case is not None:
entry["setup"] = case.steps
entry["stub"] = case.stub
if REPLAY:
old = next(c for c in frozen[_g] if c["name"] == name)
for key in ("input", "expected"):
if json.dumps(entry[key], sort_keys=True) != json.dumps(
old[key], sort_keys=True):
print(f"FACADE DIVERGE [{_g}] {name} {key}")
print(" got: ", json.dumps(entry[key], sort_keys=True)[:600])
print(" want:", json.dumps(old[key], sort_keys=True)[:600])
_r.append({"name": name, "diverged": key})
return
_r.append(entry)
for case in cases:
total += 1
inp, want = run_scenario(case, tmp_root,
lambda i, c=case: _catch(
FACADE_CALL[c.group], i))
rec(case.name, inp, want, case)
if REPLAY:
old = next(c for c in frozen[case.group] if c["name"] == case.name)
repo, wt, stub = kernel_paths(case, tmp_root, old)
call = lambda: KERNEL_CALL[case.group](
de_substitute(copy.deepcopy(old["input"]), repo, wt, stub))
if old.get("stub"):
with _PathEnv(stub):
kout = _catch(lambda: call())
else:
kout = _catch(call)
kwant = old["expected"]
if not (isinstance(kout, dict) and kout.get("__raised__")):
kout = normalize(kout, repo, wt, stub)
if isinstance(kwant, dict) and kwant.get("raises"):
ok = kout == {"__raised__": kwant["raises"]} or (
kout == {"__raised__": "RuntimeError"})
else:
ok = json.dumps(kout, sort_keys=True) == json.dumps(
kwant, sort_keys=True)
if not ok:
failures += 1
print(f"KERNEL DIVERGE [{case.group}] {case.name}")
print(" got: ", json.dumps(kout, sort_keys=True)[:600])
print(" want:", json.dumps(kwant, sort_keys=True)[:600])
groups[group] = recs
if REPLAY:
facade_div = sum(1 for v in groups.values() for c in v if "diverged" in c)
print(f"replay: facade {total - facade_div}/{total}, "
f"kernel {total - failures}/{total}")
return 1 if (failures or facade_div) else 0
doc = {
"frozen": datetime.now(timezone.utc).isoformat(),
"oracle": "src/rune/self_modify.py",
"pin": "RUNE_SMOD_CORE=python",
"note": "Frozen via scripts/gen_self_modify_truth.py against the live "
"Python oracle; wall clock frozen at " + FROZEN_AT + "; git "
"fixtures use fixed identity+dates so the seed SHA is "
"literal. @REPO@/@WORKTREE@/@STUBDIR@ are placeholders "
"substituted at replay. 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, sort_keys=True)
f.write("\n")
print(f"frozen {total} cases → {os.path.relpath(TRUTH_PATH, REPO)}")
return 0
def kernel_paths(case, tmp_root, old_entry):
"""(repo, wt, stub) for kernel-level replay of one frozen entry."""
case_root = tempfile.mkdtemp(prefix=f"smod_kernel_{case.name}_", dir=tmp_root)
repo = make_fixture(case_root)
wt = run_setup(repo, old_entry.get("setup", []))
stub = None
if old_entry.get("stub"):
stub = make_stub_dir(case_root)
live_wt = wt or os.path.join(case_root, "wt-input")
live_stub = stub or os.path.join(case_root, "stub-unused")
return repo, live_wt, live_stub
if __name__ == "__main__":
sys.exit(main())