#!/usr/bin/env python3
"""Client-perf bench harness v2: no input injection, one launch per config session.

    bench.py boot    --n 3 [--variant C1] [--join]          # cold-launch reliability + startup time
    bench.py session --variant C1 --plan s1,s2,s3 --reps 3 [--label B2] [--dry-run]
    bench.py disarm                                          # remove a leftover autoplan datapack
    bench.py meta                                            # print the environment snapshot

`session` applies the variant, pins options.txt, writes a per-session datapack whose
`#minecraft:load` hook waits for the player and then drives every scene x rep itself,
printing `BENCH_MARK warm|start|end <scene> <rep>` via /say. The harness only tails
latest.log for those markers, watches for crashes, kills the client when the plan is
done, slices the MangoHud CSV per window and appends one JSON row per window to
~/onion-bench/results/session.jsonl. Stdlib only; see README.md.
"""
import argparse
import datetime as dt
import hashlib
import json
import os
import re
import shutil
import signal
import subprocess
import sys
import time
from pathlib import Path

HERE = Path(__file__).resolve().parent
REPO = HERE.parents[2]
PROPOSED = HERE.parent / "proposed-configs"
BENCH = Path.home() / "onion-bench"
PRISM = BENCH / "prism"
INSTANCE = "OnionBench"
INST_DIR = PRISM / "instances" / INSTANCE
GAME = INST_DIR / ".minecraft"
WORLD = "onionbench"
MANGOLOGS = BENCH / "mangologs"
RESULTS = BENCH / "results"
PLAN_PACK = "onion-bench-autoplan"
MACHINE = "proxy-9900X-7800XT"

PINS = [
    ("onboardAccessibility", "false"),
    ("pauseOnLostFocus", "false"),
    ("tutorialStep", "none"),
    ("narrator", "0"),
    ("skipMultiplayerWarning", "true"),
    ("joinedFirstServer", "true"),
    ("maxFps", "260"),
    ("enableVsync", "false"),
]

# BENCH-MATRIX §3: one session runs its scenes in this order (S1 first).
SCENE_ORDER = ["s1", "s2", "s2s", "s3", "s4", "s5"]
SCENE_FN = {"s1": "bench:s1", "s2": "bench:s2", "s2s": "bench:s2", "s3": "bench:s3",
            "s4": "bench:s4", "s5": "bench:s1"}
S5_PREWAIT, S5_CAPTURE = 10, 120  # S5: at S1, wait 10 s, tour; all 120 s count

# BENCH-MATRIX §1 JVM sets (applied to instance.cfg only when --jvm is given).
JVM_SETS = {
    "J0": "-XX:+UseG1GC -XX:+ParallelRefProcEnabled -XX:MaxGCPauseMillis=200 "
          "-XX:+UnlockExperimentalVMOptions -XX:+DisableExplicitGC -XX:G1NewSizePercent=30 "
          "-XX:G1MaxNewSizePercent=40 -XX:G1HeapRegionSize=8M -XX:G1ReservePercent=20 "
          "-XX:G1HeapWastePercent=5 -XX:G1MixedGCCountTarget=4 "
          "-XX:InitiatingHeapOccupancyPercent=15 -XX:G1MixedGCLiveThresholdPercent=90 "
          "-XX:G1RSetUpdatingPauseTimePercent=5 -XX:SurvivorRatio=32 "
          "-XX:+PerfDisableSharedMem -XX:MaxTenuringThreshold=1",
    "J1": "-XX:+UseG1GC -XX:MaxGCPauseMillis=37 -XX:G1HeapRegionSize=16M "
          "-XX:+UnlockExperimentalVMOptions -XX:G1NewSizePercent=23 -XX:G1ReservePercent=20 "
          "-XX:G1MixedGCCountTarget=3 -XX:G1HeapWastePercent=20 "
          "-XX:InitiatingHeapOccupancyPercent=10 -XX:G1RSetUpdatingPauseTimePercent=0 "
          "-XX:MaxTenuringThreshold=1 -XX:GCTimeRatio=99 -XX:+DisableExplicitGC "
          "-XX:+PerfDisableSharedMem",
    "J2": "-XX:+UseZGC -XX:+ZGenerational -XX:+DisableExplicitGC -XX:+PerfDisableSharedMem",
    "J3": "-XX:+UseShenandoahGC -XX:+DisableExplicitGC -XX:+PerfDisableSharedMem",
}

LOG_TS = re.compile(r"^\[(\d{2})([A-Za-z]{3})(\d{4}) (\d{2}):(\d{2}):(\d{2})\.(\d{3})\]")
MONTHS = {m: i + 1 for i, m in enumerate(
    ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"])}
MARK_RE = re.compile(r"BENCH_MARK (warm|start|end) (\w+) (\d+)\b")
DONE_RE = re.compile(r"BENCH_DONE (\w+)")
GAME_TOOK_RE = re.compile(r"Game took ([\d.]+) seconds to start")
JOINED_RE = re.compile(r"\]: \S+ joined the game")
ATLAS_RE = re.compile(r"Created: (\d+x\d+x\d+) minecraft:textures/atlas/blocks\.png-atlas")
CRASH_RES = [
    ("mod_instance", re.compile(r"Failed to create mod instance")),
    ("fatal", re.compile(r"\[[^\]]*/FATAL\]")),
    ("broken_mod_state", re.compile(r"Cowardly refusing")),
    ("crash_report", re.compile(r"---- Minecraft Crash Report ----|Crash report saved to")),
]
BACKUP_RE = re.compile(r"Server Backup started|backup starting|Saving chunks|Autosav",
                       re.IGNORECASE)


def log(msg):
    print(f"[bench] {msg}", flush=True)


# ---------------------------------------------------------------- pure helpers (tested)

def log_time(line):
    """Wall-clock epoch of a Minecraft log line ('[15Oct2026 15:25:53.147] ...'), or None."""
    m = LOG_TS.match(line)
    if not m or m.group(2) not in MONTHS:
        return None
    d, mon, y, hh, mm, ss, ms = m.groups()
    t = dt.datetime(int(y), MONTHS[mon], int(d), int(hh), int(mm), int(ss))
    return time.mktime(t.timetuple()) + int(ms) / 1000.0


def parse_markers(lines):
    """Return ({(kind, scene, rep): epoch}, done_epoch). First occurrence wins (the
    server line precedes the client [CHAT] echo)."""
    marks, done = {}, None
    for line in lines:
        m = MARK_RE.search(line)
        if m:
            key = (m.group(1), m.group(2), int(m.group(3)))
            t = log_time(line)
            if t is not None and key not in marks:
                marks[key] = t
            continue
        if done is None and DONE_RE.search(line):
            done = log_time(line)
    return marks, done


def classify_line(line):
    """Crash class for one log line, or None."""
    for name, rx in CRASH_RES:
        if rx.search(line):
            return name
    return None


def scan_events(lines):
    """Startup facts from log lines: game_took_s, title/joined epochs, atlas, crash."""
    out = {"game_took_s": None, "title_t": None, "joined_t": None, "atlas": None,
           "crash": None, "crash_line": None}
    for line in lines:
        if out["game_took_s"] is None:
            m = GAME_TOOK_RE.search(line)
            if m:
                out["game_took_s"] = float(m.group(1))
                out["title_t"] = log_time(line)
        if out["joined_t"] is None and JOINED_RE.search(line):
            out["joined_t"] = log_time(line)
        if out["atlas"] is None:
            m = ATLAS_RE.search(line)
            if m:
                out["atlas"] = m.group(1)
        if out["crash"] is None:
            c = classify_line(line)
            if c:
                out["crash"], out["crash_line"] = c, line.strip()[:300]
    return out


def backup_times(lines):
    return [t for t in (log_time(l) for l in lines if BACKUP_RE.search(l)) if t is not None]


def parse_mangohud_csv(path):
    """Return (meta, rows). meta = the spec line (os/cpu/gpu/...) as a dict; rows =
    [(elapsed_s, frametime_ms, fps)]. The column header is located by name."""
    lines = Path(path).read_text(errors="replace").splitlines()
    hdr_i = None
    for i, line in enumerate(lines[:10]):
        cols = line.split(",")
        if "frametime" in cols and "elapsed" in cols:
            hdr_i = i
            break
    if hdr_i is None:
        raise ValueError(f"no frametime/elapsed header in {path}")
    meta = {}
    if hdr_i >= 2:
        meta = dict(zip(lines[hdr_i - 2].split(","), lines[hdr_i - 1].split(",")))
    cols = lines[hdr_i].split(",")
    e_i, ft_i, fps_i = cols.index("elapsed"), cols.index("frametime"), cols.index("fps")
    rows = []
    for line in lines[hdr_i + 1:]:
        parts = line.split(",")
        if len(parts) != len(cols):
            continue
        try:
            rows.append((float(parts[e_i]) / 1e9, float(parts[ft_i]), float(parts[fps_i])))
        except ValueError:
            continue
    return meta, rows


def csv_name_time(path):
    """Epoch from MangoHud's '<proc>_YYYY-MM-DD_HH-MM-SS.csv' name (local time)."""
    m = re.search(r"_(\d{4})-(\d{2})-(\d{2})_(\d{2})-(\d{2})-(\d{2})\.csv$", str(path))
    if not m:
        return None
    return time.mktime(dt.datetime(*map(int, m.groups())).timetuple())


def file_birth(path):
    """(epoch, method) for when the file was created; falls back to the name stamp."""
    st = os.stat(path)
    bt = getattr(st, "st_birthtime", None)
    if bt:
        return float(bt), "st_birthtime"
    try:
        out = subprocess.run(["stat", "-c", "%.9W", str(path)], capture_output=True,
                             text=True, timeout=5).stdout.strip()
        if out and out not in ("0", "-") and float(out) > 0:
            return float(out), "stat-W"
    except (OSError, ValueError, subprocess.SubprocessError):
        pass
    nt = csv_name_time(path)
    if nt is not None:
        return nt, "filename"
    return st.st_mtime, "mtime"


def pct(values, p):
    """Linear-interpolated percentile of an unsorted list, p in [0, 1]."""
    if not values:
        return float("nan")
    s = sorted(values)
    k = (len(s) - 1) * p
    lo, hi = int(k), min(int(k) + 1, len(s) - 1)
    return s[lo] + (s[hi] - s[lo]) * (k - lo)


def window_metrics(rows, origin, t_start, t_end):
    """Metrics of rows whose wall time (origin + elapsed_s) is in [t_start, t_end]."""
    ft, fps = [], []
    for e, frametime, f in rows:
        wall = origin + e
        if t_start <= wall <= t_end:
            ft.append(frametime)
            fps.append(f)
    if not ft:
        return None
    n = len(ft)
    k = max(1, int(n * 0.01))
    worst_ft = sorted(ft, reverse=True)[:k]
    worst_fps = sorted(fps)[:k]
    mean_ft = sum(ft) / n
    return {
        "n": n,
        "p50_ms": round(pct(ft, 0.50), 3),
        "p95_ms": round(pct(ft, 0.95), 3),
        "p99_ms": round(pct(ft, 0.99), 3),
        "peak_ms": round(max(ft), 3),
        "avg_fps": round(1000.0 / mean_ft, 2) if mean_ft > 0 else None,
        "one_low_fps": round(1000.0 / (sum(worst_ft) / k), 2),
        "one_low_fps_legacy": round(sum(worst_fps) / k, 2),
        "coverage_s": round(min(t_end, origin + rows[-1][0]) - max(t_start, origin + rows[0][0]), 2),
    }


def validity_flags(m, duration_s, refresh_rates=(60.0,), backups=(), t_start=0.0, t_end=0.0):
    """Warning flags for one window. m may be None (no frames)."""
    flags = []
    if m is None:
        return ["no_data"]
    for hz in set([60.0, *refresh_rates]):
        if hz > 0 and abs(m["p50_ms"] - 1000.0 / hz) <= 0.03 * (1000.0 / hz):
            flags.append("vsync_suspect")
            break
    expected = 0.5 * duration_s * (m["avg_fps"] or 0)
    if m["n"] < 500 or m["n"] < expected:
        flags.append("low_n")
    if any(t_start - 5 <= b <= t_end for b in backups):
        flags.append("backup_in_window")
    return flags


def build_timeline(plan, reps, warmup_s, capture_s, sid):
    """Datapack timeline: list of (commands, delay_ticks_after). Pure; tested."""
    tl = []
    for scene in plan:
        for rep in range(1, reps + 1):
            go = f"execute as @a[limit=1] at @s run function {SCENE_FN[scene]}"
            if scene == "s5":
                tl.append(([go, f"say BENCH_MARK warm s5 {rep}"], S5_PREWAIT * 20))
                tl.append(([f"execute as @a[limit=1] at @s run function bench:tour_start",
                            f"say BENCH_MARK start s5 {rep}"], S5_CAPTURE * 20))
                tl.append(([f"say BENCH_MARK end s5 {rep}", "function bench:tour_stop"], 0))
                continue
            pre = [go]
            if scene == "s2s":
                pre.append("function bench:storm_on")
            tl.append((pre + [f"say BENCH_MARK warm {scene} {rep}"], max(1, warmup_s * 20)))
            tl.append(([f"say BENCH_MARK start {scene} {rep}"], capture_s * 20))
            post = [f"say BENCH_MARK end {scene} {rep}"]
            if scene == "s2s":
                post.append("function bench:storm_off")
            tl.append((post, 0))
    tl.append(([f"say BENCH_DONE {sid}"], 0))
    # merge zero-delay steps into the next one (schedule needs > 0 ticks)
    merged, buf = [], []
    for cmds, delay in tl:
        buf.extend(cmds)
        if delay > 0:
            merged.append((buf, delay))
            buf = []
    if buf:
        merged.append((buf, 0))
    return merged


def plan_seconds(plan, reps, warmup_s, capture_s):
    total = 0
    for scene in plan:
        total += reps * ((S5_PREWAIT + S5_CAPTURE) if scene == "s5" else (warmup_s + capture_s))
    return total


def generate_datapack(sid, plan, reps, warmup_s, capture_s, settle_s):
    """Return {relative path: text} for the per-session autoplan datapack."""
    ns = f"bench:autoplan/{sid}"
    holder = f"#{sid}"
    files = {
        "pack.mcmeta": json.dumps({"pack": {
            "pack_format": 48,
            "description": f"Onion bench autoplan {sid} (generated by bench.py; "
                           f"deleted at session end)"}}) + "\n",
        "data/minecraft/tags/function/load.json":
            json.dumps({"values": [f"{ns}/load"]}) + "\n",
        f"data/bench/function/autoplan/{sid}/load.mcfunction":
            "# armed by bench.py for one session; the pack is removed afterwards\n"
            "scoreboard objectives add bench_plan dummy\n"
            f"execute unless score {holder} bench_plan matches 1.. run "
            f"schedule function {ns}/poll 20t\n",
        f"data/bench/function/autoplan/{sid}/poll.mcfunction":
            f"execute if entity @a run function {ns}/begin\n"
            f"execute unless entity @a run schedule function {ns}/poll 20t\n",
        f"data/bench/function/autoplan/{sid}/begin.mcfunction":
            f"scoreboard players set {holder} bench_plan 1\n"
            "function bench:storm_off\n"
            "function bench:tour_stop\n"
            f"say BENCH_PLAN {sid} begin\n"
            f"schedule function {ns}/step0 {max(1, settle_s * 20)}t\n",
    }
    tl = build_timeline(plan, reps, warmup_s, capture_s, sid)
    for i, (cmds, delay) in enumerate(tl):
        body = list(cmds)
        if delay > 0 and i + 1 < len(tl):
            body.append(f"schedule function {ns}/step{i + 1} {delay}t")
        files[f"data/bench/function/autoplan/{sid}/step{i}.mcfunction"] = "\n".join(body) + "\n"
    return files


def merge_pins(text, pins=PINS):
    """Apply key:value pins to options.txt text, keeping every other line."""
    want = dict(pins)
    out, seen = [], set()
    for line in text.splitlines():
        k = line.split(":", 1)[0]
        if k in want and ":" in line:
            line = f"{k}:{want[k]}"
            seen.add(k)
        out.append(line)
    out += [f"{k}:{v}" for k, v in pins if k not in seen]
    return "\n".join(out) + "\n"


# ---------------------------------------------------------------- environment

def bench_env():
    uid = os.getuid()
    run = Path(f"/run/user/{uid}")
    env = dict(os.environ)
    env.setdefault("DISPLAY", ":1")
    if not env.get("XAUTHORITY"):
        cands = sorted(run.glob("xauth_*"), key=lambda p: p.stat().st_mtime)
        if cands:
            env["XAUTHORITY"] = str(cands[-1])
    env.setdefault("XDG_RUNTIME_DIR", str(run))
    local_bin = str(Path.home() / ".local/bin")
    if local_bin not in env.get("PATH", "").split(":"):
        env["PATH"] = local_bin + ":" + env.get("PATH", "")
    return env


def sha256_file(p):
    h = hashlib.sha256()
    with open(p, "rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def tree_hash(root, skip=lambda rel: False):
    h = hashlib.sha256()
    n = 0
    for p in sorted(Path(root).rglob("*")):
        if not p.is_file():
            continue
        rel = p.relative_to(root).as_posix()
        if skip(rel):
            continue
        h.update(rel.encode() + b"\0" + sha256_file(p).encode() + b"\n")
        n += 1
    return h.hexdigest(), n


def config_skip(rel):
    """Runtime junk that changes without a config change (spark buffers, backups)."""
    return rel.endswith((".bak", ".tmp", ".jfr", ".lock", ".log")) or "/tmp" in "/" + rel


def mods_hash(mods_dir):
    h = hashlib.sha256()
    files = sorted(p for p in Path(mods_dir).glob("*") if p.is_file())
    for p in files:
        h.update(f"{p.name}\0{p.stat().st_size}\n".encode())
    return h.hexdigest(), len(files)


def read_text(p, default=None):
    try:
        return Path(p).read_text().strip()
    except OSError:
        return default


def read_cfg(path):
    out = {}
    for line in (read_text(path, "") or "").splitlines():
        if "=" in line and not line.startswith("["):
            k, v = line.split("=", 1)
            out[k.strip()] = v.strip()
    return out


def meminfo_available_mb():
    for line in (read_text("/proc/meminfo", "") or "").splitlines():
        if line.startswith("MemAvailable:"):
            return int(line.split()[1]) // 1024
    return None


def proc_cmdline(pid):
    try:
        return Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0")
    except OSError:
        return []


def java_procs():
    """[(pid, [args])] for every process whose comm is java."""
    out = []
    for d in Path("/proc").iterdir():
        if not d.name.isdigit():
            continue
        try:
            if (d / "comm").read_text().strip() != "java":
                continue
        except OSError:
            continue
        args = [a.decode(errors="replace") for a in proc_cmdline(d.name) if a]
        out.append((int(d.name), args))
    return out


def client_pids():
    """Java PIDs of the bench client only. Never the dedicated server (unix_args.txt)."""
    pids = []
    for pid, args in java_procs():
        joined = " ".join(args)
        if f"instances/{INSTANCE}" in joined and "unix_args.txt" not in joined:
            pids.append(pid)
    return pids


def server_procs():
    return [pid for pid, args in java_procs()
            if any("unix_args.txt" in a for a in args)
            and not any(f"instances/{INSTANCE}" in a for a in args)]


def cpu_ticks(pid):
    try:
        f = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()
        return int(f[11]) + int(f[12])  # utime + stime
    except (OSError, IndexError, ValueError):
        return None


def server_cpu_pct(pids, sample=1.0):
    if not pids:
        return None
    hz = os.sysconf("SC_CLK_TCK")
    a = {p: cpu_ticks(p) for p in pids}
    time.sleep(sample)
    b = {p: cpu_ticks(p) for p in pids}
    tot = sum((b[p] - a[p]) for p in pids if a[p] is not None and b[p] is not None)
    return round(100.0 * tot / hz / sample, 1)


def gpu_info():
    cards = []
    for dev in sorted(Path("/sys/class/drm").glob("card[0-9]*/device")):
        if "-" in dev.parent.name:
            continue
        info = {"card": dev.parent.name, "device": read_text(dev / "device"),
                "dpm_level": read_text(dev / "power_dpm_force_performance_level")}
        sclk = read_text(dev / "pp_dpm_sclk", "") or ""
        cur = [l for l in sclk.splitlines() if l.rstrip().endswith("*")]
        info["sclk"] = cur[0].split(":", 1)[1].strip(" *") if cur else None
        cards.append(info)
    return cards


def refresh_rates(env):
    try:
        out = subprocess.run(["xrandr", "--current"], env=env, capture_output=True,
                             text=True, timeout=5).stdout
    except (OSError, subprocess.SubprocessError):
        return []
    rates = []
    for line in out.splitlines():
        for tok in line.split():
            if "*" in tok:
                try:
                    rates.append(float(tok.strip("*+")))
                except ValueError:
                    pass
    return sorted(set(rates))


def git_state():
    try:
        sha = subprocess.run(["git", "-C", str(REPO), "rev-parse", "HEAD"],
                             capture_output=True, text=True, timeout=10).stdout.strip()
        dirty = bool(subprocess.run(["git", "-C", str(REPO), "status", "--porcelain"],
                                    capture_output=True, text=True, timeout=30).stdout.strip())
        return sha, dirty
    except (OSError, subprocess.SubprocessError):
        return None, None


def java_info():
    inst = read_cfg(INST_DIR / "instance.cfg")
    glob_cfg = read_cfg(PRISM / "prismlauncher.cfg")
    java = inst.get("JavaPath") if inst.get("OverrideJavaLocation") == "true" else glob_cfg.get("JavaPath")
    if inst.get("OverrideJavaArgs") == "true":
        jvm_args = inst.get("JvmArgs", "")
    else:
        jvm_args = glob_cfg.get("JvmArgs", "")
    mem = inst if inst.get("OverrideMemory") == "true" else glob_cfg
    version = None
    if java:
        try:
            r = subprocess.run([java, "-version"], capture_output=True, text=True, timeout=20)
            version = (r.stderr or r.stdout).splitlines()[0] if (r.stderr or r.stdout) else None
        except (OSError, subprocess.SubprocessError):
            pass
    return {"java_path": java, "java_version": version, "jvm_args_cfg": jvm_args,
            "max_mem_mb": mem.get("MaxMemAlloc") or inst.get("MaxMemAlloc"),
            "wrapper": inst.get("WrapperCommand")}


def snapshot(env, fast=False):
    sha, dirty = git_state()
    srv = server_procs()
    mh, mc = mods_hash(GAME / "mods") if (GAME / "mods").is_dir() else (None, 0)
    lvl = GAME / "saves" / WORLD / "level.dat"
    return {
        "machine": MACHINE,
        "git_sha": sha, "git_dirty": dirty,
        "harness_sha": sha256_file(Path(__file__))[:16],
        "mods_hash": mh, "mods_count": mc,
        **java_info(),
        "kernel": os.uname().release,
        "cpu_governor": read_text("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor"),
        "gpus": gpu_info(),
        "dri_prime": env.get("DRI_PRIME"),
        "mem_available_mb": meminfo_available_mb(),
        "loadavg": list(os.getloadavg()),
        "server_running": bool(srv),
        "server_cpu_pct": None if fast else server_cpu_pct(srv),
        "level_dat_sha": sha256_file(lvl)[:16] if lvl.exists() else None,
        "refresh_rates": refresh_rates(env),
    }


# ---------------------------------------------------------------- launch / monitor

class LogTail:
    """Incremental reader of latest.log; survives the file being (re)created."""

    def __init__(self, path):
        self.path, self.ino, self.pos, self.buf, self.lines = Path(path), None, 0, "", []

    def poll(self):
        try:
            st = self.path.stat()
        except FileNotFoundError:
            return []
        if st.st_ino != self.ino or st.st_size < self.pos:
            self.ino, self.pos, self.buf = st.st_ino, 0, ""
        with open(self.path, "r", errors="replace") as fh:
            fh.seek(self.pos)
            data = fh.read()
            self.pos = fh.tell()
        data = self.buf + data
        parts = data.split("\n")
        self.buf = parts.pop()
        self.lines.extend(parts)
        return parts


def crash_files():
    names = set()
    d = GAME / "crash-reports"
    if d.is_dir():
        names |= {p.name for p in d.iterdir()}
    names |= {p.name for p in GAME.glob("hs_err_pid*.log")}
    return names


def launch(env, world):
    if (GAME / "logs" / "latest.log").exists():
        (GAME / "logs" / "latest.log").unlink()
    cmd = ["setsid", "prismlauncher", "-d", str(PRISM), "--launch", INSTANCE]
    if world:
        cmd += ["--world", world]
    return subprocess.Popen(cmd, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


def kill_client(prism):
    """TERM then KILL the bench client's java PIDs, then our own Prism child."""
    for sig, wait in ((signal.SIGTERM, 10), (signal.SIGKILL, 5)):
        pids = client_pids()
        for pid in pids:
            try:
                os.kill(pid, sig)
            except ProcessLookupError:
                pass
        t0 = time.time()
        while client_pids() and time.time() - t0 < wait:
            time.sleep(0.5)
        if not client_pids():
            break
    if prism is not None and prism.poll() is None:
        prism.terminate()
        try:
            prism.wait(5)
        except subprocess.TimeoutExpired:
            prism.kill()


def monitor(tail, prism, before_crash, until, timeout, on_line=None):
    """Poll until until(state) is true, a crash, an exit, or timeout.
    Returns (outcome, detail, elapsed_s)."""
    t0 = time.time()
    seen_java = False
    while True:
        for line in tail.poll():
            c = classify_line(line)
            if c:
                return "crash", f"{c}: {line.strip()[:200]}", time.time() - t0
            if on_line:
                on_line(line)
        if until():
            return "ok", None, time.time() - t0
        new = crash_files() - before_crash
        if new:
            return "crash", "new crash file: " + ", ".join(sorted(new)), time.time() - t0
        pids = client_pids()
        seen_java = seen_java or bool(pids)
        if seen_java and not pids:
            return "crash", "client process exited", time.time() - t0
        if not seen_java and prism.poll() is not None and time.time() - t0 > 60:
            return "crash", "launcher exited without starting java", time.time() - t0
        if time.time() - t0 > timeout:
            return "timeout", f"no completion within {timeout}s", time.time() - t0
        time.sleep(1.0)


def archive_log(tag):
    src = GAME / "logs" / "latest.log"
    if src.exists():
        dst = RESULTS / "logs" / f"{tag}.log"
        dst.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(src, dst)
        return str(dst)
    return None


def append_jsonl(path, row):
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "a") as fh:
        fh.write(json.dumps(row, sort_keys=True) + "\n")


def preflight(kill_stale):
    pids = client_pids()
    if pids:
        if not kill_stale:
            sys.exit(f"bench client already running (pids {pids}); close it or pass --kill-stale")
        kill_client(None)


def apply_variant(vid):
    r = subprocess.run([sys.executable, str(PROPOSED / "apply_variant.py"), vid, str(GAME)],
                       capture_output=True, text=True)
    if r.returncode != 0:
        sys.exit(f"apply_variant {vid} failed:\n{r.stdout}{r.stderr}")
    return r.stdout.splitlines()


def apply_pins():
    opts = GAME / "options.txt"
    opts.write_text(merge_pins(opts.read_text() if opts.exists() else ""))


def apply_jvm(jvm, heap_mb):
    """Set JvmArgs / MaxMemAlloc in the bench instance.cfg (bench instance only)."""
    cfg = INST_DIR / "instance.cfg"
    orig = INST_DIR / "instance.cfg.harness-orig"
    if not orig.exists():
        shutil.copyfile(cfg, orig)
    lines = cfg.read_text().splitlines()
    want = {}
    if jvm:
        want.update({"JvmArgs": JVM_SETS[jvm], "OverrideJavaArgs": "true"})
    if heap_mb:
        want.update({"MaxMemAlloc": str(heap_mb), "OverrideMemory": "true"})
    out, seen = [], set()
    for line in lines:
        k = line.split("=", 1)[0]
        if k in want:
            line = f"{k}={want[k]}"
            seen.add(k)
        out.append(line)
    out += [f"{k}={v}" for k, v in want.items() if k not in seen]
    cfg.write_text("\n".join(out) + "\n")


def plan_dir():
    return GAME / "saves" / WORLD / "datapacks" / PLAN_PACK


def disarm():
    d = plan_dir()
    if d.exists():
        shutil.rmtree(d)
        return True
    return False


def write_datapack(files):
    disarm()
    d = plan_dir()
    for rel, text in files.items():
        p = d / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(text)


def effective_jvm_flags(pids):
    for pid in pids:
        args = [a.decode(errors="replace") for a in proc_cmdline(pid) if a]
        return [a for a in args if a.startswith(("-X", "-XX"))]
    return None


def pick_csvs(t_launch):
    """MangoHud CSVs born at/after launch, newest first, with (birth, method)."""
    out = []
    for p in MANGOLOGS.glob("*.csv"):
        if p.name.endswith("_summary.csv"):
            continue
        birth, method = file_birth(p)
        if birth >= t_launch - 2:
            out.append((birth, method, p))
    return sorted(out, reverse=True)


# ---------------------------------------------------------------- subcommands

def cmd_boot(a):
    env = bench_env()
    preflight(a.kill_stale)
    disarm()
    if a.variant:
        apply_variant(a.variant)
    apply_pins()
    meta = snapshot(env)
    variant = read_text(GAME / "BENCH_VARIANT.txt")
    counts = {}
    for i in range(1, a.n + 1):
        before = crash_files()
        t_launch = time.time()
        prism = launch(env, WORLD if a.join else None)
        tail = LogTail(GAME / "logs" / "latest.log")
        target = JOINED_RE if a.join else GAME_TOOK_RE
        hit = {"v": False}

        def on_line(line):
            if target.search(line):
                hit["v"] = True

        try:
            outcome, detail, _ = monitor(tail, prism, before, lambda: hit["v"], a.timeout,
                                         on_line)
            time.sleep(2)
            tail.poll()
        finally:
            kill_client(prism)
        ev = scan_events(tail.lines)
        tag = f"boot-{time.strftime('%Y%m%d-%H%M%S', time.localtime(t_launch))}-{i}"
        row = {
            "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(t_launch)),
            "kind": "boot", "label": a.label, "launch": i, "of": a.n, "join": a.join,
            "variant": variant, "outcome": outcome, "detail": detail,
            "startup_s": ev["game_took_s"],
            "title_s": round(ev["title_t"] - t_launch, 2) if ev["title_t"] else None,
            "join_s": round(ev["joined_t"] - t_launch, 2) if ev["joined_t"] else None,
            "atlas": ev["atlas"], "log": archive_log(tag), **meta,
            "mem_available_after_mb": meminfo_available_mb(),
        }
        append_jsonl(RESULTS / "boot.jsonl", row)
        counts[outcome] = counts.get(outcome, 0) + 1
        log(f"launch {i}/{a.n}: {outcome} startup={row['startup_s']}s title={row['title_s']}s "
            f"join={row['join_s']}s {detail or ''}")
        time.sleep(a.cooldown)
    log("totals: " + ", ".join(f"{k}={v}" for k, v in sorted(counts.items())))
    return 0 if counts.get("ok", 0) == a.n else 1


def session_plan(a):
    plan = [s.strip().lower() for s in a.plan.split(",") if s.strip()]
    bad = [s for s in plan if s not in SCENE_FN]
    if bad:
        sys.exit(f"unknown/unsupported scene(s) {bad}; s6 (GUI) is manual. Use {SCENE_ORDER}")
    if len(set(plan)) != len(plan):
        sys.exit(f"duplicate scene in plan {plan}; use --reps")
    if not a.any_order and plan != sorted(plan, key=SCENE_ORDER.index):
        sys.exit(f"plan {plan} breaks the S1->S6 session order (BENCH-MATRIX §3); "
                 f"pass --any-order to override")
    return plan


def cmd_session(a):
    plan = session_plan(a)
    sid = "p" + time.strftime("%Y%m%d%H%M%S")
    files = generate_datapack(sid, plan, a.reps, a.warmup, a.capture, a.settle)
    budget = plan_seconds(plan, a.reps, a.warmup, a.capture)
    timeout = a.timeout or int(a.boot_timeout + a.settle + budget * 1.25 + 120)
    if not (PROPOSED / "out" / a.variant).is_dir():
        sys.exit(f"variant {a.variant} not in {PROPOSED / 'out'} - run build.py first")
    if a.dry_run:
        log(f"dry run: session {sid}, variant {a.variant}, plan {plan} x{a.reps}, "
            f"est. scene time {budget}s, timeout {timeout}s")
        log(f"would pin options.txt: {', '.join(f'{k}:{v}' for k, v in PINS)}")
        if a.jvm or a.heap:
            log(f"would set instance.cfg: jvm={a.jvm} heap={a.heap}")
        log(f"would write datapack {plan_dir()}:")
        for rel, text in files.items():
            print(f"--- {PLAN_PACK}/{rel}")
            print(text, end="")
        return 0
    if budget + a.settle + 120 > 1700:
        log("WARNING: plan exceeds ~28 min; mangohud-local log_duration=1800 will truncate the CSV")
    env = bench_env()
    preflight(a.kill_stale)
    apply_out = apply_variant(a.variant)
    apply_pins()
    if a.jvm or a.heap:
        apply_jvm(a.jvm, a.heap)
    cfg_hash, cfg_n = tree_hash(GAME / "config", config_skip)
    opt_hash = sha256_file(GAME / "options.txt")
    meta = snapshot(env)
    write_datapack(files)
    before = crash_files()
    t_launch = time.time()
    log(f"session {sid}: {a.variant} {plan} x{a.reps}, timeout {timeout}s")
    prism = launch(env, WORLD)
    tail = LogTail(GAME / "logs" / "latest.log")
    state = {"done": False, "last": None}

    def on_line(line):
        m = MARK_RE.search(line)
        if m and "[CHAT]" not in line:
            log(f"  {m.group(1):5s} {m.group(2)} rep {m.group(3)}")
        if DONE_RE.search(line):
            state["done"] = True

    jvm_flags = None
    try:
        def until():
            nonlocal jvm_flags
            if jvm_flags is None:
                jvm_flags = effective_jvm_flags(client_pids())
            return state["done"]
        outcome, detail, _ = monitor(tail, prism, before, until, timeout, on_line)
        time.sleep(2)  # let MangoHud flush the tail of the capture
        tail.poll()
    finally:
        kill_client(prism)
        disarm()
    mem_after = meminfo_available_mb()
    lines = tail.lines
    ev = scan_events(lines)
    marks, _ = parse_markers(lines)
    backups = backup_times(lines)
    log_path = archive_log(f"session-{sid}")
    csvs = pick_csvs(t_launch)
    parsed = []
    for birth, method, p in csvs:
        try:
            cmeta, rows = parse_mangohud_csv(p)
        except (OSError, ValueError) as e:
            log(f"skip {p.name}: {e}")
            continue
        if rows:
            parsed.append((birth, method, p, cmeta, rows))
    if not parsed:
        log("WARNING: no MangoHud CSV created after launch")
    base = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(t_launch)),
        "kind": "window", "session": sid, "label": a.label, "variant": a.variant,
        "jvm_set": a.jvm, "heap_mb": a.heap, "jvm_flags": jvm_flags,
        "config_hash": cfg_hash[:16], "config_files": cfg_n, "options_hash": opt_hash[:16],
        "apply_output": apply_out,
        "session_outcome": outcome, "session_detail": detail,
        "startup_s": ev["game_took_s"],
        "join_s": round(ev["joined_t"] - t_launch, 2) if ev["joined_t"] else None,
        "atlas": ev["atlas"], "log": log_path, **meta, "mem_available_after_mb": mem_after,
    }
    n_ok = 0
    for scene in plan:
        warmup = 0 if scene == "s5" else a.warmup
        capture = S5_CAPTURE if scene == "s5" else a.capture
        for rep in range(1, a.reps + 1):
            ts, te = marks.get(("start", scene, rep)), marks.get(("end", scene, rep))
            row = dict(base, scene=scene, rep=rep, warmup_s=warmup, capture_s=capture,
                       t_start=ts, t_end=te)
            if ts is None or te is None:
                row.update(metrics=None, flags=[{"crash": "crashed", "timeout": "timeout"}
                                                .get(outcome, "no_markers")])
                append_jsonl(RESULTS / "session.jsonl", row)
                log(f"{scene} rep {rep}: no complete window ({outcome})")
                continue
            m, src = None, None
            for birth, method, p, cmeta, rows in parsed:  # newest first
                if birth <= ts and birth + rows[-1][0] >= te - 0.5:
                    m, src = window_metrics(rows, birth, ts, te), (birth, method, p, cmeta)
                    break
            if m is None and parsed:
                birth, method, p, cmeta, rows = parsed[0]
                m, src = window_metrics(rows, birth, ts, te), (birth, method, p, cmeta)
            flags = validity_flags(m, te - ts, meta["refresh_rates"], backups, ts, te)
            if outcome != "ok":
                flags.append("session_" + outcome)  # warn: window itself completed
            row.update(metrics=m, flags=flags)
            if src:
                row.update(csv=src[2].name, align=src[1], csv_gpu=src[3].get("gpu", "").strip(),
                           csv_birth=round(src[0], 3))
            append_jsonl(RESULTS / "session.jsonl", row)
            if m:
                n_ok += 1
                log(f"{scene} rep {rep}: n={m['n']} p50={m['p50_ms']} p99={m['p99_ms']} "
                    f"1%low={m['one_low_fps']} avg={m['avg_fps']} {','.join(flags)}")
    log(f"session {sid} {outcome}: {n_ok} windows recorded -> {RESULTS / 'session.jsonl'}")
    return 0 if outcome == "ok" else 1


def cmd_disarm(a):
    log("removed " + str(plan_dir()) if disarm() else "nothing armed")
    return 0


def cmd_meta(a):
    print(json.dumps(snapshot(bench_env()), indent=2))
    return 0


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0],
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)

    b = sub.add_parser("boot", help="N cold launches; startup time + crash classification")
    b.add_argument("--n", type=int, default=3)
    b.add_argument("--variant", help="apply this variant once before the launches")
    b.add_argument("--join", action="store_true", help="launch into the bench world, wait for join")
    b.add_argument("--label", default=None)
    b.add_argument("--timeout", type=int, default=300)
    b.add_argument("--cooldown", type=int, default=5, help="seconds between launches")
    b.add_argument("--kill-stale", action="store_true")
    b.set_defaults(fn=cmd_boot)

    s = sub.add_parser("session", help="one launch, datapack-driven scene plan")
    s.add_argument("--variant", required=True)
    s.add_argument("--plan", default="s1", help="comma list in S1->S5 order: s1,s2,s2s,s3,s4,s5")
    s.add_argument("--reps", type=int, default=3)
    s.add_argument("--warmup", type=int, default=15)
    s.add_argument("--capture", type=int, default=30)
    s.add_argument("--settle", type=int, default=10, help="seconds after join before the plan")
    s.add_argument("--label", default=None, help="matrix block, e.g. B0")
    s.add_argument("--jvm", choices=sorted(JVM_SETS),
                   help="write this BENCH-MATRIX JVM set into instance.cfg (default: untouched)")
    s.add_argument("--heap", type=int, help="MaxMemAlloc MB to write into instance.cfg")
    s.add_argument("--boot-timeout", type=int, default=300)
    s.add_argument("--timeout", type=int, default=None, help="overall session timeout (s)")
    s.add_argument("--any-order", action="store_true")
    s.add_argument("--kill-stale", action="store_true")
    s.add_argument("--dry-run", action="store_true", help="print the generated datapack only")
    s.set_defaults(fn=cmd_session)

    d = sub.add_parser("disarm", help="remove a leftover autoplan datapack from the world")
    d.set_defaults(fn=cmd_disarm)
    m = sub.add_parser("meta", help="print the environment/metadata snapshot")
    m.set_defaults(fn=cmd_meta)

    a = ap.parse_args(argv)
    return a.fn(a)


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