rune/scripts/check_tiered_executor_parity.py

#!/usr/bin/env python3
"""check_tiered_executor_parity.py — live oracle-vs-Go parity sweep for
the tiered_executor gap closure (dichotomy wave #2479, contract
factory-gap-cs).

Run under the scripts/te_freeze_seams rig (PYTHONPATH), which pins
time.time (both sides then ride the TE_LATENCY_MS=0 wall-clock seam) and
starts the OLLAMA echo server the frozen envelope was captured with.
THIS script drives the LIVE Python oracle and the Go binary (`op` stdin
dispatch) across every op family — job/result serializers via the real
constructors, deterministic family paths incl. rune-slicing truncation,
no-profile early returns through subset registries, model paths through
the echo server (payload bytes ride back as the response), crash-parity
error envelopes, _ctx_budget rows, the TestExecutor self-test — plus the
CLI surface (test / execute / usage / unknown), comparing sort_keys JSON
or exact lines. Exit 0 on 100% parity.

Usage: PYTHONPATH=scripts/te_freeze_seams REPO_ROOT=$PWD python3 \\
       scripts/check_tiered_executor_parity.py [--bin src/go/tiered-executor/tiered-executor]
"""
import argparse
import contextlib
import io
import json
import os
import re
import subprocess
import sys
from pathlib import Path

LAT_RE = re.compile(r'"latency_ms": [0-9.eE+-]+')


def norm_latency(text):
    """Wall-clock law: the CLI execute output embeds a measured latency
    (oracle: live time.time; Go: measured unless the seam pin rides env)
    — replaced structurally on BOTH sides before the byte compare."""
    return LAT_RE.sub('"latency_ms": <wall-clock>', text)

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO / "src"))

import rune.tiered_executor as te  # rig-bound: _te_* seams + frozen time  # noqa: E402
from rune.model_registry import ModelProfile  # noqa: E402

T3_PROFILE = {
    "id": "qwen3.6:27b", "provider": "ollama", "tier": "T3", "local": True,
    "context_window": 32768, "supports_tools": False,
    "supports_structured_output": False, "supports_reasoning_field": False,
    "estimated_latency": "high", "estimated_cost": "0", "enabled": True,
}
EMPTY_REG = {"T9": {**T3_PROFILE, "id": "none", "tier": "T9"}}

BUDGET_T0 = {"start_tier": "T0", "max_tier": "T0"}
BUDGET_T2 = {"start_tier": "T2", "max_tier": "T4"}


def echo_url():
    return os.environ["OLLAMA_URL"]


# ── oracle side (rig seams already give registry specs / per-case urls) ──────

def py_op(op, req):
    if op == "to_dict_job":
        return te._te_job_to_dict(req["job"])
    if op == "to_dict_result":
        return te._te_result_to_dict(req["result"])
    if op == "execute":
        return te._te_execute(req.get("registry"), req["job"], req["budget"])
    if op == "execute_ollama":
        return te._te_execute_ollama(req.get("url"), req.get("registry"),
                                     req["job"], req["budget"])
    if op == "ctx_budget":
        return te._ctx_budget(req["model"])
    if op == "test_executor":
        # te.test_executor prints its report inline — capture it (the CLI
        # parity block below verifies those printed lines byte-exactly)
        sink = io.StringIO()
        with contextlib.redirect_stdout(sink):
            return te._te_test_executor()
    raise ValueError(f"unknown op {op}")


# ── Go side (one fresh process per op) ───────────────────────────────────────

def go_op(binpath, op, req):
    full = dict(req)
    full["op"] = op
    env = dict(os.environ)
    env["TE_LATENCY_MS"] = "0"
    p = subprocess.run([str(binpath), "op"], input=json.dumps(full),
                       capture_output=True, text=True, env=env, cwd=str(REPO))
    if p.returncode != 0:
        raise RuntimeError(f"go op {op} exited {p.returncode}: {p.stdout} {p.stderr}")
    return json.loads(p.stdout)


def jdump(v):
    return json.dumps(_norm(v), sort_keys=True, ensure_ascii=True)


def _norm(v):
    """Canonical walk: integral floats collapse to int for the COMPARISON
    only (dichotomy quirk — Python renders 0.0, Go 0; parsed-equal same
    number). Non-integral floats (latency 12.5) stay untouched."""
    if isinstance(v, bool):
        return v
    if isinstance(v, float) and v.is_integer():
        return int(v)
    if isinstance(v, list):
        return [_norm(x) for x in v]
    if isinstance(v, dict):
        return {k: _norm(x) for k, x in v.items()}
    return v


OPS = [
    ("to_dict_job", {"job": {"job_id": "p1", "task_family": "greeting", "objective": "Respond to hello"}}),
    ("to_dict_job", {"job": {"job_id": "p2", "task_family": "coding", "objective": "日本語 café",
                             "inputs": {"path": "src/x.py"}, "risk": "high",
                             "reversibility": "irreversible", "verification_method": "pytest",
                             "preferred_tiers": ["T0", "T1"], "max_tier": "T3",
                             "required_output_schema": "diff"}}),
    ("to_dict_result", {"result": {"job_id": "p3", "tier_used": "T0", "model_used": "deterministic", "success": True}}),
    ("to_dict_result", {"result": {"job_id": "p4", "tier_used": "T1", "model_used": "qwen2.5:3b",
                                   "success": False, "output": None, "error": "kaput",
                                   "verification_result": {"ok": 1}, "attempts": 2,
                                   "latency_ms": 9.75, "tokens_used": 7}}),
    ("execute", {"job": {"job_id": "p5", "task_family": "greeting", "objective": "salut"}, "budget": BUDGET_T0}),
    ("execute", {"job": {"job_id": "p6", "task_family": "lookup", "objective": "find x"}, "budget": BUDGET_T0}),
    ("execute", {"job": {"job_id": "p7", "task_family": "other", "objective": "plain"}, "budget": BUDGET_T0}),
    # [:100] char-slice rows (dichotomy quirk: rune-safe truncation)
    ("execute", {"job": {"job_id": "p8", "task_family": "greeting", "objective": "x" * 140}, "budget": BUDGET_T0}),
    ("execute", {"job": {"job_id": "p9", "task_family": "lookup", "objective": "日本語" * 60}, "budget": BUDGET_T0}),
    ("execute", {"job": {"job_id": "p10", "task_family": "coding", "objective": "héllo wörld", "preferred_tiers": ["T0"]},
                 "budget": BUDGET_T2}),
    # no-profile early returns through a subset registry (empty dict is FALSY)
    ("execute", {"registry": EMPTY_REG, "job": {"job_id": "p11", "task_family": "coding", "objective": "x",
                                                "preferred_tiers": ["T4"]}, "budget": {"start_tier": "T4", "max_tier": "T4"}}),
    ("execute", {"registry": EMPTY_REG, "job": {"job_id": "p12", "task_family": "coding", "objective": "x"},
                 "budget": {"start_tier": "T3", "max_tier": "T3"}}),
    # model paths through the rig's echo server (payload rides back)
    ("execute_ollama", {"job": {"job_id": "p13", "task_family": "coding", "objective": "fix it"},
                        "budget": {"start_tier": "T1", "max_tier": "T2"}}),
    ("execute_ollama", {"job": {"job_id": "p14", "task_family": "coding",
                                "objective": ('héllo wörld ' * 80) + '日本語 "q" \\ \t\n' + ("🚀" * 30)},
                        "budget": {"start_tier": "T1", "max_tier": "T2"}}),
    ("execute_ollama", {"job": {"job_id": "p15", "task_family": "coding", "objective": "preferred",
                                "preferred_tiers": ["T2"]},
                        "budget": {"start_tier": "T0", "max_tier": "T4"}}),
    ("execute_ollama", {"registry": {"T1": {**T3_PROFILE, "id": "custom-1b", "tier": "T1"}},
                        "job": {"job_id": "p16", "task_family": "coding", "objective": "custom"},
                        "budget": {"start_tier": "T1", "max_tier": "T2"}}),
    # crash-parity rows through the model path
    ("execute_ollama", {"url": "definitely-not-a-url",
                        "job": {"job_id": "p17", "task_family": "coding", "objective": "bad"},
                        "budget": {"start_tier": "T1", "max_tier": "T2"}}),
    ("execute_ollama", {"url": "http://127.0.0.1:9",
                        "job": {"job_id": "p18", "task_family": "coding", "objective": "refused"},
                        "budget": {"start_tier": "T1", "max_tier": "T2"}}),
    ("ctx_budget", {"model": "qwen2.5:3b"}),
    ("ctx_budget", {"model": "qwen3:14b"}),
    ("ctx_budget", {"model": "mimo-v2.5-pro"}),
    ("ctx_budget", {"model": ""}),
    ("ctx_budget", {"model": "deterministic"}),
    ("ctx_budget", {"model": "no-such-model"}),
    ("test_executor", {}),
]


def cli_lines(argv, env_extra=None):
    env = dict(os.environ)
    if env_extra:
        env.update(env_extra)
    env["TE_LATENCY_MS"] = "0"
    p = subprocess.run(argv, capture_output=True, text=True, env=env, cwd=str(REPO))
    return p.returncode, p.stdout, p.stderr


def cli_parity(binpath):
    """CLI surface: test / execute T0 / execute T1 (echo) / usage /
    unknown arg / execute with a missing arg — the oracle's main()."""
    checks = []
    oracle = [sys.executable, str(REPO / "src/rune/tiered_executor.py")]

    rc_p, out_p, _ = cli_lines(oracle + ["test"], {"OLLAMA_URL": echo_url()})
    rc_g, out_g, _ = cli_lines([str(binpath), "test"], {"OLLAMA_URL": echo_url()})
    checks.append(("cli test", rc_p == rc_g and out_p == out_g, out_p, out_g))

    def run(argv, env_extra):
        rc, out, _ = cli_lines(argv, env_extra)
        return rc, norm_latency(out), _

    job = json.dumps({"job_id": "c1", "task_family": "greeting", "objective": "hi"})
    rc_p, out_p, _ = run(oracle + ["execute", job, json.dumps(BUDGET_T0)], {"OLLAMA_URL": echo_url()})
    rc_g, out_g, _ = run([str(binpath), "execute", job, json.dumps(BUDGET_T0)], {"OLLAMA_URL": echo_url()})
    checks.append(("cli execute T0 (latency normalized)", rc_p == rc_g and out_p == out_g, out_p, out_g))

    job_t1 = json.dumps({"job_id": "c2", "task_family": "coding", "objective": "fix it"})
    rc_p, out_p, _ = run(oracle + ["execute", job_t1, json.dumps({"start_tier": "T1", "max_tier": "T2"})],
                         {"OLLAMA_URL": echo_url()})
    rc_g, out_g, _ = run([str(binpath), "execute", job_t1, json.dumps({"start_tier": "T1", "max_tier": "T2"})],
                         {"OLLAMA_URL": echo_url()})
    checks.append(("cli execute T1 via echo (latency normalized)", rc_p == rc_g and out_p == out_g, out_p, out_g))

    rc_p, out_p, _ = cli_lines(oracle + ["execute", job])
    rc_g, out_g, _ = cli_lines([str(binpath), "execute", job])
    checks.append(("cli execute missing budget -> usage", rc_p == rc_g and out_p == out_g, out_p, out_g))

    rc_p, out_p, _ = cli_lines(oracle + ["bogus"])
    rc_g, out_g, _ = cli_lines([str(binpath), "bogus"])
    checks.append(("cli unknown", rc_p == rc_g and out_p == out_g, out_p, out_g))

    rc_p, out_p, _ = cli_lines(oracle)
    rc_g, out_g, _ = cli_lines([str(binpath)])
    checks.append(("cli usage", rc_p == rc_g and out_p == out_g, out_p, out_g))
    return checks


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--bin", default=str(REPO / "src/go/tiered-executor/tiered-executor"))
    args = ap.parse_args()
    binpath = Path(args.bin)

    failures = 0
    total = 0
    for op, req in OPS:
        total += 1
        py = py_op(op, req)
        go = go_op(binpath, op, req)
        if jdump(py) != jdump(go):
            failures += 1
            print(f"DIVERGE {op} {json.dumps(req)[:100]}")
            print(f"  python: {jdump(py)[:400]}")
            print(f"  go:     {jdump(go)[:400]}")

    for name, ok, want, got in cli_parity(binpath):
        total += 1
        if not ok:
            failures += 1
            print(f"DIVERGE {name}\n  python: {want!r}\n  go:     {got!r}")

    print(f"parity: {total - failures}/{total} match" + (" — FAILURES" if failures else ""))
    return 1 if failures else 0


if __name__ == "__main__":
    raise SystemExit(main())