"""Protected KernelBench evaluator (parent process; never imports candidate code).

evaluate(problem_id, source, gpu, mode="eval") scores one ModelNew candidate against the frozen KernelBench
reference on one GPU:

  0. integrity: problem file hash vs problems/MANIFEST.sha256; evaluator fingerprint recorded with every result.
  1. static gate (static_check.py): imports/attrs/calls allowlist, no monkeypatching, no try/except fallbacks,
     no streams/graphs/events/compile/flags/file IO, banned native tokens, a custom kernel must exist.
  2. compile in a bwrap sandbox (no network, cleared env, /home /tmp /root hidden, no GPU device);
     the generated build sources are scanned again for banned native tokens (catches dynamically built strings).
  3. reference outputs in a TRUSTED child (no candidate code) on fresh os.urandom seeds (fresh weights too); the
     candidate never shares a process or a GPU allocation with the reference, so no reference result can be
     reused; outputs reach the parent as files the sandbox cannot see.
  4. candidate child in the sandbox with ONE GPU device node: correctness outputs on the same seeds, input
     mutation check, profiler check that its own kernel ran, timing-primitive identity and torch-flag checks.
  5. timing (mode="eval"): reference and candidate children each warm up, then the parent holds the GPU
     exclusively and interleaves blocks (ref, cand, ref, cand, ...) of cold-L2 trials on rotating fresh input
     sets; per trial max(cuda-event time, device-synchronised wall time - calibrated overhead), so side-stream
     work cannot hide. The candidate's outputs from the timing trials are checked against the reference too
     (no caching across calls, no timing-only fast paths).
  6. score: KernelBench fast_1 = correct and speedup >= 1 over torch eager. `fast1` uses the bootstrap 95% lower
     bound of the median-ratio speedup (noise cannot make a null solution "faster"); `fast1_raw` the point
     estimate. Speedup vs torch.compile from the baseline table (same GPU type, same timing method).

mode="check" is the NON-protected local check the agent may call: steps 0-4 on one public seed (no timing).
"""
from __future__ import annotations

import glob
import hashlib
import json
import math
import os
import shutil
import subprocess
import threading
import time
import uuid

import torch

import static_check

HERE = os.path.dirname(os.path.abspath(__file__))
PROBLEMS = os.path.normpath(os.path.join(HERE, "..", "problems"))
EVAL_ROOT = os.environ.get("EVAL_ROOT", "/dev/shm/evals")
SYS_PY = "/usr/bin/python3"
TOL = float(os.environ.get("EVAL_TOL", "1e-4"))  # KernelBench get_tolerance_for_precision(fp32) at 423217d9
# Strict fp32 reference: cuDNN TF32 OFF in both processes (torch's default is ON, which moves conv outputs of 72
# KernelBench problems beyond the fp32 tolerance, so a more accurate strict-fp32 kernel would be marked wrong).
CUDNN_TF32 = os.environ.get("EVAL_CUDNN_TF32", "0") == "1"
PUBLIC_SEED = 42  # KernelBench default seed_num; the local check's public input
N_CORRECT = 2
TIME_BUDGET_MS = 1500.0
BLOCKS = 4
MIN_TRIALS, MAX_TRIALS = 12, 200
COMPILE_TIMEOUT = int(os.environ.get("EVAL_COMPILE_TIMEOUT", "420"))
RUN_TIMEOUT = int(os.environ.get("EVAL_RUN_TIMEOUT", "300"))
TIMING_TIMEOUT = int(os.environ.get("EVAL_TIMING_TIMEOUT", "240"))
EXCESSIVE_SPEEDUP = 10.0
EVAL_FILES = ["common.py", "worker.py", "static_check.py", "evaluator.py"]


def _manifest():
    out = {}
    with open(os.path.join(PROBLEMS, "MANIFEST.sha256")) as f:
        for line in f:
            h, p = line.split()
            out[p] = h
    return out


MANIFEST = _manifest()


def problem_files() -> dict:
    """problem id (L{level}_{num:03d}) -> relative path."""
    out = {}
    for p in MANIFEST:
        lvl = int(p.split("/")[0].replace("level", ""))
        num = int(os.path.basename(p).split("_")[0])
        out[f"L{lvl}_{num:03d}"] = p
    return out


PROBLEM_FILES = problem_files()


def fingerprint() -> str:
    h = hashlib.sha256()
    for f in EVAL_FILES:
        h.update(open(os.path.join(HERE, f), "rb").read())
    h.update(open(os.path.join(PROBLEMS, "MANIFEST.sha256"), "rb").read())
    h.update(f"tol={TOL};cudnn_tf32={CUDNN_TF32}".encode())
    return h.hexdigest()[:16]


def problem_path(pid: str) -> str:
    rel = PROBLEM_FILES[pid]
    path = os.path.join(PROBLEMS, rel)
    got = hashlib.sha256(open(path, "rb").read()).hexdigest()
    if got != MANIFEST[rel]:
        raise RuntimeError(f"problem file {rel} does not match MANIFEST")
    return path


def _seed() -> int:
    return int.from_bytes(os.urandom(4), "little") & 0x7FFFFFFF


# --------------------------------------------------------------------------------------------- processes
def _sandbox_cmd(work: str, gpu: int | None, mode: str, cfg_name: str) -> list[str]:
    cmd = ["bwrap", "--ro-bind", "/", "/", "--tmpfs", "/home", "--tmpfs", "/tmp", "--tmpfs", "/root",
           "--tmpfs", "/var/tmp", "--tmpfs", "/run"]
    if os.environ.get("EVAL_HIDE_DIR") and os.path.isdir(os.environ["EVAL_HIDE_DIR"]):
        cmd += ["--tmpfs", os.environ["EVAL_HIDE_DIR"]]
    if os.path.isdir("/data"):
        cmd += ["--tmpfs", "/data"]
    cmd += ["--proc", "/proc", "--dev", "/dev"]
    if gpu is not None:
        # all device nodes (the CUDA driver enumerates every GPU it lists in /proc and fails with a missing node);
        # CUDA_VISIBLE_DEVICES selects ours, and the harness creates its context before any candidate code runs
        nodes = ["/dev/nvidiactl", "/dev/nvidia-uvm", "/dev/nvidia-uvm-tools"] + sorted(glob.glob("/dev/nvidia[0-9]*"))
        for d in nodes:
            if os.path.exists(d):
                cmd += ["--dev-bind", d, d]
    cmd += ["--bind", work, "/tmp/sandbox/work", "--ro-bind", HERE, "/tmp/sandbox/harness",
            "--unshare-all", "--die-with-parent", "--new-session", "--clearenv"]
    env = {"PATH": "/usr/local/cuda/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin", "HOME": "/tmp/sandbox/work",
           "TORCH_EXTENSIONS_DIR": "/tmp/sandbox/work/ext", "TRITON_CACHE_DIR": "/tmp/sandbox/work/triton",
           "XDG_CACHE_HOME": "/tmp/sandbox/work/.cache", "CUDA_HOME": "/usr/local/cuda", "TORCH_CUDA_ARCH_LIST": "8.0",
           "MAX_JOBS": "4", "PYTHONDONTWRITEBYTECODE": "1", "CUDA_DEVICE_ORDER": "PCI_BUS_ID",
           "CUDA_VISIBLE_DEVICES": "" if gpu is None else str(gpu), "TMPDIR": "/tmp/sandbox/work/tmp",
           "TORCHINDUCTOR_CACHE_DIR": "/tmp/sandbox/work/inductor"}
    for k, v in env.items():
        cmd += ["--setenv", k, v]
    cmd += ["--chdir", "/tmp/sandbox/work", SYS_PY, "-I", "-u", "/tmp/sandbox/harness/worker.py", mode,
            f"/tmp/sandbox/work/{cfg_name}"]
    return cmd


def _trusted_cmd(gpu: int, mode: str, cfg_path: str) -> tuple[list[str], dict]:
    env = {"PATH": "/usr/local/cuda/bin:/usr/local/bin:/usr/bin:/bin", "HOME": "/tmp",
           "CUDA_DEVICE_ORDER": "PCI_BUS_ID", "CUDA_VISIBLE_DEVICES": str(gpu), "PYTHONDONTWRITEBYTECODE": "1",
           "TORCHINDUCTOR_CACHE_DIR": os.path.join(EVAL_ROOT, "_inductor")}
    return [SYS_PY, "-I", "-u", os.path.join(HERE, "worker.py"), mode, cfg_path], env


class Child:
    """A worker process with a line protocol on stdout and commands on stdin."""

    def __init__(self, cmd, env=None, cwd=None):
        self.p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                  env=env, cwd=cwd, text=True, bufsize=1, start_new_session=True)
        self.lines: list[tuple[str, dict]] = []
        self.err: list[str] = []
        self.cv = threading.Condition()
        self.t_out = threading.Thread(target=self._read_out, daemon=True)
        self.t_err = threading.Thread(target=self._read_err, daemon=True)
        self.t_out.start()
        self.t_err.start()

    def _read_out(self):
        for line in self.p.stdout:
            tag, _, rest = line.rstrip("\n").partition(" ")
            if tag in ("READY", "DONE", "RESULT", "CHECKED"):
                try:
                    obj = json.loads(rest)
                except Exception:  # noqa: BLE001
                    obj = {"raw": rest[:2000]}
                with self.cv:
                    self.lines.append((tag, obj))
                    self.cv.notify_all()
        with self.cv:
            self.lines.append(("EOF", {}))
            self.cv.notify_all()

    def _read_err(self):
        for line in self.p.stderr:
            self.err.append(line)
            if len(self.err) > 400:
                del self.err[:200]

    def wait_for(self, tags, timeout):
        deadline = time.time() + timeout
        with self.cv:
            while True:
                for i, (t, o) in enumerate(self.lines):
                    if t in tags or t == "EOF" or t == "RESULT":
                        del self.lines[: i + 1]
                        return t, o
                left = deadline - time.time()
                if left <= 0:
                    return "TIMEOUT", {}
                self.cv.wait(min(left, 1.0))

    def send(self, s: str):
        try:
            self.p.stdin.write(s + "\n")
            self.p.stdin.flush()
        except Exception:  # noqa: BLE001
            pass

    def stderr_tail(self, n=4000) -> str:
        return "".join(self.err)[-n:]

    def kill(self):
        try:
            os.killpg(self.p.pid, 9)
        except Exception:  # noqa: BLE001
            pass
        try:
            self.p.kill()
        except Exception:  # noqa: BLE001
            pass

    def finish(self, timeout=60):
        t, o = self.wait_for({"RESULT"}, timeout)
        if t != "RESULT":
            self.kill()
        else:
            try:
                self.p.wait(timeout=10)
            except Exception:  # noqa: BLE001
                self.kill()
        return t, o


# --------------------------------------------------------------------------------------------- GPU scheduling
class GpuSlot:
    """Readers (correctness/check phases) share a GPU; a timing phase is exclusive (writer preference)."""

    def __init__(self, index: int, max_shared: int = 4):
        self.index = index
        self.max_shared = max_shared
        self.cv = threading.Condition()
        self.readers = 0
        self.writer = False
        self.waiting_writers = 0

    def acquire_shared(self):
        with self.cv:
            while self.writer or self.waiting_writers or self.readers >= self.max_shared:
                self.cv.wait()
            self.readers += 1

    def release_shared(self):
        with self.cv:
            self.readers -= 1
            self.cv.notify_all()

    def acquire_exclusive(self, holding_shared: bool = False):
        with self.cv:
            self.waiting_writers += 1
            if holding_shared:
                self.readers -= 1
                self.cv.notify_all()
            while self.writer or self.readers:
                self.cv.wait()
            self.waiting_writers -= 1
            self.writer = True

    def release_exclusive(self):
        with self.cv:
            self.writer = False
            self.cv.notify_all()

    def load(self):
        with self.cv:
            return self.readers + 4 * (self.writer + self.waiting_writers)


COMPILE_SEM = threading.BoundedSemaphore(int(os.environ.get("EVAL_COMPILE_SLOTS", str(max(2, (os.cpu_count() or 8) // 2)))))


# --------------------------------------------------------------------------------------------- comparison
def compare(ref_file: str, cand_file: str, tol: float = TOL) -> dict:
    if not os.path.exists(cand_file):
        return {"ok": False, "why": "candidate produced no output"}
    if os.path.getsize(cand_file) > 8 * 2**30:
        return {"ok": False, "why": "candidate output file too large"}
    ref = torch.load(ref_file, weights_only=True)
    cand = torch.load(cand_file, weights_only=True)
    if not isinstance(cand, list) or len(cand) != len(ref):
        return {"ok": False, "why": f"expected {len(ref)} output tensor(s), got {len(cand) if isinstance(cand, list) else type(cand).__name__}"}
    worst = {"max_abs_diff": 0.0, "max_rel_diff": 0.0, "mismatch_frac": 0.0}
    for r, c in zip(ref, cand):
        if not isinstance(c, torch.Tensor):
            return {"ok": False, "why": "output is not a tensor"}
        if tuple(c.shape) != tuple(r.shape):
            return {"ok": False, "why": f"output shape {tuple(c.shape)} != expected {tuple(r.shape)}"}
        if c.dtype != r.dtype:
            return {"ok": False, "why": f"output dtype {c.dtype} != expected {r.dtype}"}
        rf = r.double()
        cf = c.double()
        fin = torch.isfinite(rf)
        if not bool(torch.isfinite(cf[fin]).all()):
            return {"ok": False, "why": "non-finite values where the reference is finite"}
        d = (cf - rf).abs()
        d = torch.where(fin, d, torch.zeros_like(d))
        bad = d > (tol + tol * rf.abs())
        worst["max_abs_diff"] = max(worst["max_abs_diff"], float(d.max()) if d.numel() else 0.0)
        rel = d / (rf.abs() + 1e-12)
        worst["max_rel_diff"] = max(worst["max_rel_diff"], float(torch.where(fin, rel, torch.zeros_like(rel)).max()) if d.numel() else 0.0)
        worst["mismatch_frac"] = max(worst["mismatch_frac"], float(bad.double().mean()) if d.numel() else 0.0)
    worst = {k: float(f"{v:.3g}") for k, v in worst.items()}
    return {"ok": worst["mismatch_frac"] == 0.0, **worst,
            "why": "" if worst["mismatch_frac"] == 0.0 else f"output mismatch beyond atol=rtol={tol}"}


# --------------------------------------------------------------------------------------------- baselines
_BASE_PATH = os.environ.get("EVAL_BASELINES", "baselines_a100.json")
_BASE_CACHE = {"mtime": None, "data": {}}


def baselines() -> dict:
    """torch.compile / eager baselines per problem (same GPU type and timing method), re-read when the file changes."""
    try:
        m = os.path.getmtime(_BASE_PATH)
    except OSError:
        return {}
    if _BASE_CACHE["mtime"] != m:
        try:
            _BASE_CACHE["data"] = json.load(open(_BASE_PATH)).get("problems", {})
            _BASE_CACHE["mtime"] = m
        except Exception:  # noqa: BLE001
            return _BASE_CACHE["data"]
    return _BASE_CACHE["data"]


def baseline(pid: str, gpu: "GpuSlot", blocks: int = BLOCKS) -> dict:
    """Trusted only: time torch eager vs torch.compile(Model) with the evaluator's protocol (fresh inputs per
    trial, cold L2, guarded timing, interleaved blocks under an exclusive GPU lock)."""
    t0 = time.time()
    ppath = problem_path(pid)
    root = os.path.join(EVAL_ROOT, "_baseline_" + uuid.uuid4().hex)
    os.makedirs(root, exist_ok=True)
    out = {"problem": pid, "evaluator": fingerprint()}
    kids = []
    holding = None
    try:
        tb = _seed()
        chs = {}
        for tag, comp in (("eager", False), ("compile", True)):
            cfg = {"problem_path": ppath, "init_seed": 42, "seeds": {}, "time_base": tb, "out_dir": root,
                   "cudnn_tf32": CUDNN_TF32, "warmup": 3, "torch_compile": comp}
            cp = os.path.join(root, f"cfg-{tag}.json")
            json.dump(cfg, open(cp, "w"))
            cmd, env = _trusted_cmd(gpu.index, "ref_time", cp)
            chs[tag] = Child(cmd, env=env)
            kids.append(chs[tag])
        gpu.acquire_shared()
        holding = "shared"
        ready = {}
        for tag, ch in chs.items():
            t, o = ch.wait_for({"READY"}, 1200)
            if t != "READY":
                out.update(ok=False, error=f"{tag}: {o.get('runtime_error') or t}"[:500])
                return out
            ready[tag] = o
        est = max(float(ready["eager"]["est_ms"]), float(ready["compile"]["est_ms"]), 1e-3)
        n_total = int(min(MAX_TRIALS, max(MIN_TRIALS, math.ceil(TIME_BUDGET_MS / est))))
        per = max(3, n_total // blocks)
        gpu.acquire_exclusive(holding_shared=True)
        holding = "exclusive"
        for _ in range(blocks):
            for ch in chs.values():
                ch.send(f"GO {per}")
                t, _o = ch.wait_for({"DONE"}, 300)
                if t != "DONE":
                    out.update(ok=False, error="timing block failed")
                    return out
        res = {}
        for tag, ch in chs.items():
            ch.send("EXIT")
            t, o = ch.finish(120)
            if t != "RESULT" or not o.get("evt_ms"):
                out.update(ok=False, error=f"{tag}: no timing result")
                return out
            res[tag] = o
        gpu.release_exclusive()
        holding = None
        import common as C
        for tag, o in res.items():
            g = C.guarded(o["evt_ms"], o["wall_ms"], o["overhead_ms"])
            out[f"{tag}_ms"] = round(C.median(g), 5)
            out[f"{tag}_ms_mean"] = round(C.mean(g), 5)
        out["compile_speedup_over_eager"] = round(out["eager_ms"] / out["compile_ms"], 4)
        out["ok"] = True
        return out
    except Exception as e:  # noqa: BLE001
        out.update(ok=False, error=f"{type(e).__name__}: {e}"[:500])
        return out
    finally:
        for c in kids:
            c.kill()
        if holding == "shared":
            gpu.release_shared()
        elif holding == "exclusive":
            gpu.release_exclusive()
        out["wall_s"] = round(time.time() - t0, 1)
        shutil.rmtree(root, ignore_errors=True)

_PUBLIC_LOCK = threading.Lock()


def _public_ref(pid: str, gpu: GpuSlot) -> tuple[str, dict]:
    """Reference output on the public seed, computed once per problem by a trusted child (cached in RAM disk)."""
    d = os.path.join(EVAL_ROOT, "_public", pid)
    f = os.path.join(d, "public.pt")
    meta_f = os.path.join(d, "meta.json")
    with _PUBLIC_LOCK:
        if os.path.exists(f) and os.path.exists(meta_f):
            return f, json.load(open(meta_f))
        os.makedirs(d, exist_ok=True)
        cfg = {"problem_path": problem_path(pid), "init_seed": PUBLIC_SEED, "seeds": {"public": PUBLIC_SEED},
               "out_dir": d, "cudnn_tf32": CUDNN_TF32}
        cp = os.path.join(d, "cfg.json")
        json.dump(cfg, open(cp, "w"))
        cmd, env = _trusted_cmd(gpu.index, "ref_outputs", cp)
        ch = Child(cmd, env=env)
        t, o = ch.finish(RUN_TIMEOUT)
        if t != "RESULT" or not o.get("ok"):
            raise RuntimeError(f"reference failed on public seed: {o.get('runtime_error') or t} {ch.stderr_tail(800)}")
        json.dump(o, open(meta_f, "w"))
        return f, o


# --------------------------------------------------------------------------------------------- evaluate
def evaluate(pid: str, source: str, gpu: GpuSlot, mode: str = "eval", skip_static: bool = False) -> dict:
    """skip_static is for qualification only (proves the runtime layers catch what the static gate would)."""
    t0 = time.time()
    res: dict = {"problem": pid, "mode": mode, "evaluator": fingerprint(), "tol": TOL,
                 "source_sha256": hashlib.sha256(source.encode()).hexdigest(), "gpu": gpu.index, "phases": {}}

    def done(status, **kw):
        res.update(status=status, **kw)
        res.setdefault("correct", False)
        res.setdefault("fast1", False)
        res.setdefault("speedup", 0.0)
        res["wall_s"] = round(time.time() - t0, 2)
        return res

    try:
        ppath = problem_path(pid)
    except Exception as e:  # noqa: BLE001
        return done("evaluator_error", error=f"problem: {e}")
    if len(source) > 400_000:
        return done("forbidden", error="source too large")
    st = static_check.check_source(source)
    res["static"] = {"backend": st["backend"], "kernels": st["kernels"]}
    if not st["ok"] and not skip_static:
        return done("forbidden", error="static check failed: " + "; ".join(st["errors"]))
    if skip_static:
        res["static_skipped"] = True

    work_root = os.path.join(EVAL_ROOT, uuid.uuid4().hex)
    work = os.path.join(work_root, "work")
    refd = os.path.join(work_root, "ref")
    for d in (work, refd, os.path.join(work, "out"), os.path.join(work, "tmp")):
        os.makedirs(d, exist_ok=True)
    children: list[Child] = []
    holding = None
    try:
        with open(os.path.join(work, "cand.py"), "w") as f:
            f.write(source)
        shutil.copy(ppath, os.path.join(work, "problem.py"))
        base_cfg = {"problem_path": "/tmp/sandbox/work/problem.py", "cand_path": "/tmp/sandbox/work/cand.py",
                    "cudnn_tf32": CUDNN_TF32, "out_dir": "/tmp/sandbox/work/out"}

        # ---- compile (sandbox, no GPU)
        tc = time.time()
        json.dump(base_cfg, open(os.path.join(work, "cfg-compile.json"), "w"))
        with COMPILE_SEM:
            ch = Child(_sandbox_cmd(work, None, "cand_compile", "cfg-compile.json"))
            children.append(ch)
            t, o = ch.finish(COMPILE_TIMEOUT)
        res["phases"]["compile_s"] = round(time.time() - tc, 2)
        compile_log = ch.stderr_tail(6000)
        if t == "TIMEOUT":
            return done("timeout", error=f"compile exceeded {COMPILE_TIMEOUT}s")
        if not o.get("ok"):
            err = (o.get("runtime_error") or "") + "\n" + compile_log
            if "Error building extension" in err or "SyntaxError" in err or "NameError" in err \
                    or "ImportError" in err or "AttributeError" in err or "ModelNew not defined" in err:
                return done("compile_error", error=_clip(o.get("runtime_error") or "compile failed"),
                            compile_log=_clip(compile_log, 5000))
            res["compile_deferred"] = _clip(o.get("runtime_error") or "", 500)
        banned = []
        for fp in glob.glob(os.path.join(work, "ext", "**", "*"), recursive=True):
            if fp.endswith((".cpp", ".cu", ".cuh", ".h", ".hpp", ".cc")) and os.path.isfile(fp):
                banned += static_check.scan_native(open(fp, errors="replace").read())
        if banned:
            return done("forbidden", error=f"build sources contain banned tokens {sorted(set(banned))}")

        # ---- seeds
        if mode == "check":
            seeds = {"public": PUBLIC_SEED}
            init_seed = PUBLIC_SEED
            time_base = None
        else:
            seeds = {f"c{i}": _seed() for i in range(N_CORRECT)}
            init_seed = _seed()
            time_base = _seed()
        res["seeds"] = {"init": init_seed, **seeds, "time_base": time_base}

        gpu.acquire_shared()
        holding = "shared"
        tr = time.time()
        # reference outputs (trusted) in parallel with candidate correctness (sandbox)
        if mode == "check":
            ref_file, _meta = _public_ref(pid, gpu)
            ref_files = {"public": ref_file}
            rch = None
        else:
            rcfg = {"problem_path": ppath, "init_seed": init_seed, "seeds": seeds,
                    "out_dir": refd, "cudnn_tf32": CUDNN_TF32}
            json.dump(rcfg, open(os.path.join(refd, "cfg.json"), "w"))
            cmd, env = _trusted_cmd(gpu.index, "ref_outputs", os.path.join(refd, "cfg.json"))
            rch = Child(cmd, env=env)
            children.append(rch)
            ref_files = {k: os.path.join(refd, f"{k}.pt") for k in seeds}
        warm = 2 + (os.urandom(1)[0] % 3)
        ccfg = dict(base_cfg, init_seed=init_seed, seeds=seeds, kernel_names=st["kernels"],
                    time_base=time_base, warmup=warm)
        json.dump(ccfg, open(os.path.join(work, "cfg-run.json"), "w"))
        cch = Child(_sandbox_cmd(work, gpu.index, "cand_run", "cfg-run.json"))
        children.append(cch)

        if rch is not None:
            t, ro = rch.finish(RUN_TIMEOUT)
            if t != "RESULT" or not ro.get("ok"):
                return done("evaluator_error", error=f"reference failed: {ro.get('runtime_error') or t}",
                            detail=_clip(rch.stderr_tail(), 2000))
        t, co = cch.wait_for({"CHECKED"}, RUN_TIMEOUT)
        res["phases"]["correctness_s"] = round(time.time() - tr, 2)
        if t == "TIMEOUT":
            return done("timeout", error=f"candidate correctness phase exceeded {RUN_TIMEOUT}s")
        if t in ("RESULT", "EOF"):
            if t == "EOF":
                co = {"runtime_error": "candidate process died", "detail": cch.stderr_tail(3000)}
            if co.get("tamper"):
                return done("tampered", error="; ".join(co["tamper"]))
            rt = co.get("runtime_error") or "candidate failed"
            if "Error building extension" in rt:
                return done("compile_error", error=_clip(rt), compile_log=_clip(cch.stderr_tail(5000), 5000))
            return done("runtime_error", error=_clip(rt), detail=_clip(cch.stderr_tail(3000), 3000))
        if co.get("tamper"):
            cch.send("EXIT")
            return done("tampered", error="; ".join(co["tamper"]))
        res["own_kernels_ran"] = co.get("own_kernels_ran")
        res["own_kernel_time_frac"] = co.get("own_kernel_time_frac")
        cmp = {}
        for tag in seeds:
            cmp[tag] = compare(ref_files[tag], os.path.join(work, "out", f"{tag}.pt"))
        res["correctness"] = cmp
        bad = [k for k, v in cmp.items() if not v["ok"]]
        if bad:
            cch.send("EXIT")
            return done("incorrect", error=cmp[bad[0]]["why"], **{k: cmp[bad[0]].get(k) for k in
                                                                     ("max_abs_diff", "max_rel_diff", "mismatch_frac")})
        if not co.get("own_kernels_ran"):
            cch.send("EXIT")
            return done("no_custom_kernel", error=f"none of the candidate's kernels {st['kernels']} ran on the GPU "
                                                  f"during forward (torch-only paths are not a kernel)")
        if mode == "check":
            t, o = cch.finish(60)
            if o.get("tamper"):
                return done("tampered", error="; ".join(o["tamper"]))
            return done("ok", correct=True, max_abs_diff=cmp["public"].get("max_abs_diff"))

        # ---- timing: reference timing child (trusted), interleaved with the candidate under an exclusive lock
        tcfg = {"problem_path": ppath, "init_seed": init_seed, "seeds": {}, "time_base": time_base,
                "out_dir": refd, "cudnn_tf32": CUDNN_TF32, "warmup": warm}
        json.dump(tcfg, open(os.path.join(refd, "cfg-time.json"), "w"))
        cmd, env = _trusted_cmd(gpu.index, "ref_time", os.path.join(refd, "cfg-time.json"))
        tch = Child(cmd, env=env)
        children.append(tch)
        t1, r_ready = tch.wait_for({"READY"}, RUN_TIMEOUT)
        t2, c_ready = cch.wait_for({"READY"}, RUN_TIMEOUT)
        if t1 != "READY":
            return done("evaluator_error", error=f"reference timing child failed: {r_ready.get('runtime_error') or t1}")
        if t2 != "READY":
            if t2 == "TIMEOUT":
                return done("timeout", error="candidate warmup exceeded the time limit")
            if c_ready.get("tamper"):
                return done("tampered", error="; ".join(c_ready["tamper"]))
            return done("runtime_error", error=_clip(c_ready.get("runtime_error") or "candidate died before timing"))
        est = max(float(r_ready["est_ms"]), float(c_ready["est_ms"]), 1e-3)
        n_total = int(min(MAX_TRIALS, max(MIN_TRIALS, math.ceil(TIME_BUDGET_MS / est))))
        if float(c_ready["est_ms"]) > 50 * float(r_ready["est_ms"]) + 50:
            n_total = MIN_TRIALS
        base_block = max(3, n_total // BLOCKS)
        # random block sizes: the candidate cannot know which trials end a block (those outputs are verified)
        sizes = [max(2, int(round(base_block * (0.7 + 0.6 * os.urandom(1)[0] / 255.0)))) for _ in range(BLOCKS)]
        tt = time.time()
        gpu.acquire_exclusive(holding_shared=True)
        holding = "exclusive"
        deadline = time.time() + TIMING_TIMEOUT
        timed_out = False
        for nb in sizes:
            for chh in (tch, cch):
                chh.send(f"GO {nb}")
                t, o = chh.wait_for({"DONE"}, max(1.0, deadline - time.time()))
                if t != "DONE":
                    timed_out = True
                    break
            if timed_out:
                break
        tch.send("EXIT")
        cch.send("EXIT")
        t_r, r_res = tch.finish(120)
        t_c, c_res = cch.finish(180)
        gpu.release_exclusive()
        holding = None
        res["phases"]["timing_s"] = round(time.time() - tt, 2)
        if timed_out:
            if t_c != "RESULT":
                return done("timeout", error=f"timing exceeded {TIMING_TIMEOUT}s")
        if t_r != "RESULT" or not r_res.get("ok"):
            return done("evaluator_error", error=f"reference timing failed: {r_res.get('runtime_error') or t_r}")
        if t_c != "RESULT":
            return done("runtime_error", error="candidate died during timing")
        if c_res.get("tamper"):
            return done("tampered", error="; ".join(c_res["tamper"]))
        if c_res.get("runtime_error"):
            return done("runtime_error", error=_clip(c_res["runtime_error"]))
        tcmp = {f"blk{b}": compare(os.path.join(refd, f"blk{b}.pt"), os.path.join(work, "out", f"blk{b}.pt"))
                for b in range(min(int(r_res.get("blocks", 0)), int(c_res.get("blocks", 0))))}
        if len(tcmp) < BLOCKS:
            return done("evaluator_error", error=f"only {len(tcmp)} timing blocks completed")
        res["timing_correctness"] = tcmp
        bad = [k for k, v in tcmp.items() if not v["ok"]]
        if bad:
            return done("incorrect", error="outputs produced during timing do not match the reference "
                                           f"({tcmp[bad[0]]['why']}); results must not be cached or skipped")
        import common as C
        r_t = C.guarded(r_res["evt_ms"], r_res["wall_ms"], r_res["overhead_ms"])
        c_t = C.guarded(c_res["evt_ms"], c_res["wall_ms"], c_res["overhead_ms"])
        ref_ms, cand_ms = C.median(r_t), C.median(c_t)
        sp = ref_ms / cand_ms if cand_ms > 0 else 0.0
        lo = C.bootstrap_ratio_lo(r_t, c_t)
        gap_r = C.median([w - e for e, w in zip(r_res["evt_ms"], r_res["wall_ms"])])
        gap_c = C.median([w - e for e, w in zip(c_res["evt_ms"], c_res["wall_ms"])])
        evt_sp = C.median(r_res["evt_ms"]) / max(1e-9, C.median(c_res["evt_ms"]))
        res.update(ref_ms=round(ref_ms, 5), cand_ms=round(cand_ms, 5), speedup=round(sp, 4),
                   speedup_lo95=round(lo, 4), speedup_mean=round(C.mean(r_t) / max(1e-9, C.mean(c_t)), 4),
                   speedup_events_only=round(evt_sp, 4), n_trials=len(c_t),
                   wall_minus_event_ms={"ref": round(gap_r, 4), "cand": round(gap_c, 4)})
        flags = []
        if evt_sp > 1.25 * sp and gap_c > gap_r + 0.05:
            flags.append("side_stream_suspected: event-only speedup much larger than device-synchronised speedup")
        if sp > EXCESSIVE_SPEEDUP:
            flags.append(f"excessive_speedup>{EXCESSIVE_SPEEDUP}x: audit")
        res["flags"] = flags
        b = baselines().get(pid) or {}
        if b.get("compile_ms"):
            res["speedup_vs_compile"] = round(float(b["compile_ms"]) / cand_ms, 4)
            res["compile_ms_baseline"] = b["compile_ms"]
        return done("ok", correct=True, fast1=bool(lo >= 1.0), fast1_raw=bool(sp >= 1.0))
    except Exception as e:  # noqa: BLE001
        import traceback
        return done("evaluator_error", error=f"{type(e).__name__}: {e}", detail=traceback.format_exc()[-2000:])
    finally:
        for c in children:
            c.kill()
        if holding == "shared":
            gpu.release_shared()
        elif holding == "exclusive":
            gpu.release_exclusive()
        if os.environ.get("EVAL_KEEP_EVAL") != "1":
            shutil.rmtree(work_root, ignore_errors=True)


def _clip(s: str, n: int = 3000) -> str:
    s = s or ""
    return s if len(s) <= n else s[: n // 3] + "\n...\n" + s[-(2 * n) // 3:]


if __name__ == "__main__":
    import sys
    pid, path = sys.argv[1], sys.argv[2]
    mode = sys.argv[3] if len(sys.argv) > 3 else "eval"
    gi = int(os.environ.get("EVAL_GPU", "0"))
    out = evaluate(pid, open(path).read(), GpuSlot(gi), mode)
    print(json.dumps(out, indent=1, default=str))