#!/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())