rune/scripts/check_builder_bridge_parity.py

#!/usr/bin/env python3
"""check_builder_bridge_parity.py — LIVE oracle↔Go parity sweep for the
builder_bridge gap closure (dichotomy wave, worktree rune-wave-cw).

The adaptive_router precedent: preflight returns a BuilderCapsule the
freeze harness cannot JSON-serialize and stamps generated_at from an
in-function datetime import, and _default_selector/_get_selector return
live selector instances. Those surfaces are pinned HERE, by driving the
live Python oracle and the Go binary across every branch with full
to_dict + to_markdown comparison (generated_at normalized — both sides
render the same wall-clock shape, the value is the caller's).

Case families:
  * route branches: LOCAL_FIX defect, STANDARD_CHANGE, RESEARCHED_CHANGE
    floor (incl. the doubled floor-prefix quirk), MEASUREMENT_REQUIRED
    language, inspection-verb neutral class, unicode, >200-rune titles.
  * construction branches: committed fixtures (active / quiet / terminal)
    AND the DEFAULT store branch (both sides read the real repo store
    live — equal at the same moment regardless of its content).
  * memory: degrade-safe abstention under an isolated HOME (both sides).
  * selector observables: lazy singleton identity + non-None default
    selector (Python-observable contract; Go's is compile-time).

Usage (from the repo root):
    python3 scripts/check_builder_bridge_parity.py
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
GO_DIR = REPO / "src" / "go" / "builder-bridge"
GO_BIN = Path(tempfile.mkdtemp(prefix="bb-parity-")) / "builder-bridge"
FIX = GO_DIR / "fixtures"

NOW = "2026-10-13T11:00:00+00:00"

PY_DRIVER = r"""
import json, os
import sys
sys.path.insert(0, os.environ["BB_REPO"] + "/src")
from rune import builder_bridge as bb
task = os.environ["BB_TASK"]
cap = bb.preflight(task)
d = cap.to_dict()
sel = bb._get_selector()
print(json.dumps({
    "dict": d,
    "md": cap.to_markdown(),
    "lines": bb._active_construction_lines(),
    "store": bb._construction_store(),
    "selector_identity": sel is bb._get_selector(),
    "default_selector": type(bb._default_selector()).__name__,
    "retrieve_memory": bb._retrieve_memory(task),
    "cache_key": bb._cache_key(task),
}, ensure_ascii=False))
"""


def canonical(v):
    if isinstance(v, dict):
        return {k: canonical(v[k]) for k in sorted(v)}
    if isinstance(v, list):
        return [canonical(x) for x in v]
    return v


def normalize(doc):
    doc = json.loads(json.dumps(doc))
    d = doc.get("dict")
    if isinstance(d, dict):
        d["generated_at"] = "TS"
    store = doc.get("store")
    if isinstance(store, str):
        doc["store"] = store.replace(str(REPO) + "/", "REPO/")
    return doc


def run_python(task: str, construction: str | None) -> dict:
    env = dict(os.environ)
    home = tempfile.mkdtemp(prefix="bb-parity-py-")
    env["HOME"] = home
    env["RUNE_HOME"] = os.path.join(home, ".rune")
    env["BB_REPO"] = str(REPO)
    env["BB_TASK"] = task
    if construction:
        env["RUNE_CONSTRUCTION_STORE"] = construction
    else:
        env.pop("RUNE_CONSTRUCTION_STORE", None)
    out = subprocess.run(
        [sys.executable, "-c", PY_DRIVER], capture_output=True, text=True,
        env=env, cwd=str(REPO), timeout=120,
    )
    if out.returncode != 0:
        raise RuntimeError(f"python driver failed: {out.stderr[-2000:]}")
    return normalize(json.loads(out.stdout))


def run_go(task: str, construction: str | None, mode: str = "--json") -> dict:
    env = dict(os.environ)
    home = tempfile.mkdtemp(prefix="bb-parity-go-")
    env["HOME"] = home
    env["RUNE_HOME"] = os.path.join(home, ".rune")
    if construction:
        env["RUNE_CONSTRUCTION_STORE"] = construction
    else:
        env.pop("RUNE_CONSTRUCTION_STORE", None)
    args = [str(GO_BIN), "parity", task, "--now", NOW, mode]
    out = subprocess.run(args, capture_output=True, text=True,
                         env=env, cwd=str(REPO), timeout=60)
    if out.returncode != 0:
        raise RuntimeError(f"go driver failed: {out.stderr[-2000:]}")
    if mode == "--json":
        doc = json.loads(out.stdout)
        doc["generated_at"] = "TS"
        return doc
    if mode == "--store":
        return out.stdout.strip().replace(str(REPO) + "/", "REPO/")
    # --md: fmt.Println appends exactly one trailing newline; to_markdown
    # output never ends with one, so strip it for the raw-string compare.
    if out.stdout.endswith("\n"):
        return out.stdout[:-1]
    return out.stdout


CASES = [
    ("local_defect_active", "Fix the telemetry summary log level constant in stage_telemetry.py", "active"),
    ("research_floor", "New subsystem: evaluate and adopt an external vector database dependency", "active"),
    ("standard_change", "Extend the config struct with a new field for the retry budget", "quiet"),
    ("quiet_local_fix", "Fix typo in CLI help text", "quiet"),
    ("terminal_store", "Fix the crash in the retry loop when the queue is empty", "terminal"),
    ("default_store", "Fix the log level constant", None),
    ("unicode_task", "Réparez le café ☕ débordé du module de télémétrie et le radix élevé naïf", "active"),
    ("long_title", "Measure and fix " + "the very long running parsing pipeline stage " * 12, "active"),
    ("mechanism_language", "measure the throughput of the parser under load", "active"),
    ("inspection_verb", "explain what the retry module does with the backoff", "quiet"),
    ("external_ref", "see https://example.com/api docs for the field names", "active"),
]


def main() -> int:
    build = subprocess.run(
        ["go", "build", "-o", str(GO_BIN), "."],
        cwd=str(GO_DIR), capture_output=True, text=True,
    )
    if build.returncode != 0:
        print("go build failed:", build.stderr)
        return 1

    fails = 0
    for name, task, store in CASES:
        construction = str(FIX / f"store_{store}") if store else None
        py = run_python(task, construction)
        go = run_go(task, construction, "--json")

        # capsule dict parity (timestamp-normalized)
        py_dict = canonical(py["dict"])
        go_dict = canonical(go)
        if py_dict != go_dict:
            fails += 1
            print(f"FAIL {name} dict:\n  PY: {json.dumps(py_dict, sort_keys=True, ensure_ascii=False)[:600]}\n  GO: {json.dumps(go_dict, sort_keys=True, ensure_ascii=False)[:600]}")

        # markdown parity (rendered from the same capsule)
        py_md = py["md"]
        go_md = run_go(task, construction, "--md")
        if py_md != go_md:
            fails += 1
            print(f"FAIL {name} md:\n  PY: {py_md[:300]!r}\n  GO: {go_md[:300]!r}")

        # construction projection parity (lines mode)
        py_lines = canonical(py["lines"])
        go_lines = canonical(json.loads(run_go(task, construction, "--lines")))
        if py_lines != go_lines:
            fails += 1
            print(f"FAIL {name} lines:\n  PY: {py_lines}\n  GO: {go_lines}")

        # store resolution parity (env verbatim / repo-root default)
        py_store = py["store"]
        go_store = run_go("x", construction, "--store")
        if py_store != go_store:
            fails += 1
            print(f"FAIL {name} store: PY {py_store!r} GO {go_store!r}")

        print(f"ok {name}: route={py_dict.get('task_route')} floor={py_dict.get('floor')} "
              f"lines={len(py['lines'])} constraints={len(py_dict.get('constraints', []))}")

    # selector observables (Python-side contract; Go pins via compile-time
    # singleton + the frozen abstention rows)
    py = run_python("selector probe", str(FIX / "store_active"))
    if not py["selector_identity"]:
        fails += 1
        print("FAIL selector identity: _get_selector() is not a singleton")
    if py["default_selector"] != "MemorySelector":
        fails += 1
        print(f"FAIL default selector: {py['default_selector']!r} != 'MemorySelector'")
    if py["retrieve_memory"] != []:
        fails += 1
        print(f"FAIL retrieve under isolated HOME: {py['retrieve_memory']!r} != []")
    print("ok selector: singleton identity, MemorySelector default, abstention []")

    if fails:
        print(f"PARITY FAILED: {fails} divergence(s)")
        return 1
    print(f"PARITY OK: {len(CASES)} case families byte-equal (dict + markdown + lines + store)")
    return 0


if __name__ == "__main__":
    sys.exit(main())