rune/scripts/gen_parallel_processor_truth.py

#!/usr/bin/env python3
"""gen_parallel_processor_truth.py — freeze the parallel_processor oracle
truth (dichotomy gap-close, wave #2479) through the SHARED harness
(scripts/freeze_truth.py) under the scripts/pp_freeze_seams rig.

Row grammar: raw values ride directly (non-callable / non-tuple error
branches need no rig); callable behavior rides the kind/fn/value spec
grammar the rig turns into real Python callables and the Go battery
rebuilds into specCallable tasks.

Usage: PYTHONPATH=scripts/pp_freeze_seams:src python3 scripts/gen_parallel_processor_truth.py
"""
import sys
from pathlib import Path

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

from freeze_truth import freeze, write_envelope  # noqa: E402

FROZEN_NOW = "2026-10-14T08:00:00+00:00"
OUT = REPO / "src" / "go" / "parallel-processor" / "truth.json"


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


CASES = [
    # module run_parallel — raw non-callables (direct rows, no seam)
    ("run_parallel", ([42],), {}),
    ("run_parallel", ([42, "x", None, True, 4.5, [1], {"a": 1}],), {}),
    ("run_parallel", ([],), {}),
    # module run_parallel_with_args — raw non-tuples + the len-0 list task
    ("run_parallel_with_args", ([42, "x", None],), {}),
    ("run_parallel_with_args", ([[]],), {}),
    # rig seams — callable specs through the real executor paths
    ("_pp_run_parallel", ([s("call", fn="const", ret=7), s("call", fn="boom", msg="kaput"), s("call", fn="const", ret="x")],), {}),
    ("_pp_run_parallel", ([s("call", fn="sleepret", ms=60, ret="a"), s("call", fn="sleepret", ms=10, ret="b"), s("call", fn="sleepret", ms=35, ret="c"), s("call", fn="boom", msg="late"), s("raw", value=None)],), {}),
    ("_pp_run_parallel", ([],), {}),
    ("_pp_class_run_parallel", (3, [s("call", fn="const", ret=1), s("call", fn="const", ret=2), s("call", fn="boom", msg="nope"), s("call", fn="const", ret=4), s("call", fn="sleepret", ms=15, ret=5)]), {}),
    ("_pp_class_run_parallel", (1, [s("call", fn="sleepret", ms=30, ret="w1"), s("call", fn="sleepret", ms=5, ret="w2"), s("call", fn="const", ret="w3")]), {}),
    # with-args branch matrix
    ("_pp_run_parallel_with_args", ([
        s("call", fn="add", args=[1, 2]),
        s("call", fn="add", args=[1.5, 2]),
        s("call", fn="echo"),
        s("call", fn="echo", kwargs={"k": 1, "b": 2}),
        s("call", fn="boom", msg="bad-task"),
    ],), {}),
    ("_pp_run_parallel_with_args", ([
        s("raw", value=42),
        s("raw", value="x"),
        s("raw", value=None),
        s("emptytuple"),
        s("notcall", value=42),
        s("notcall", value="x"),
        s("notcall", value=None),
        s("notcall", value=True),
        s("notcall", value=4.5),
        s("notcall", value=[1, 2]),
    ],), {}),
    ("_pp_run_parallel_with_args", ([
        s("argsnotseq", fn="echo", value="ab"),
        s("argsnotseq", fn="echo", value=5),
        s("kwnotdict", fn="echo", value=["a"]),
        s("call", fn="sleepret", ms=10, ret="z"),
    ],), {}),
    ("_pp_class_run_parallel_with_args", (2, [
        s("call", fn="echo", args=[1, 2], kwargs={"scale": 10}),
        s("raw", value=3.5),
        s("call", fn="boom", msg="class-branch"),
        s("notcall", value=False),
    ]), {}),
    # status surface (active_threads pinned by the rig on both sides)
    ("get_status", (), {}),
    ({"fn": "ParallelProcessor.get_status", "recv": {"cls": "ParallelProcessor", "max_workers": 1}, "args": [], "kwargs": {}}),
    ({"fn": "ParallelProcessor.get_status", "recv": {"cls": "ParallelProcessor", "max_workers": 9}, "args": [], "kwargs": {}}),
    ("_pp_get_processor_status", (), {}),
    # lifecycle: shutdown shapes, crash parity, global reset
    ("_pp_shutdown_then_status", (), {}),
    ("_pp_shutdown_then_run", (), {}),
    ("_pp_new_bad", (), {}),
    ("_pp_shutdown_processor_cycle", (), {}),
    # oracle self-test fn (returns None; its printed report is wall-clock,
    # verified structurally by the parity sweep instead)
    ("test_parallel", (), {}),
]


def main() -> int:
    truth = freeze("rune.parallel_processor", CASES, oracle_module="rune.parallel_processor")
    write_envelope(FROZEN_NOW, truth, OUT)

    # spot-check the shapes the battery depends on
    by_fn = {}
    for row in truth:
        by_fn.setdefault(row["fn"], []).append(row)
    assert len(truth) == len(CASES), f"freeze produced {len(truth)} rows for {len(CASES)} cases"
    assert by_fn["run_parallel"][0]["expected"] == [{"error": "'int' object is not callable"}]
    assert by_fn["_pp_shutdown_then_run"][0]["expected"]["raised"] == "cannot schedule new futures after shutdown"
    assert by_fn["_pp_new_bad"][0]["expected"]["raised"] == "max_workers must be greater than 0"
    assert by_fn["_pp_shutdown_processor_cycle"][0]["expected"] == {
        "processor_is_none": True, "recreated_max_workers": 4}
    assert by_fn["get_status"][0]["expected"] == {
        "max_workers": 4, "pool_threads": 4, "active_threads": 4}
    assert by_fn["_pp_class_run_parallel"][1]["expected"] == ["w1", "w2", "w3"]

    print(f"wrote {OUT}: {len(truth)} cases, frozen_now={FROZEN_NOW}")
    for fn, rows in by_fn.items():
        print(f"  {fn}: {len(rows)} row(s)")
    return 0


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