harness/common.py

#!/usr/bin/env python3
"""Shared pieces: loading, fence stripping, the two oracles, and the mandatory harness self-tests.

Two oracles are applied to the SAME predictions on the SAME benchmark, so the delta between them
is a measurement rather than an opinion.

  OURS   -- `pybytecode_core.verify.code_fingerprint`. Recompile the prediction and require the
            resulting code object to be byte-identical to the reference's, recursively, INCLUDING
            docstrings and `co_exceptiontable`. Sound: a pass is a proof, never a guess.
  THEIRS -- `pylingual.equivalence_check.compare_pyc`, imported not reimplemented, so no one can
            say we loosened their bar. CFG-coarsened and docstring-blind (measured, not assumed).
            OPTIONAL: absent PyLingual, everything below still runs on our oracle alone.

Two self-tests gate every score this harness prints, per standing foundry discipline:

  PRE-FLIGHT     grade every reference label against itself. A byte-perfect model MUST score 100%.
                 Anything less means the harness is broken and no score may be quoted.
  MUTATION TEST  deliberately corrupt each label and confirm the oracle KILLS it. A grader that
                 passes mutants is a stub and its scores are meaningless.
"""
from __future__ import annotations

import ast
import json
import py_compile
import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from config import resolve_bench_asset  # noqa: E402
from pybytecode_core.verify import code_fingerprint  # noqa: E402

FENCE = re.compile(r"```(?:python|py)?\s*\n(.*?)(?:```|\Z)", re.S)


def strip_fences(text: str) -> str:
    m = FENCE.search(text or "")
    return (m.group(1) if m else (text or "")).strip()


def load_jsonl(path: str | Path) -> list[dict]:
    return [json.loads(l) for l in Path(path).read_text().splitlines() if l.strip()]


def load_bench(path: str | Path) -> tuple[dict[int, dict], Path]:
    """Return {i: row} with every row's `pyc_path` rewritten to a path that exists HERE."""
    bench_file = Path(path).resolve()
    rows = {}
    for r in load_jsonl(bench_file):
        r["pyc_path"] = str(resolve_bench_asset(bench_file, r.get("pyc_path", ""), "pyc", r["i"]))
        r["src_path"] = str(resolve_bench_asset(bench_file, r.get("src_path", ""), "src", r["i"]))
        rows[r["i"]] = r
    return rows, bench_file


# ------------------------------------------------------------------ our oracle
def ours_ok(pred_src: str, expected_src: str) -> bool:
    """Byte-identical code object, docstrings and exception tables included."""
    try:
        g = compile(pred_src, "<pred>", "exec", dont_inherit=True, optimize=0)
        w = compile(expected_src, "<ref>", "exec", dont_inherit=True, optimize=0)
        return code_fingerprint(g) == code_fingerprint(w)
    except Exception:  # noqa: BLE001
        return False


# ---------------------------------------------------------------- their oracle
_compare_pyc = None


def pylingual_available() -> bool:
    global _compare_pyc
    if _compare_pyc is None:
        try:
            from pylingual.equivalence_check import compare_pyc
            _compare_pyc = compare_pyc
        except Exception:  # noqa: BLE001
            _compare_pyc = False
    return _compare_pyc is not False


def theirs_ok(pred_src: str, ref_pyc: Path, tmp: Path, tag: str) -> tuple[bool, str]:
    """PyLingual's own definition of Perfect: recompile, compare_pyc, all-or-nothing."""
    if not pylingual_available():
        return False, "pylingual not installed"
    if not pred_src.strip():
        return False, "empty"
    try:
        ast.parse(pred_src)
    except SyntaxError:
        return False, "syntax error"
    p, c = tmp / f"{tag}.py", tmp / f"{tag}.pyc"
    try:
        p.write_text(pred_src, encoding="utf-8")
        py_compile.compile(str(p), cfile=str(c), doraise=True, optimize=0)
    except Exception:  # noqa: BLE001
        return False, "does not compile"
    try:
        results = _compare_pyc(Path(ref_pyc), c)
    except Exception as e:  # noqa: BLE001
        return False, f"oracle error: {type(e).__name__}"
    if not results:
        return False, "oracle returned no results"
    if all(r.success for r in results):
        return True, "PERFECT"
    notes = [str(getattr(r, "note", "")) for r in results if not r.success]
    return False, "semantic error: " + "; ".join(n for n in notes[:2] if n)[:120]


# ------------------------------------------------------------------ docstrings
def docstrings_of(src: str) -> list[str]:
    out = []
    try:
        tree = ast.parse(src)
    except SyntaxError:
        return out
    for n in ast.walk(tree):
        if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Module)):
            d = ast.get_docstring(n, clean=False)
            if d is not None:
                out.append(d)
    return out


# ------------------------------------------------------------------ self-tests
def mutations(src: str) -> list[tuple[str, str]]:
    """Semantically REAL corruptions of `src`, each of which a sound oracle must reject.

    Candidates that do not actually change the program are discarded rather than counted. The
    `return_none` rewrite turns `return x` into `return None #x`, which is a genuine change --
    but applied to a bare `return None` it produces `return None #None`, differing only by a
    comment. Counting that as a surviving mutant would blame the oracle for being right; the
    original harness scored 131/131 only because no row in its first 120 had a bare
    `return None`, and this benchmark has three.

    The no-op filter compares ASTs, NOT the oracle under test, so it cannot launder a real
    mutant into a discarded one: `ast.dump` is blind to comments and formatting and to nothing
    else.
    """
    try:
        base = ast.dump(ast.parse(src))
    except SyntaxError:
        return []

    candidates = []
    for name, a, b in (("plus_to_minus", " + ", " - "), ("eq_to_ne", " == ", " != "),
                       ("lt_to_gt", " < ", " > "), ("and_to_or", " and ", " or ")):
        if a in src:
            candidates.append((name, src.replace(a, b, 1)))
    if "return " in src:
        candidates.append(("return_none", src.replace("return ", "return None #", 1)))

    out = []
    for name, m in candidates:
        try:
            if ast.dump(ast.parse(m)) != base:
                out.append((name, m))
        except SyntaxError:
            continue  # a mutant that does not parse tests nothing about the oracle
    return out


def self_test(bench: dict[int, dict], oracle: str = "ours", mutation_rows: int = 120) -> dict:
    """Pre-flight + mutation test. `oracle` is "ours" or "theirs"."""
    import tempfile

    with tempfile.TemporaryDirectory() as td:
        tmp = Path(td)

        def ok(src: str, row: dict, tag: str) -> bool:
            if oracle == "ours":
                return ours_ok(src, row["expected"])
            return theirs_ok(src, Path(row["pyc_path"]), tmp, tag)[0]

        pf_pass, pf_fail, failures = 0, 0, []
        for i, r in bench.items():
            if ok(r["expected"], r, f"pf{i}"):
                pf_pass += 1
            else:
                pf_fail += 1
                if len(failures) < 5:
                    failures.append({"i": i, "func": r.get("csn_func", "")})

        killed = survived = 0
        survivors = []
        for i, r in list(bench.items())[:mutation_rows]:
            for name, m in mutations(r["expected"]):
                if ok(m, r, f"mut{i}"):
                    survived += 1
                    if len(survivors) < 5:
                        survivors.append({"i": i, "mutation": name})
                else:
                    killed += 1

    total = killed + survived
    return {
        "oracle": oracle,
        "preflight_n": len(bench),
        "preflight_perfect": pf_pass,
        "preflight_failed": pf_fail,
        "preflight_pct": round(100 * pf_pass / max(1, len(bench)), 2),
        "preflight_failures": failures,
        "mutation_total": total,
        "mutation_killed": killed,
        "mutation_survived": survived,
        "mutation_kill_rate_pct": round(100 * killed / max(1, total), 2),
        "mutation_survivors": survivors,
        "SOUND": pf_fail == 0 and survived == 0,
    }


def require_sound(st: dict) -> None:
    """A harness that fails either self-test may not report a score. Refuse, loudly."""
    if not st["SOUND"]:
        print(json.dumps(st, indent=2), file=sys.stderr)
        raise SystemExit(
            f"REFUSING TO SCORE: preflight {st['preflight_perfect']}/{st['preflight_n']}, "
            f"mutation kill rate {st['mutation_kill_rate_pct']}%. Both must be 100%."
        )