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