rune/scripts/check_execution_trace_parity.py

#!/usr/bin/env python3
"""Live parity: execution_trace Rust kernel vs Python oracle (dichotomy
wave). Layers:
  1. FROZEN — the 20 frozen-oracle cases replayed through the facade in
     BOTH core modes (RUNE_ET_CORE=python floor vs auto kernel); exact
     compare, journal contents rebuilt verbatim from the truth.
  2. FUZZ   — 400 seeded random journals/tasks/payloads (seed 2479)
     driven through both modes; exact compare.
"""
import copy
import json
import os
import random
import sys
import tempfile
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO / "src"))
sys.path.insert(0, str(REPO / "src" / "rune"))

import execution_trace as et  # noqa: E402
from rune import et_loader  # noqa: E402

if not et_loader.available():
    print("FATAL: Rust execution_trace core not built")
    sys.exit(2)

TRUTH = json.loads(
    (REPO / "src" / "rust" / "execution_trace" / "frozen_et_truth.json")
    .read_text())

fails = []
counts = {"frozen": 0, "fuzz": 0}


def fail(layer, name, detail):
    fails.append((layer, name, detail))
    print(f"FAIL[{layer}] {name}: {detail}")


def drive(mode, fn, *args, **kwargs):
    os.environ["RUNE_ET_CORE"] = mode
    try:
        return ("ok", fn(*copy.deepcopy(args), **copy.deepcopy(kwargs)))
    except Exception as e:  # noqa: BLE001 — crash parity is part of the law
        return ("raise", type(e).__name__)
    finally:
        os.environ["RUNE_ET_CORE"] = "auto"


def journal_file(lines):
    if lines is None:
        fd, path = tempfile.mkstemp(suffix=".jsonl")
        os.close(fd)
        os.unlink(path)
        return path
    with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False,
                                     encoding="utf-8") as f:
        f.write("\n".join(lines) + "\n")
        return f.name


def frozen():
    for group, fn, key in (
        ("build_receipt_cases", et.build_receipt, "args"),
        ("trace_task_cases", None, "args"),
    ):
        for c in TRUTH[group]:
            name = c["name"]
            counts["frozen"] += 1
            if group == "build_receipt_cases":
                py = drive("python", et.build_receipt,
                           c["args"]["task"], c["args"]["outcome_payload"])
                rs = drive("auto", et.build_receipt,
                           c["args"]["task"], c["args"]["outcome_payload"])
                if py != rs:
                    fail("frozen", f"build_receipt/{name}", f"{py!r} != {rs!r}")
                continue
            path = journal_file(c["args"]["lines"])
            try:
                py = drive("python", et.trace_task, c["args"]["task_id"],
                           c["args"]["task_record"], path)
                rs = drive("auto", et.trace_task, c["args"]["task_id"],
                           c["args"]["task_record"], path)
            finally:
                if c["args"]["lines"] is not None:
                    os.unlink(path)
            if py != rs:
                fail("frozen", f"trace_task/{name}", f"{py!r} != {rs!r}")
    for c in TRUTH["read_outcome_events_cases"]:
        name = c["name"]
        counts["frozen"] += 1
        path = journal_file(c["lines"])
        try:
            py = drive("python", et._read_outcome_events, path)
            rs = drive("auto", et._read_outcome_events, path)
        finally:
            if c["lines"] is not None:
                os.unlink(path)
        if py != rs:
            fail("frozen", f"read_outcome_events/{name}",
                 f"{py!r} != {rs!r}")


WORDS = ["done", "failed", "running", "", "détaché", "line\\nbreak"]
MODELS = ["", "qwen2.5:3b", "cloud-x", None]


def rand_journal(rng):
    lines = []
    for _ in range(rng.randint(0, 6)):
        r = rng.random()
        tid = f"t{rng.randint(0, 5)}"
        if r < 0.15:
            lines.append(rng.choice(["{not json", "", "[1,2]"]))
            continue
        ev = rng.choice(["task.outcome", "task.started", "task.outcome"])
        payload = {"task_id": tid if rng.random() > 0.1 else "",
                   "status": rng.choice(WORDS)}
        if rng.random() < 0.5:
            payload["model"] = rng.choice(MODELS)
        if rng.random() < 0.5:
            payload["completed_at"] = rng.choice(["", "T-a", "T-b"])
        lines.append(json.dumps({"event_type": ev, "payload": payload}))
    return lines


def rand_task(rng):
    task = {}
    if rng.random() < 0.9:
        task["task_id"] = f"t{rng.randint(0, 7)}"
    if rng.random() < 0.7:
        task["description"] = rng.choice(WORDS)
    if rng.random() < 0.8:
        task["status"] = rng.choice(WORDS)
    if rng.random() < 0.6:
        task["assigned_model"] = rng.choice(MODELS)
    if rng.random() < 0.5:
        task["memories_used"] = ["m1"] if rng.random() < 0.5 else []
    if rng.random() < 0.4:
        task["rules_used"] = []
    if rng.random() < 0.3:
        task["capability"] = rng.choice(["", "forge"])
    if rng.random() < 0.4:
        task["actions"] = ["a"] if rng.random() < 0.5 else []
    if rng.random() < 0.5:
        task["completed_at"] = rng.choice(["", "T-task"])
    return task


def rand_record(rng):
    r = rng.random()
    if r < 0.4:
        return None
    if r < 0.5:
        return {}
    return rand_task(rng)


def fuzz():
    rng = random.Random(2479)
    for i in range(400):
        lines = rand_journal(rng)
        task = rand_task(rng)
        record = rand_record(rng)
        tid = f"t{rng.randint(0, 7)}"
        path = journal_file(lines)
        try:
            py = drive("python", et.trace_task, tid, record, path)
            rs = drive("auto", et.trace_task, tid, record, path)
        finally:
            if lines is not None:
                os.unlink(path)
        counts["fuzz"] += 1
        if py != rs:
            fail("fuzz", f"case-{i}", f"{py!r} != {rs!r}")
            continue
        payload_rng = random.Random(1000 + i)
        t = rand_task(payload_rng)
        p = None if payload_rng.random() < 0.4 else {
            "model": rng.choice(MODELS), "status": rng.choice(WORDS),
            "completed_at": rng.choice(["", "T-p"])}
        py = drive("python", et.build_receipt, t, p)
        rs = drive("auto", et.build_receipt, t, p)
        counts["fuzz"] += 1
        if py != rs:
            fail("fuzz", f"receipt-{i}", f"{py!r} != {rs!r}")


frozen()
fuzz()
total = counts["frozen"] + counts["fuzz"]
print(f"parity: {total - len(fails)}/{total} "
      f"(frozen {counts['frozen']}, fuzz {counts['fuzz']})")
sys.exit(1 if fails else 0)