rune/scripts/check_parallel_processor_parity.py

#!/usr/bin/env python3
"""check_parallel_processor_parity.py — live oracle-vs-Go parity sweep for
the parallel_processor gap closure (#2479, contract factory-gap-pp).

The frozen envelope (truth.json) pins the ported surface byte-identically.
THIS script drives the LIVE Python oracle and the Go binary (stdin op
dispatch, `parallel-processor op`) across every op family — run_parallel,
run_parallel_with_args (full parse-branch matrix), class pools at several
worker counts, status, singleton/global lifecycle, crash parity — plus the
CLI surface (status / test / unknown / usage), comparing sort_keys JSON
(structural) or exact lines. Exit 0 on 100% parity.

Usage: PYTHONPATH=scripts/pp_freeze_seams:src python3 scripts/check_parallel_processor_parity.py
  [--bin src/go/parallel-processor/parallel-processor]
"""
import argparse
import json
import os
import subprocess
import sys
from pathlib import Path

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

# the rig (on PYTHONPATH) pins threading.active_count to this and binds
# the spec seams; the Go seam is pinned to the same value per process.
FROZEN_ACTIVE = 4

import rune.parallel_processor as pp  # noqa: E402


def s(kind, **kw):
    d = {"kind": kind}
    d.update(kw)
    return d


# ── oracle side (rig seams give the spec grammar real callables) ────────────

def py_op(op, req):
    if op == "run_parallel":
        return pp.run_parallel([_zero(t) for t in req["tasks"]])
    if op == "run_parallel_with_args":
        return pp.run_parallel_with_args([_task(t) for t in req["tasks"]])
    if op == "class_run_parallel":
        p = pp.ParallelProcessor(req["max_workers"])
        r = p.run_parallel([_zero(t) for t in req["tasks"]])
        p.shutdown(wait=True)
        return r
    if op == "class_run_parallel_with_args":
        p = pp.ParallelProcessor(req["max_workers"])
        r = p.run_parallel_with_args([_task(t) for t in req["tasks"]])
        p.shutdown(wait=True)
        return r
    if op == "get_status":
        return pp.get_status()
    if op == "class_get_status":
        p = pp.ParallelProcessor(req["max_workers"])
        r = {"result": p.get_status(), "max_workers": req["max_workers"]}
        p.shutdown(wait=True)
        return r
    if op == "get_processor_status":
        return pp._pp_get_processor_status()
    if op == "shutdown_then_status":
        return pp._pp_shutdown_then_status()
    if op == "shutdown_then_run":
        return pp._pp_shutdown_then_run()
    if op == "new_bad":
        return pp._pp_new_bad()
    if op == "shutdown_processor_cycle":
        return pp._pp_shutdown_processor_cycle()
    raise ValueError(f"unknown op {op}")


def _zero(spec):
    return pp._pp_spec_zero(spec)


def _task(spec):
    return pp._pp_spec_task(spec)


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

def go_op(binpath, op, req):
    full = dict(req)
    full["op"] = op
    env = dict(os.environ)
    env["PP_ACTIVE_THREADS"] = str(FROZEN_ACTIVE)
    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(v, sort_keys=True)


# ── the sweep ───────────────────────────────────────────────────────────────

OPS = [
    ("run_parallel", {"tasks": [s("call", fn="const", ret=1), s("call", fn="boom", msg="x"), 42, None, "s", True, 4.5, [1], {"a": 1}]}),
    ("run_parallel", {"tasks": []}),
    ("run_parallel", {"tasks": [s("call", fn="sleepret", ms=50, ret="a"), s("call", fn="sleepret", ms=5, ret="b"), s("call", fn="sleepret", ms=25, ret="c")]}),
    ("run_parallel_with_args", {"tasks": [
        s("call", fn="echo", args=[1, 2], kwargs={"scale": 10}),
        s("call", fn="add", args=[7, 8]),
        s("call", fn="add", args=[2.25, 1]),
        s("call", fn="boom", msg="wops"),
        s("raw", value=3.5),
        s("emptytuple"),
        s("notcall", value=[3, 4]),
        s("argsnotseq", fn="echo", value="xy"),
        s("kwnotdict", fn="echo", value="kw"),
    ]}),
    ("class_run_parallel", {"max_workers": 2, "tasks": [s("call", fn="const", ret="p"), s("call", fn="const", ret="q"), s("call", fn="boom", msg="in-class"), s("call", fn="sleepret", ms=15, ret="r")]}),
    ("class_run_parallel", {"max_workers": 8, "tasks": [s("raw", value=False)] * 3 + [s("call", fn="const", ret="ok")]}),
    ("class_run_parallel_with_args", {"max_workers": 1, "tasks": [s("call", fn="echo"), s("notcall", value=0), s("call", fn="add", args=[20, 22])]}),
    ("get_status", {}),
    ("class_get_status", {"max_workers": 1}),
    ("class_get_status", {"max_workers": 9}),
    ("get_processor_status", {}),
    ("shutdown_then_status", {}),
    ("shutdown_then_run", {}),
    ("new_bad", {}),
    ("shutdown_processor_cycle", {}),
]


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


def cli_parity(binpath):
    """CLI surface: status / test / unknown / usage — the oracle's main()."""
    checks = []
    go_env = {"PP_ACTIVE_THREADS": str(FROZEN_ACTIVE)}

    rc_p, out_p, _ = cli_lines([sys.executable, str(REPO / "src/rune/parallel_processor.py"), "status"])
    rc_g, out_g, _ = cli_lines([str(binpath), "status"], go_env)
    checks.append(("cli status", rc_p == rc_g and out_p == out_g, out_p, out_g))

    rc_p, out_p, _ = cli_lines([sys.executable, str(REPO / "src/rune/parallel_processor.py"), "test"])
    rc_g, out_g, _ = cli_lines([str(binpath), "test"])
    # wall-clock lines vary by design; the Results line and the verdict
    # line are the deterministic contract
    det = lambda out: (out.strip().splitlines()[-2], out.strip().splitlines()[-1])  # noqa: E731
    checks.append(("cli test (results+verdict)", rc_p == rc_g and det(out_p) == det(out_g), det(out_p), det(out_g)))

    rc_p, out_p, _ = cli_lines([sys.executable, str(REPO / "src/rune/parallel_processor.py"), "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([sys.executable, str(REPO / "src/rune/parallel_processor.py")])
    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/parallel-processor/parallel-processor"))
    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)[:80]}")
            print(f"  python: {jdump(py)[:300]}")
            print(f"  go:     {jdump(go)[:300]}")

    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())