Nick
00839d1af7
Port the 11 audit-missing functions of src/rune/structural_model.py to the Rust crate (name-preserving, logic-1:1): workspace_record, workspace_advance, _init_project, save, reload, add, relate, state_owned, verification_surfaces, canonical_state_owner, supersession — plus a true deployment_surfaces (the old deployment_surfaces_for stub is gone, no callers), the lowercase-stale filters on dependencies/consumers, and the full §18F impact_closure shape the W6-era port had reduced. Fix-first: the delegation seam imported rune.sm_loader — the SHARD_MANAGER loader (no query symbol), so every kernel call died into the Python floor. The seam now targets stm_loader (the crate's sm_* FFI, RUNE_STM_CORE=python pins the floor). workspace_advance keeps the oracle's in-place law. Verification (verification/structural_model-summary.md): frozen truth 70 cases byte-identical (gen_structural_model_truth.py --verify), battery 16/16 with truth embedded via include_str!, live parity 4620/4620 (frozen replay + 300-doc seeded fuzz + dual-mode facade), clippy -D warnings clean, Python suites green in both pins (pre-existing 2F/12E verified identical on the stashed base). Audit: 9/20 PARTIAL -> 21/21 FULL.
489 lines
18 KiB
Python
489 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""Live parity: structural_model Rust kernels vs Python oracle (dichotomy
|
|
wave, issue #2479 tranche). Layers:
|
|
|
|
1. FROZEN — the frozen truth corpus (frozen_sm_truth.json, generated by
|
|
scripts/gen_structural_model_truth.py against the LIVE oracle with
|
|
RUNE_STM_CORE=python + RUNE_STRUCTURAL_CORE=python pinned) replayed
|
|
through the ctypes FFI, case by case, deep-copied inputs; save() /
|
|
_init_project() compared BYTE-EXACT on real files.
|
|
2. FUZZ — 300 seeded random models (seed 2479): oracle floor vs FFI
|
|
on every kernel, exact compare; save() byte-exact.
|
|
3. LIVE — dual-mode facade battery: the same facade calls executed
|
|
with RUNE_STM_CORE=python (oracle floor) and with the Rust core
|
|
live; results must be identical (clock pinned).
|
|
|
|
Exit 0 only at 100% parity.
|
|
"""
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import random
|
|
import string
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
os.environ["RUNE_STM_CORE"] = "python" # oracle floor for compare
|
|
os.environ["RUNE_STRUCTURAL_CORE"] = "python" # the floor's store core
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO / "src"))
|
|
|
|
from rune import structural_store as ss # noqa: E402
|
|
from rune import structural_model as sm # noqa: E402
|
|
from rune import stm_loader # noqa: E402
|
|
|
|
FIXED_NOW = "2026-10-13T01:02:03Z"
|
|
ss.now = lambda: FIXED_NOW # both sides stamp with the pinned clock
|
|
|
|
TRUTH = json.loads((REPO / "src" / "rust" / "structural_model" / "src" / "frozen_sm_truth.json").read_text())
|
|
|
|
fails: list[str] = []
|
|
counts = {"frozen": 0, "fuzz": 0, "live": 0}
|
|
|
|
|
|
def j(x):
|
|
return json.dumps(x, sort_keys=True, ensure_ascii=True)
|
|
|
|
|
|
def fail(layer, name, detail):
|
|
fails.append(f"[{layer}] {name}: {detail}")
|
|
print(f"FAIL[{layer}] {name}: {detail[:240]}")
|
|
|
|
|
|
def check(layer, name, want, got):
|
|
counts[layer] += 1
|
|
if j(want) != j(got):
|
|
fail(layer, name, f"want={j(want)[:280]} got={j(got)[:280]}")
|
|
|
|
|
|
def receiver(doc, project_id="proj", path="/tmp/sm-parity/structural.json"):
|
|
m = sm.StructuralModel.__new__(sm.StructuralModel)
|
|
m.store_dir = None
|
|
m.project_id = project_id
|
|
m.path = path
|
|
m.doc = doc
|
|
return m
|
|
|
|
|
|
def oracle_call(fn, a):
|
|
"""The oracle floor (delegation pinned off) for one frozen case."""
|
|
if fn in {
|
|
"children",
|
|
"beneath",
|
|
"owning_domain",
|
|
"dependencies",
|
|
"consumers",
|
|
"interfaces_for",
|
|
"state_owned",
|
|
"verification_surfaces",
|
|
"deployment_surfaces",
|
|
"canonical_state_owner",
|
|
"supersession",
|
|
"impact_closure",
|
|
}:
|
|
m = receiver(copy.deepcopy(a["doc"]))
|
|
if fn == "impact_closure":
|
|
return m.impact_closure(a["eid"], max_depth=a.get("max_depth", 8))
|
|
return getattr(m, fn)(a["eid"])
|
|
if fn == "add":
|
|
m = receiver(copy.deepcopy(a["doc"]))
|
|
return {
|
|
"doc": m.doc,
|
|
"eid": m.add(
|
|
a["kind"],
|
|
a["name"],
|
|
parent=a.get("parent"),
|
|
ext=a.get("ext"),
|
|
evidence_ref=a.get("evidence_ref", "structural-model"),
|
|
),
|
|
}
|
|
if fn == "relate":
|
|
m = receiver(copy.deepcopy(a["doc"]))
|
|
return {
|
|
"doc": m.doc,
|
|
"rel": m.relate(
|
|
a["from_id"],
|
|
a["rel"],
|
|
a["to_id"],
|
|
evidence_ref=a.get("evidence_ref", "structural-model"),
|
|
authority=a.get("authority", "model-author"),
|
|
status=a.get("status", "active"),
|
|
ext=a.get("ext"),
|
|
),
|
|
}
|
|
if fn == "workspace_record":
|
|
return sm.workspace_record(
|
|
contract_id=a["contract_id"],
|
|
construction_base=a["construction_base"],
|
|
runtime_generation=a.get("runtime_generation", ""),
|
|
candidate_generation=a.get("candidate_generation", ""),
|
|
promotion_target=a.get("promotion_target", "gitea/master"),
|
|
rollback_generation=a.get("rollback_generation", ""),
|
|
)
|
|
if fn == "workspace_advance":
|
|
record = copy.deepcopy(a["record"])
|
|
return sm.workspace_advance(record, a["stage"], evidence=a.get("evidence", ""))
|
|
raise AssertionError(fn)
|
|
|
|
|
|
def ffi_call(fn, a):
|
|
if fn in {
|
|
"children",
|
|
"beneath",
|
|
"owning_domain",
|
|
"dependencies",
|
|
"consumers",
|
|
"interfaces_for",
|
|
"state_owned",
|
|
"verification_surfaces",
|
|
"deployment_surfaces",
|
|
"canonical_state_owner",
|
|
"supersession",
|
|
"impact_closure",
|
|
}:
|
|
return stm_loader.query(copy.deepcopy(a["doc"]), a["eid"], fn, max_depth=a.get("max_depth"))
|
|
if fn == "add":
|
|
return stm_loader.add(
|
|
copy.deepcopy(a["doc"]),
|
|
a["kind"],
|
|
a["name"],
|
|
parent=a.get("parent"),
|
|
ext=a.get("ext"),
|
|
evidence_ref=a.get("evidence_ref", "structural-model"),
|
|
now=FIXED_NOW,
|
|
)
|
|
if fn == "relate":
|
|
return stm_loader.relate(
|
|
copy.deepcopy(a["doc"]),
|
|
a["from_id"],
|
|
a["rel"],
|
|
a["to_id"],
|
|
evidence_ref=a.get("evidence_ref", "structural-model"),
|
|
authority=a.get("authority", "model-author"),
|
|
status=a.get("status", "active"),
|
|
ext=a.get("ext"),
|
|
now=FIXED_NOW,
|
|
)
|
|
if fn == "workspace_record":
|
|
return stm_loader.workspace_record(
|
|
contract_id=a["contract_id"],
|
|
construction_base=a["construction_base"],
|
|
runtime_generation=a.get("runtime_generation", ""),
|
|
candidate_generation=a.get("candidate_generation", ""),
|
|
promotion_target=a.get("promotion_target", "gitea/master"),
|
|
rollback_generation=a.get("rollback_generation", ""),
|
|
)
|
|
if fn == "workspace_advance":
|
|
return stm_loader.workspace_advance(copy.deepcopy(a["record"]), a["stage"], evidence=a.get("evidence", ""))
|
|
raise AssertionError(fn)
|
|
|
|
|
|
def call_side(call):
|
|
"""Run a case body; errors become comparable markers."""
|
|
try:
|
|
return {"__ok__": call()}
|
|
except (ValueError, KeyError) as exc:
|
|
return {"__err__": f"{type(exc).__name__}: {exc}"}
|
|
|
|
|
|
def frozen():
|
|
for fn, cases in TRUTH["groups"].items():
|
|
if fn in {"save", "reload", "init_project"}:
|
|
continue # byte-law layer below
|
|
for c in cases:
|
|
a = copy.deepcopy(c["args"])
|
|
want = call_side(lambda: oracle_call(fn, a))
|
|
got = call_side(lambda: ffi_call(fn, a))
|
|
check("frozen", f"{fn}/{c['name']}", want, got)
|
|
|
|
|
|
def byte_law(tmp):
|
|
"""save / _init_project / reload must be byte-exact through the FFI."""
|
|
for c in TRUTH["groups"]["save"]:
|
|
name = c["name"]
|
|
m = receiver(copy.deepcopy(c["args"]["doc"]), path=tmp / f"py/{name}.json")
|
|
m.save()
|
|
want = {"content": Path(m.path).read_text(), "doc": m.doc}
|
|
out = stm_loader.save(copy.deepcopy(c["args"]["doc"]), str(tmp / f"rs/{name}.json"))
|
|
got = {"content": Path(out["path"]).read_text(), "doc": out["doc"]}
|
|
check("frozen", f"save-bytes/{name}", want, got)
|
|
c = TRUTH["groups"]["init_project"][0]
|
|
m = sm.StructuralModel(tmp / "py-init", c["args"]["project_id"])
|
|
want = {"content": Path(m.path).read_text(), "doc": m.doc}
|
|
out = stm_loader.init_project(
|
|
copy.deepcopy(c["args"]["doc"]), c["args"]["project_id"], str(tmp / "rs-init/structural.json"), FIXED_NOW
|
|
)
|
|
got = {"content": Path(out["path"]).read_text(), "doc": out["doc"]}
|
|
check("frozen", "init_project-bytes", want, got)
|
|
# reload: FFI doc equals the floor's reloaded doc, byte-exact source
|
|
m2 = sm.StructuralModel(tmp / "py-init", c["args"]["project_id"])
|
|
m2.reload()
|
|
got_doc = stm_loader.reload(str(tmp / "py-init/structural.json"))
|
|
check("frozen", "reload-roundtrip", m2.doc, got_doc)
|
|
check("frozen", "reload-missing-none", None, stm_loader.reload(str(tmp / "nope.json")))
|
|
|
|
|
|
# ── fuzz (seeded) ───────────────────────────────────────────────────────────
|
|
|
|
KINDS = list(sm.ENTITY_KINDS) + ["widget"]
|
|
RELS = list(sm.RELATION_TYPES) + ["hates"]
|
|
IDS = [
|
|
"project:p",
|
|
"domain:d",
|
|
"component:a",
|
|
"component:b",
|
|
"interface:i",
|
|
"state_store:s",
|
|
"test_suite:t",
|
|
"deployment_surface:ds",
|
|
"environment:e",
|
|
"component:old",
|
|
"é-unit",
|
|
"x" * 40,
|
|
"",
|
|
]
|
|
STATUSES = [None, None, "active", "stale", "STALE", "CURRENT", "root"]
|
|
|
|
|
|
def rand_rel(rnd):
|
|
r = {
|
|
"type": rnd.choice(RELS),
|
|
"from": rnd.choice(IDS),
|
|
"to": rnd.choice(IDS),
|
|
"evidence_refs": sorted({f"ev-{rnd.randint(0, 2)}" for _ in range(rnd.randint(0, 2))}),
|
|
"observed_at": FIXED_NOW,
|
|
}
|
|
st = rnd.choice(STATUSES)
|
|
if st:
|
|
r["status"] = st
|
|
if rnd.random() < 0.2:
|
|
r["authority"] = rnd.choice(["model-author", "importer", "user"])
|
|
if rnd.random() < 0.15:
|
|
r["ext"] = {"n": rnd.randint(0, 5)}
|
|
return r
|
|
|
|
|
|
def rand_doc(rnd):
|
|
d = {"schema_version": 1, "generated_at": FIXED_NOW, "entities": [], "relationships": [], "observations": []}
|
|
for _ in range(rnd.randint(0, 6)):
|
|
e = {
|
|
"id": rnd.choice(IDS) or f"component:f{rnd.randint(0, 8)}",
|
|
"kind": rnd.choice(sm.ENTITY_KINDS),
|
|
"maturity": rnd.choice(["DECLARED", "PRESENT", "WIRED", "VERIFIED"]),
|
|
"tags": [],
|
|
"identities": {},
|
|
"evidence_refs": sorted({f"ev-{rnd.randint(0, 2)}" for _ in range(rnd.randint(0, 2))}),
|
|
"observed_at": FIXED_NOW,
|
|
}
|
|
if rnd.random() < 0.25:
|
|
e["aliases"] = [f"component:old{rnd.randint(0, 3)}"]
|
|
if rnd.random() < 0.2:
|
|
e.pop("identities") # the KeyError crash-parity class
|
|
d["entities"].append(e)
|
|
for _ in range(rnd.randint(0, 8)):
|
|
d["relationships"].append(rand_rel(rnd))
|
|
return d
|
|
|
|
|
|
def fuzz(tmp):
|
|
rnd = random.Random(2479)
|
|
for i in range(300):
|
|
d = rand_doc(rnd)
|
|
eid = rnd.choice(IDS) or "component:a"
|
|
for fn in [
|
|
"children",
|
|
"beneath",
|
|
"owning_domain",
|
|
"dependencies",
|
|
"consumers",
|
|
"interfaces_for",
|
|
"state_owned",
|
|
"verification_surfaces",
|
|
"deployment_surfaces",
|
|
"canonical_state_owner",
|
|
"supersession",
|
|
]:
|
|
want = call_side(lambda: oracle_call(fn, {"doc": d, "eid": eid}))
|
|
got = call_side(lambda: ffi_call(fn, {"doc": d, "eid": eid}))
|
|
check("fuzz", f"{fn}/{i}", want, got)
|
|
depth = rnd.choice([1, 2, 8])
|
|
want = call_side(lambda: oracle_call("impact_closure", {"doc": d, "eid": eid, "max_depth": depth}))
|
|
got = call_side(lambda: ffi_call("impact_closure", {"doc": d, "eid": eid, "max_depth": depth}))
|
|
check("fuzz", f"impact_closure/{i}", want, got)
|
|
# mutations
|
|
kind, name = rnd.choice(KINDS), f"n{i}"
|
|
a = {
|
|
"doc": d,
|
|
"kind": kind,
|
|
"name": name,
|
|
"parent": rnd.choice([None, "domain:d"]),
|
|
"ext": rnd.choice([None, {"k": i}, {}]),
|
|
"evidence_ref": rnd.choice(["structural-model", "", f"ev-{i}"]),
|
|
}
|
|
want = call_side(lambda: oracle_call("add", a))
|
|
got = call_side(lambda: ffi_call("add", a))
|
|
check("fuzz", f"add/{i}", want, got)
|
|
ra = {
|
|
"doc": d,
|
|
"from_id": rnd.choice(IDS),
|
|
"rel": rnd.choice(RELS),
|
|
"to_id": rnd.choice(IDS),
|
|
"evidence_ref": rnd.choice(["structural-model", "ev"]),
|
|
"authority": rnd.choice(["model-author", "importer"]),
|
|
"status": rnd.choice(["active", "CURRENT"]),
|
|
"ext": rnd.choice([None, {"z": 1}, {}]),
|
|
}
|
|
want = call_side(lambda: oracle_call("relate", ra))
|
|
got = call_side(lambda: ffi_call("relate", ra))
|
|
check("fuzz", f"relate/{i}", want, got)
|
|
# lifecycle
|
|
stages = rnd.choice([[], ["base_verified"], ["base_verified", "worktree_created"]])
|
|
record = sm.workspace_record(contract_id=f"C{i}", construction_base=f"sha{i}")
|
|
for s in stages:
|
|
record = sm.workspace_advance(record, s)
|
|
stage = rnd.choice(sm.WORKSPACE_LIFECYCLE_STAGES + ("bogus",))
|
|
wa = {"record": record, "stage": stage, "evidence": rnd.choice(["", "e" * rnd.randint(1, 250)])}
|
|
want = call_side(lambda: oracle_call("workspace_advance", wa))
|
|
got = call_side(lambda: ffi_call("workspace_advance", wa))
|
|
check("fuzz", f"workspace_advance/{i}", want, got)
|
|
# save byte law on random docs
|
|
if i % 6 == 0:
|
|
m = receiver(copy.deepcopy(d), path=tmp / f"fz-py/{i}.json")
|
|
try:
|
|
m.save()
|
|
want_c = Path(m.path).read_text()
|
|
except ValueError as exc:
|
|
want_c = f"__err__ {exc}"
|
|
out = call_side(lambda: stm_loader.save(copy.deepcopy(d), str(tmp / f"fz-rs/{i}.json")))
|
|
if isinstance(out.get("__ok__"), dict):
|
|
got_c = Path(out["__ok__"]["path"]).read_text()
|
|
else:
|
|
got_c = f"__err__ {out.get('__err__', '')}"
|
|
counts["fuzz"] += 1
|
|
if want_c != got_c:
|
|
fail("fuzz", f"save-bytes/{i}", f"py={want_c[:120]!r} rs={got_c[:120]!r}")
|
|
|
|
|
|
# ── live dual-mode facade battery ───────────────────────────────────────────
|
|
|
|
FACADE_CODE = r"""
|
|
import copy, json, os, sys, tempfile, calendar, time as _time, types
|
|
sys.path.insert(0, "src")
|
|
from rune import structural_store as ss
|
|
FIXED = "2026-10-13T01:02:03Z"
|
|
ss.now = lambda: FIXED
|
|
from rune import structural_model as sm
|
|
out = {}
|
|
with tempfile.TemporaryDirectory() as td:
|
|
m = sm.StructuralModel(td + "/store", "scaffolding")
|
|
m.add("domain", "construction", parent="project:scaffolding")
|
|
m.add("subsystem", "kernel", parent="domain:construction")
|
|
m.add("component", "scaffold", parent="subsystem:kernel")
|
|
m.add("component", "beta", parent="subsystem:kernel")
|
|
m.add("interface", "api", parent="subsystem:kernel")
|
|
m.add("state_store", "st", parent="project:scaffolding",
|
|
ext={"engine": "sqlite"})
|
|
m.add("test_suite", "kt", parent="project:scaffolding")
|
|
m.add("deployment_surface", "live", parent="project:scaffolding")
|
|
try:
|
|
m.add("widget", "x")
|
|
out["bad_kind"] = "NO RAISE"
|
|
except ValueError as e:
|
|
out["bad_kind"] = str(e)
|
|
try:
|
|
m.relate("component:scaffold", "hates", "component:beta")
|
|
out["bad_rel"] = "NO RAISE"
|
|
except ValueError as e:
|
|
out["bad_rel"] = str(e)
|
|
m.relate("component:scaffold", "provides", "interface:api")
|
|
m.relate("component:beta", "consumes", "interface:api")
|
|
m.relate("component:beta", "depends_on", "component:scaffold")
|
|
m.relate("component:scaffold", "owns", "state_store:st")
|
|
m.relate("component:scaffold", "verified_by", "test_suite:kt")
|
|
m.relate("component:scaffold", "deployed_by", "deployment_surface:live")
|
|
m.relate("component:scaffold", "supersedes", "component:beta")
|
|
m.relate("component:beta", "legacy_of", "component:scaffold")
|
|
m.relate("component:scaffold", "reads", "state_store:st",
|
|
authority="importer", status="CURRENT", ext={"mode": "ro"})
|
|
m.save()
|
|
out["doc"] = m.doc
|
|
out["file_sha"] = __import__("hashlib").sha256(
|
|
open(td + "/store/structural.json", "rb").read()).hexdigest()
|
|
m2 = sm.StructuralModel(td + "/store", "scaffolding")
|
|
m2.reload()
|
|
out["reload_same"] = m2.doc == m.doc
|
|
eid = "component:scaffold"
|
|
for q in ["children", "beneath", "owning_domain", "dependencies",
|
|
"consumers", "interfaces_for", "state_owned",
|
|
"verification_surfaces", "deployment_surfaces",
|
|
"canonical_state_owner", "supersession"]:
|
|
out[q] = getattr(m, q)(eid)
|
|
out["closure"] = m.impact_closure(eid)
|
|
out["closure_root"] = m.impact_closure("project:scaffolding")
|
|
out["closure_unknown"] = m.impact_closure("nope")
|
|
rec = sm.workspace_record(contract_id="WC", construction_base="sha-base")
|
|
out["record"] = rec
|
|
# snapshot at capture: the ORACLE mutates the record in place and
|
|
# returns the same dict — a later call through the same reference
|
|
# would alias into the captured result (Python-only artifact, not a
|
|
# kernel divergence; the kernel returns fresh dicts)
|
|
out["advance"] = copy.deepcopy(
|
|
sm.workspace_advance(rec, "base_verified", evidence="seed"))
|
|
out["advance2"] = copy.deepcopy(
|
|
sm.workspace_advance(rec, "worktree_created"))
|
|
try:
|
|
sm.workspace_advance(out["advance2"], "pushed")
|
|
out["advance_err"] = "NO RAISE"
|
|
except ValueError as e:
|
|
out["advance_err"] = str(e)
|
|
print(json.dumps(out, sort_keys=True))
|
|
"""
|
|
|
|
|
|
def facade_battery():
|
|
def run(tag):
|
|
env = dict(os.environ)
|
|
env.pop("RUNE_STM_CORE", None)
|
|
env.pop("RUNE_STRUCTURAL_CORE", None)
|
|
if tag == "python":
|
|
env["RUNE_STM_CORE"] = "python"
|
|
env["RUNE_STRUCTURAL_CORE"] = "python"
|
|
r = subprocess.run(
|
|
[sys.executable, "-c", FACADE_CODE], capture_output=True, text=True, cwd=str(REPO), env=env, timeout=180
|
|
)
|
|
if r.returncode != 0:
|
|
fail("live", tag, r.stderr[-400:])
|
|
return None
|
|
return json.loads(r.stdout)
|
|
|
|
py = run("python")
|
|
rs = run("rust")
|
|
if py is None or rs is None:
|
|
return
|
|
check("live", "dual-mode-facade", py, rs)
|
|
|
|
|
|
def main():
|
|
if not stm_loader.available():
|
|
print("Rust core not built — run cargo build --release in src/rust/structural_model")
|
|
return 2
|
|
with tempfile.TemporaryDirectory() as td:
|
|
frozen()
|
|
byte_law(Path(td))
|
|
fuzz(Path(td))
|
|
facade_battery()
|
|
total = sum(counts.values())
|
|
print(
|
|
f"\nPARITY: {total} checks, {len(fails)} failures "
|
|
f"(frozen={counts['frozen']} fuzz={counts['fuzz']} live={counts['live']})"
|
|
)
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|