drippu/scripts/recompiler_compat_baseline.py

#!/usr/bin/env python3
"""Run the self-contained recompiler compatibility baseline.

The baseline intentionally uses the synthetic CodeSet fixture in
``recomp_stack_harness``.  It is not a title compatibility claim: it records
the emulator/backend behavior that can be reproduced without firmware, keys,
or copyrighted game data.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import platform
import signal
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path


SCHEMA_VERSION = 1
WORKLOADS = {
    "boot": "bootstrap SetRecompLookup + application KProcess ArmRecomp",
    "sustained_execution": "identical JIT vs hybrid AOT workload (slice/startup/compile/memory/size)",
    "save_load": "ThreadContext save/load restores AOT execution and TLS",
    "restart": "new process ArmRecomp (fresh icache) re-runs Translate AOT",
}


def git(cwd: Path, *args: str) -> str:
    try:
        return subprocess.run(["git", *args], cwd=cwd, text=True,
                              stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                              check=False).stdout.strip()
    except OSError:
        return ""


def tool_version(name: str) -> str:
    tool = shutil.which(name)
    if not tool:
        return ""
    try:
        return subprocess.run([tool, "--version"], text=True, stdout=subprocess.PIPE,
                              stderr=subprocess.STDOUT, check=False).stdout.splitlines()[0]
    except (OSError, IndexError):
        return ""


def parse_settings(values: list[str]) -> dict[str, str]:
    settings: dict[str, str] = {
        "cpu_backend": "ArmRecomp with Dynarmic fallback",
        "fixture_format": "synthetic CodeSet (not NSO/NRO)",
        "firmware": "not required",
        "keys": "not required",
    }
    for value in values:
        if "=" not in value:
            raise ValueError(f"--setting requires KEY=VALUE: {value}")
        key, setting = value.split("=", 1)
        if not key:
            raise ValueError(f"--setting has an empty key: {value}")
        settings[key] = setting
    return settings


def terminate_process_group(proc: subprocess.Popen[str]) -> None:
    """Terminate the harness and every child it started, on both host OSes."""
    if proc.poll() is not None:
        return
    if os.name == "nt":
        # CREATE_NEW_PROCESS_GROUP lets CTRL_BREAK_EVENT reach the harness and
        # its console-group children. Use taskkill /T /F only if the group did
        # not exit during the graceful window; unlike proc.terminate(), this
        # cannot leave a compiler/helper child behind after the parent exits.
        try:
            proc.send_signal(signal.CTRL_BREAK_EVENT)
        except (AttributeError, OSError):
            # Some Windows hosts do not expose a console for CTRL_BREAK. The
            # tree kill below remains the hard cleanup path.
            pass
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"],
                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                           check=False)
            proc.wait()
        return
    try:
        process_group = os.getpgid(proc.pid)
    except ProcessLookupError:
        return
    # TERM first gives the harness a bounded, graceful shutdown window.
    os.killpg(process_group, signal.SIGTERM)
    try:
        proc.wait(timeout=5)
    except subprocess.TimeoutExpired:
        pass
    else:
        # The parent exited, but descendants may still be attached to the
        # process group. We still kill the complete group below.
        pass
    try:
        os.killpg(process_group, signal.SIGKILL)
    except ProcessLookupError:
        pass
    if proc.poll() is None:
        proc.kill()
    proc.wait()


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--harness", required=True, type=Path,
                        help="path to the built recomp_stack_harness executable")
    parser.add_argument("--output", required=True, type=Path,
                        help="JSON path to write, including on harness failure")
    parser.add_argument("--iterations", type=int, default=32,
                        help="sustained workload iteration setting (default: 32)")
    parser.add_argument("--timeout-seconds", type=float, default=60.0,
                        help="maximum harness runtime (default: 60 seconds)")
    parser.add_argument("--setting", action="append", default=[],
                        help="record an additional exact setting as KEY=VALUE")
    parser.add_argument("--allow-failures", action="store_true",
                        help="write the baseline but return success when a phase fails")
    args = parser.parse_args()

    cwd = Path.cwd()
    harness = args.harness if args.harness.is_absolute() else cwd / args.harness
    settings = parse_settings(args.setting)
    settings["benchmark_iterations"] = str(max(4, args.iterations))
    if args.timeout_seconds <= 0:
        parser.error("--timeout-seconds must be positive")
    settings["timeout_seconds"] = str(args.timeout_seconds)
    env = os.environ.copy()
    env["SUYU_RECOMP_BENCH_ITERS"] = settings["benchmark_iterations"]

    # Snapshot provenance before running user code. In particular, a harness
    # that writes generated files must not change the SHA/dirty state recorded
    # for the workload it is about to measure.
    repository = {
        "git_sha": git(cwd, "rev-parse", "HEAD"),
        "git_describe": git(cwd, "describe", "--always", "--dirty"),
        "working_tree_dirty": bool(git(cwd, "status", "--porcelain")),
    }

    command = [str(harness)]
    popen_kwargs: dict[str, object] = {
        "cwd": cwd,
        "env": env,
        "text": True,
        "stdout": subprocess.PIPE,
        "stderr": subprocess.STDOUT,
    }
    if os.name == "nt":
        popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
    else:
        popen_kwargs["start_new_session"] = True
    timed_out = False
    try:
        proc = subprocess.Popen(command, **popen_kwargs)
        try:
            output, _ = proc.communicate(timeout=args.timeout_seconds)
            returncode = proc.returncode
        except subprocess.TimeoutExpired:
            timed_out = True
            terminate_process_group(proc)
            output, _ = proc.communicate()
            returncode = proc.returncode
    except OSError as exc:
        returncode, output = 127, f"unable to launch {harness}: {exc}\n"

    # Keep the normal CI log useful while preserving a complete copy in JSON.
    print(output, end="")
    failures: list[dict[str, str]] = []
    results = []
    for name, marker in WORKLOADS.items():
        passed = marker in output
        result = {
            "name": name,
            "status": "pass" if passed else "fail",
            "evidence": marker,
        }
        if not passed:
            reason = "required harness evidence was not found"
            failures.append({"workload": name, "reason": reason})
            result["failure"] = reason
        results.append(result)
    if returncode != 0:
        failures.append({"workload": "harness", "reason": f"exit code {returncode}"})
    if timed_out:
        failures.append({"workload": "harness",
                         "reason": f"timeout after {args.timeout_seconds:g} seconds"})

    output_path = args.output if args.output.is_absolute() else cwd / args.output
    output_path.parent.mkdir(parents=True, exist_ok=True)
    record = {
        "schema_version": SCHEMA_VERSION,
        "kind": "recompiler_compatibility_baseline",
        "generated_at_utc": dt.datetime.now(dt.timezone.utc).isoformat(),
        "repository": repository,
        "host": {
            "os": platform.platform(),
            "architecture": platform.machine(),
            "python": platform.python_version(),
            "compiler": tool_version(os.environ.get("CXX", "c++")),
        },
        "command": {
            "argv": command,
            "cwd": str(cwd),
            "return_code": returncode,
            "timed_out": timed_out,
        },
        "settings": settings,
        "fixture": {
            "name": "recompiler_synthetic_homebrew",
            "description": "Generated AArch64 CodeSet image exercising ArmRecomp, Dynarmic, kernel SVCs, TLS, memory, and restart.",
            "redistributable": True,
            "external_firmware_or_keys": False,
        },
        "workloads": results,
        "failures": failures,
        "harness_output_tail": output[-12000:],
    }
    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=output_path.parent,
                                    prefix=output_path.name + ".", suffix=".tmp",
                                    delete=False) as tmp:
        json.dump(record, tmp, indent=2, sort_keys=True)
        tmp.write("\n")
        temporary = Path(tmp.name)
    temporary.replace(output_path)
    print(f"wrote compatibility baseline: {output_path}")
    if failures:
        print(f"compatibility baseline failures: {len(failures)}", file=sys.stderr)
    return 0 if (not failures or args.allow_failures) else 1


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