chore(bench): add client bench harness and Phase 2 handoff

The client now boots cleanly on the bench machine (the CAC spawn-key
crash was fixed in 8e8483d). Add the harness used to drive bench scenes
and capture frametimes, the bench-world generation workflow, and a
self-contained handoff brief. All files live under analysis/ (packwizignored),
so nothing player-facing changes.

- harness/rcon.py: minimal RCON client for the temp worldgen server
- harness/run_scene.py: apply variant -> launch Prism under MangoHud ->
  xdotool scene commands -> p50/p95/p99/1%-low from the CSV
- HANDOFF.md: state, gotchas, next steps
- build.py: disable DH autoUpdater for bench variants (suppress modal)

analysis/client-perf/harness/rcon.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Minimal Minecraft RCON client (stdlib only).
+
+Usage: python3 rcon.py <host> <port> <password> <command> [command ...]
+Exit non-zero if the server reports a failure.
+"""
+import socket
+import struct
+import sys
+
+
+class Rcon:
+    def __init__(self, host: str, port: int, password: str, timeout: float = 10.0):
+        self.host, self.port, self.password = host, port, password
+        self.timeout = timeout
+        self._id = 0
+
+    def _send(self, sock: socket.socket, ptype: int, body: str) -> int:
+        self._id += 1
+        rid = self._id
+        payload = struct.pack("<ii", rid, ptype) + body.encode("utf-8") + b"\x00\x00"
+        sock.sendall(struct.pack("<i", len(payload)) + payload)
+        return rid
+
+    def _recv(self, sock: socket.socket):
+        raw = b""
+        while len(raw) < 4:
+            chunk = sock.recv(4 - len(raw))
+            if not chunk:
+                raise ConnectionError("RCON connection closed")
+            raw += chunk
+        (length,) = struct.unpack("<i", raw)
+        body = b""
+        while len(body) < length:
+            chunk = sock.recv(length - len(body))
+            if not chunk:
+                raise ConnectionError("RCON connection closed")
+            body += chunk
+        rid, ptype = struct.unpack("<ii", body[:8])
+        return rid, ptype, body[8:-2].decode("utf-8", "replace")
+
+    def command(self, cmd: str) -> str:
+        with socket.create_connection((self.host, self.port), timeout=self.timeout) as sock:
+            sock.settimeout(self.timeout)
+            rid = self._send(sock, 3, self.password)
+            rid2, _, _ = self._recv(sock)
+            if rid2 == -1:
+                raise PermissionError("RCON auth failed")
+            self._send(sock, 2, cmd)
+            _, _, text = self._recv(sock)
+            return text
+
+
+def main(argv):
+    if len(argv) < 5:
+        print(__doc__)
+        return 2
+    host, port, password = argv[1], int(argv[2]), argv[3]
+    rc = 0
+    for cmd in argv[4:]:
+        try:
+            out = Rcon(host, port, password).command(cmd)
+        except Exception as exc:  # noqa: BLE001 - CLI surface
+            print(f"! {cmd}: {exc}", file=sys.stderr)
+            rc = 1
+            continue
+        if out.strip():
+            print(out.strip())
+        lowered = out.lower()
+        if "unknown" in lowered or "failed" in lowered or "error" in lowered:
+            rc = 1
+    return rc
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv))

analysis/client-perf/harness/run_scene.py

@@ -0,0 +1,267 @@
+#!/usr/bin/env python3
+"""Client-perf bench harness: drive one variant/scene and extract frametime metrics.
+
+Applies a generated variant, launches the Prism instance under the user-local
+MangoHud wrapper straight into the bench world, runs the scene via xdotool chat
+commands, records wall-clock markers, then slices the MangoHud CSV.
+
+Usage:
+  python3 run_scene.py --variant C1 --scene s1 [--warmup 15] [--capture 30]
+                       [--runs 1] [--record out.csv] [--pre storm_on]
+
+Scenes: s1, s2, s3, s4, s5 (tour = 120s capture, no warmup), s2s (= s2 + storm).
+Only stdlib. Must run on the desktop session that owns DISPLAY :1.
+"""
+import argparse
+import bisect
+import csv as csvmod
+import os
+import re
+import shutil
+import signal
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+PROPOSED = HERE.parent / "proposed-configs"
+ROOT = Path("/home/logan/onion-bench/prism")
+INSTANCE = "OnionBench"
+GAME = ROOT / "instances" / INSTANCE / ".minecraft"
+LOGS = GAME / "logs" / "latest.log"
+MANGOLOGS = Path("/home/logan/onion-bench/mangologs")
+WORLD = "onionbench"
+ENV = {
+    **os.environ,
+    "DISPLAY": ":1",
+    "XAUTHORITY": "/run/user/1000/xauth_HvvPAq",
+    "XDG_RUNTIME_DIR": "/run/user/1000",
+    "PATH": "/home/logan/.local/bin:" + os.environ.get("PATH", ""),
+}
+
+SCENE_FN = {
+    "s1": "bench:s1",
+    "s2": "bench:s2",
+    "s3": "bench:s3",
+    "s4": "bench:s4",
+    "s5": "bench:tour_start",
+    "s2s": "bench:s2",
+}
+
+
+def log(msg):
+    print(f"[run_scene] {msg}", flush=True)
+
+
+def xdo(*args):
+    subprocess.run(["xdotool", *args], env=ENV, check=False,
+                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+
+def mc_window():
+    out = subprocess.run(["xdotool", "search", "--name", "Minecraft"],
+                         env=ENV, capture_output=True, text=True).stdout.split()
+    return out[-1] if out else None
+
+
+def client_java_pids():
+    out = subprocess.run(["ps", "-C", "java", "-o", "pid=,args="],
+                         capture_output=True, text=True).stdout
+    pids = []
+    for line in out.splitlines():
+        if f"instances/{INSTANCE}/.minecraft" in line or f"instances/{INSTANCE}/natives" in line:
+            pids.append(int(line.split()[0]))
+    return pids
+
+
+def apply_variant(vid):
+    subprocess.run([sys.executable, str(PROPOSED / "apply_variant.py"), vid, str(GAME)],
+                   check=True, stdout=subprocess.DEVNULL)
+
+
+def launch_client():
+    if LOGS.exists():
+        LOGS.unlink()
+    p = subprocess.Popen(
+        ["setsid", "prismlauncher", "-d", str(ROOT), "--launch", INSTANCE, "--world", WORLD],
+        env=ENV, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    )
+    return p
+
+
+def wait_for_world(timeout=240):
+    t0 = time.time()
+    while time.time() - t0 < timeout:
+        if LOGS.exists():
+            txt = LOGS.read_text(errors="replace")
+            if "Sound engine started" in txt:
+                return True
+        time.sleep(2)
+    return False
+
+
+def chat(command, settle=0.6):
+    wid = mc_window()
+    if wid:
+        xdo("windowactivate", wid)
+        xdo("windowfocus", wid)
+    xdo("key", "--clearmodifiers", "t")
+    time.sleep(0.4)
+    xdo("type", "--clearmodifiers", "--delay", "12", command)
+    time.sleep(0.2)
+    xdo("key", "--clearmodifiers", "Return")
+    time.sleep(settle)
+
+
+def newest_csv():
+    files = [p for p in MANGOLOGS.glob("*.csv") if not p.name.endswith("_summary.csv")]
+    return max(files, key=lambda p: p.stat().st_mtime) if files else None
+
+
+def parse_csv(path):
+    """Return (columns, rows) where rows are lists of floats; line 3 is the header."""
+    lines = path.read_text(errors="replace").splitlines()
+    hdr_i = None
+    for i, line in enumerate(lines[:10]):
+        if "frametime" in line and "elapsed" in line:
+            hdr_i = i
+            break
+    if hdr_i is None:
+        raise RuntimeError(f"no frametime header in {path}")
+    cols = lines[hdr_i].split(",")
+    e_i = cols.index("elapsed")
+    ft_i = cols.index("frametime")
+    fps_i = 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]), float(parts[ft_i]), float(parts[fps_i]), parts))
+        except ValueError:
+            continue
+    return cols, rows
+
+
+def pct(values, p):
+    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 metrics_between(rows, t_start, t_end, run_start):
+    """rows: list of (elapsed_ns, frametime_ms, fps, raw). Wall = run_start + elapsed/1e9."""
+    ft, fps = [], []
+    for e, frametime, f, raw in rows:
+        wall = run_start + e / 1e9
+        if t_start <= wall <= t_end:
+            ft.append(frametime)
+            fps.append(f)
+    if not ft:
+        return None
+    fps_sorted = sorted(fps)
+    n1 = max(1, int(len(fps_sorted) * 0.01))
+    one_low = sum(fps_sorted[:n1]) / n1
+    return {
+        "n": len(ft),
+        "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(sum(fps) / len(fps), 2),
+        "one_low_fps": round(one_low, 2),
+    }
+
+
+def kill_client(prism):
+    try:
+        os.killpg(os.getpgid(prism.pid), signal.SIGTERM)
+    except Exception:
+        pass
+    for pid in client_java_pids():
+        try:
+            os.kill(pid, signal.SIGTERM)
+        except ProcessLookupError:
+            pass
+    time.sleep(5)
+    for pid in client_java_pids():
+        try:
+            os.kill(pid, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+
+
+def run_once(variant, scene, warmup, capture, pre):
+    apply_variant(variant)
+    p = launch_client()
+    run_start = time.time()
+    try:
+        if not wait_for_world():
+            log("ERROR: world never loaded")
+            return None
+        time.sleep(8)  # let the world settle / place player
+        fn = SCENE_FN[scene]
+        if scene != "s5":
+            chat(f"/function {fn}", settle=1.0)
+        if pre:
+            chat(f"/function bench:{pre}", settle=1.0)
+        if scene == "s5":
+            capture = 120
+            warmup = 0
+            chat("/function bench:tour_start", settle=0.5)
+        time.sleep(warmup)
+        t_start = time.time()
+        chat(f"/say BENCH_CAPTURE_START {variant} {scene}")
+        time.sleep(capture)
+        t_end = time.time()
+        chat("/say BENCH_CAPTURE_END")
+        chat("/save-all", settle=2.0)
+    finally:
+        csv_path = newest_csv()
+        kill_client(p)
+    cols, rows = parse_csv(csv_path)
+    m = metrics_between(rows, t_start, t_end, run_start)
+    if m:
+        m.update({"variant": variant, "scene": scene, "csv": csv_path.name,
+                  "run_start": round(run_start, 3)})
+    return m
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--variant", required=True)
+    ap.add_argument("--scene", required=True, choices=sorted(SCENE_FN))
+    ap.add_argument("--warmup", type=int, default=15)
+    ap.add_argument("--capture", type=int, default=30)
+    ap.add_argument("--runs", type=int, default=1)
+    ap.add_argument("--pre", default=None)
+    ap.add_argument("--record", default=None)
+    a = ap.parse_args()
+    results = []
+    for i in range(a.runs):
+        log(f"run {i + 1}/{a.runs}: {a.variant} {a.scene}")
+        m = run_once(a.variant, a.scene, a.warmup, a.capture, a.pre)
+        if m:
+            results.append(m)
+            print("RESULT " + str(m), flush=True)
+        time.sleep(3)
+    if a.record and results:
+        fields = ["variant", "scene", "n", "p50_ms", "p95_ms", "p99_ms", "peak_ms",
+                  "avg_fps", "one_low_fps", "csv", "run_start"]
+        exists = Path(a.record).exists()
+        with open(a.record, "a", newline="") as fh:
+            w = csvmod.DictWriter(fh, fieldnames=fields)
+            if not exists:
+                w.writeheader()
+            for m in results:
+                w.writerow({k: m.get(k, "") for k in fields})
+    return 0 if results else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())