#!/usr/bin/env python3
"""B14 spawn-tuning measurement (SPAWN-TUNING.md §B14), automated.

    b14.py locate  [--zip bench-world-v1.zip] [--out sites.json]   # offline, once
    b14.py run     --config {defaults,shipped} [--biome river,jungle,warm_ocean]
                   [--minutes 10] [--dry-run]
    b14.py summary [--file ~/onion-bench/results/b14.jsonl]
    b14.py parse   LOG                                              # debug: count lines
    b14.py restore                                                  # = bench.py restore

One `run` = one client launch into a throwaway world `saves/onionspawn` that holds only
the bench world's level.dat (seed `onionbench`, gamerules, datapack list), so every
chunk around the sites is generated fresh under the config being measured. A generated
datapack sets doMobSpawning true + Creative, parks the player on a barrier block at each
pre-located site (sites.json), and every 60 s prints a parsable chat line
`B14 <sid> <biome> t=<min> biome_ok=1 total=N <entity id>=N ...` counted within 128
blocks. The spawn config variant is journaled through bench.Journal, so `bench.py
restore` undoes a hard-killed run. Stdlib only.
"""
import argparse
import json
import os
import re
import signal
import statistics
import subprocess
import sys
import time
import zipfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
sys.path.insert(0, str(HERE.parent))
import bench  # noqa: E402  (harness module: launch, Journal, LogTail, monitor, ...)
import nbt  # noqa: E402

WORLD = "onionspawn"
PACK = "b14-autoplan"
SITES_FILE = HERE / "sites.json"
SOURCE_ZIP = bench.BENCH / "worlds" / "bench-world-v1.zip"
RESULTS_FILE = bench.RESULTS / "b14.jsonl"

# The three shipped partial files (d80f85b, 8e8483d). `defaults` deletes them so each mod
# regenerates its upstream defaults; `shipped` writes the committed partials.
SPAWN_FILES = ["alexsmobs-common.toml", "fowlplay.json5", "crittersandcompanions-common.json5"]

# Tuned species (verified against the jars' lang keys + registries, see README).
TUNED = [
    ("am_gazelle", "alexsmobs:gazelle"),
    ("am_capuchin", "alexsmobs:capuchin_monkey"),
    ("am_elephant", "alexsmobs:elephant"),
    ("am_straddler", "alexsmobs:straddler"),
    ("am_rocky_roller", "alexsmobs:rocky_roller"),
    ("am_underminer", "alexsmobs:underminer"),
    ("am_blobfish", "alexsmobs:blobfish"),
    ("am_mudskipper", "alexsmobs:mudskipper"),
    ("am_caiman", "alexsmobs:caiman"),
    ("fp_duck", "fowlplay:duck"),
    ("fp_goose", "fowlplay:goose"),
    ("fp_gull", "fowlplay:gull"),
    ("fp_sparrow", "fowlplay:sparrow"),
    ("cac_leaf_insect", "crittersandcompanions:leaf_insect"),
    ("cac_sea_bunny", "crittersandcompanions:sea_bunny"),
    ("cac_dragonfly", "crittersandcompanions:dragonfly"),
    ("cac_ladybug", "crittersandcompanions:ladybug"),
    ("cac_stag_beetle", "crittersandcompanions:stag_beetle"),
    ("cac_roly_poly", "crittersandcompanions:roly_poly"),
    ("cac_snail", "crittersandcompanions:snail"),
    ("cac_stick_bug", "crittersandcompanions:stick_bug"),
    ("cac_weevil", "crittersandcompanions:weevil"),
]
# Vanilla spot-check (SPAWN-TUNING: "cows, sheep still spawn normally") + biome natives.
VANILLA = [
    ("mc_cow", "minecraft:cow"), ("mc_sheep", "minecraft:sheep"), ("mc_pig", "minecraft:pig"),
    ("mc_chicken", "minecraft:chicken"), ("mc_parrot", "minecraft:parrot"),
    ("mc_ocelot", "minecraft:ocelot"), ("mc_tropical_fish", "minecraft:tropical_fish"),
    ("mc_cod", "minecraft:cod"), ("mc_squid", "minecraft:squid"),
    ("mc_dolphin", "minecraft:dolphin"),
]
# Species SPAWN-TUNING §B14 names as the point of each biome (reported first).
FOCUS = {
    "river": ["fowlplay:duck", "fowlplay:gull"],
    "jungle": ["alexsmobs:capuchin_monkey", "crittersandcompanions:leaf_insect"],
    "warm_ocean": ["crittersandcompanions:sea_bunny"],
}
BIOMES = {
    "river": {"match": {"minecraft:river"}, "check": "minecraft:river"},
    "jungle": {"match": {"minecraft:jungle", "minecraft:sparse_jungle",
                         "minecraft:bamboo_jungle"}, "check": "#minecraft:is_jungle"},
    "warm_ocean": {"match": {"minecraft:warm_ocean"}, "check": "minecraft:warm_ocean"},
}

# Upstream defaults of the keys we check after load (SPAWN-TUNING "default -> new").
EXPECT_DEFAULTS = {
    "alexsmobs-common.toml": {
        "gazelleSpawnWeight": 40, "capuchinMonkeySpawnWeight": 28, "elephantSpawnWeight": 30,
        "straddlerSpawnWeight": 70, "rockyRollerSpawnWeight": 60, "underminerSpawnWeight": 50,
        "blobfishSpawnWeight": 30, "mudskipperSpawnWeight": 28, "caimanSpawnWeight": 29},
    "fowlplay.json5": {
        "duckSpawnWeight": 4, "duckMaxGroupSize": 6, "gooseSpawnWeight": 4,
        "gooseMaxGroupSize": 4, "gullSpawnWeight": 5, "gullMaxGroupSize": 8,
        "sparrowSpawnWeight": 20, "sparrowMaxGroupSize": 6},
    "crittersandcompanions-common.json5": {
        "sea_bunny@minecraft:warm_ocean": 32, "leaf_insect@#c:is_jungle": 14,
        "leaf_insect@#c:is_forest": 6, "dragonfly@#c:is_lush": 10},
}

COUNT_RE = re.compile(r"\bB14 (?P<sid>b14-\d+) (?P<biome>[a-z_]+) t=(?P<t>\d+) (?P<rest>.*)$")
DONE_RE = re.compile(r"B14_DONE (b14-\d+)")
FUNC_FAIL_RE = re.compile(r"Failed to load function b14:|Unknown function b14:")


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


# ---------------------------------------------------------------- config variants

def shipped_texts(rev="HEAD", repo=bench.REPO):
    out = {}
    for name in SPAWN_FILES:
        r = subprocess.run(["git", "-C", str(repo), "show", f"{rev}:config/{name}"],
                           capture_output=True, text=True)
        if r.returncode != 0:
            raise RuntimeError(f"git show {rev}:config/{name} failed: {r.stderr.strip()}")
        out[name] = r.stdout
    return out


def render_variant(config, shipped):
    """{file name: text to write, or None = delete (mod regenerates defaults)}."""
    if config == "defaults":
        return {name: None for name in SPAWN_FILES}
    if config == "shipped":
        return {name: shipped[name] for name in SPAWN_FILES}
    raise ValueError(f"unknown config {config!r}")


def _cac_weight(text, mob, spec):
    m = re.search(rf'"?{re.escape(mob)}"?\s*:\s*\[(.*?)\]', text, re.S)
    if not m:
        return None
    for entry in m.group(1).split("}"):
        if f'"{spec}"' in entry:
            w = re.search(r'"?weight"?\s*:\s*(-?\d+)', entry)
            return int(w.group(1)) if w else None
    return None


def extract_values(name, text, keys):
    """Values of `keys` from a partial or mod-re-saved spawn file (None = absent)."""
    out = {}
    for k in keys:
        if name.endswith(".toml"):
            m = re.search(rf"^\s*{re.escape(k)}\s*=\s*(-?\d+)", text, re.M)
            out[k] = int(m.group(1)) if m else None
        elif "@" in k:
            mob, spec = k.split("@", 1)
            out[k] = _cac_weight(text, mob, spec)
        else:
            m = re.search(rf'"?{re.escape(k)}"?\s*:\s*(-?\d+)', text)
            out[k] = int(m.group(1)) if m else None
    return out


def expected_values(config, shipped):
    if config == "defaults":
        return {n: dict(v) for n, v in EXPECT_DEFAULTS.items()}
    return {n: extract_values(n, shipped[n], EXPECT_DEFAULTS[n]) for n in SPAWN_FILES}


def check_config(game, config, shipped):
    """Compare what the mods re-saved after load with what the variant should produce."""
    want = expected_values(config, shipped)
    got, mismatch = {}, []
    for name in SPAWN_FILES:
        p = Path(game) / "config" / name
        if not p.is_file() and config == "defaults":
            got[name] = dict(want[name])  # never written: the mod ran on its defaults
            continue
        text = p.read_text(errors="replace") if p.is_file() else ""
        got[name] = extract_values(name, text, want[name])
        for k, v in want[name].items():
            if got[name][k] != v:
                mismatch.append(f"{name}:{k} want {v} got {got[name][k]}")
    return {"ok": not mismatch, "mismatch": mismatch, "values": got}


def apply_spawn_config(game, journal, files):
    for name, text in files.items():
        rel = f"config/{name}"
        journal.backup_file(rel)
        p = Path(game) / rel
        if text is None:
            if p.exists():
                p.unlink()
        else:
            p.write_text(text)


def config_diff(game, files, shipped):
    """Key-level plan of the config change (for --dry-run): checked keys now -> after."""
    out = []
    for name, text in files.items():
        p = Path(game) / "config" / name
        cur = p.read_text(errors="replace") if p.is_file() else None
        keys = EXPECT_DEFAULTS[name]
        now = extract_values(name, cur, keys) if cur else {}
        if text is None:
            out.append(f"--- config/{name}: DELETE ({len(cur)} bytes now)" if cur
                       else f"--- config/{name}: DELETE (absent now)")
            out.append("    mod regenerates upstream defaults on load")
            after = keys
        else:
            out.append(f"--- config/{name}: WRITE shipped partial ({len(text)} bytes)")
            after = extract_values(name, shipped[name], keys)
        for k in keys:
            mark = "" if now.get(k) == after[k] else "   <- changes"
            out.append(f"    {k}: {now.get(k)} -> {after[k]}{mark}")
        if text is not None:
            out.extend("    | " + ln for ln in text.splitlines())
    return "\n".join(out)


# ---------------------------------------------------------------- datapack

def _score(name):
    return {"score": {"name": f"#{name}", "objective": "b14"}}


def count_function():
    lines = ["# B14 entity counts around the executing position (the player)",
             "scoreboard players set #tuned b14 0"]
    for alias, eid in TUNED + VANILLA:
        lines.append(f"execute store result score #{alias} b14 "
                     f"if entity @e[type={eid},distance=..128]")
    for alias, _ in TUNED:
        lines.append(f"scoreboard players operation #tuned b14 += #{alias} b14")
    lines.append("execute store result score #total b14 "
                 "if entity @e[type=!minecraft:player,distance=..128]")
    lines.append("execute store result score #nonitem b14 if entity "
                 "@e[type=!minecraft:player,type=!minecraft:item,"
                 "type=!minecraft:experience_orb,distance=..128]")
    return "\n".join(lines) + "\n"


def tellraw(sid, biome, t):
    parts = ["", {"text": f"B14 {sid} {biome} t={t} biome_ok="}, _score("biome_ok")]
    for key, alias in [("total", "total"), ("nonitem", "nonitem"), ("tuned", "tuned")]:
        parts += [{"text": f" {key}="}, _score(alias)]
    for alias, eid in TUNED + VANILLA:
        parts += [{"text": f" {eid}="}, _score(alias)]
    return "tellraw @a " + json.dumps(parts, separators=(",", ":"))


def build_timeline(sid, sites, minutes, settle_s, arrive_s=10, gen_s=5):
    """[(commands, delay_ticks_to_next)] for the unrolled plan."""
    ns = f"b14:{sid}"
    tl = []
    for biome, s in sites:
        x, y, z = s["x"], s["y"], s["z"]
        pos = f"{x + 0.5} {y} {z + 0.5}"
        # proto-chunk sites come from the pre-feature heightmap: clear a 3x3x3 pocket in
        # case a tree landed on the column, then stand on a barrier (also over water).
        hold = [f"fill {x - 1} {y} {z - 1} {x + 1} {y + 2} {z + 1} minecraft:air",
                f"setblock {x} {y - 1} {z} minecraft:barrier",
                f"execute in minecraft:overworld run tp @a {pos} 0 0"]
        tl.append(([f"execute in minecraft:overworld run tp @a {x + 0.5} {y + 2} {z + 0.5} 0 0"],
                   gen_s * 20))
        tl.append((hold + [f"say B14_ARRIVE {sid} {biome} {x} {y} {z}"], arrive_s * 20))
        for t in range(minutes + 1):
            cmds = hold + [
                f"execute as @a[limit=1] at @s run function {ns}/count",
                f"execute as @a[limit=1] at @s store success score #biome_ok b14 "
                f"if biome ~ ~ ~ {BIOMES[biome]['check']}",
                tellraw(sid, biome, t)]
            tl.append((cmds, 1200 if t < minutes else 20))
    tl.append(([f"say B14_DONE {sid}"], 0))
    return tl


def plan_seconds(n_sites, minutes, settle_s, arrive_s=10, gen_s=5):
    return settle_s + n_sites * (gen_s + arrive_s + minutes * 60 + 1)


def generate_datapack(sid, sites, minutes, settle_s):
    """{relative path: text}; namespace b14, functions under b14:<sid>/."""
    ns, holder = f"b14:{sid}", "#armed"
    base = f"data/b14/function/{sid}"
    files = {
        "pack.mcmeta": json.dumps({"pack": {
            "pack_format": 48,
            "description": f"Onion B14 spawn plan {sid} (generated by b14.py; "
                           f"throwaway world)"}}) + "\n",
        "data/minecraft/tags/function/load.json": json.dumps({"values": [f"{ns}/load"]}) + "\n",
        f"{base}/load.mcfunction":
            "scoreboard objectives add b14 dummy\n"
            f"execute unless score {holder} b14 matches 1.. run schedule function {ns}/poll 20t\n",
        f"{base}/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"{base}/begin.mcfunction":
            f"scoreboard players set {holder} b14 1\n"
            # Creative: counts as a spawning/despawning player (spectators don't), and
            # can't die. Every rule that changes spawning is pinned, not inherited.
            "gamemode creative @a\n"
            "gamerule doMobSpawning true\n"
            "gamerule doDaylightCycle false\n"
            "gamerule doWeatherCycle false\n"
            "gamerule doPatrolSpawning false\n"
            "gamerule doTraderSpawning false\n"
            "gamerule doInsomnia false\n"
            "gamerule doWardenSpawning false\n"
            "time set 6000\n"
            "weather clear\n"
            f"say B14_PLAN {sid} begin\n"
            f"schedule function {ns}/step0 {max(1, settle_s * 20)}t\n",
        f"{base}/count.mcfunction": count_function(),
    }
    tl = build_timeline(sid, sites, minutes, settle_s)
    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"{base}/step{i}.mcfunction"] = "\n".join(body) + "\n"
    return files


# ---------------------------------------------------------------- world

def world_members(zip_path):
    """Members copied into the throwaway world: level.dat, the bench datapack (it is in
    level.dat's enabled list; no load/tick tags) and serverconfig/. No chunks/entities/
    data/ (forceloads, DH db) so every chunk is generated under the measured config."""
    prefix = bench.zip_world_root(zip_path)
    with zipfile.ZipFile(zip_path) as z:
        names = z.namelist()
    keep = []
    for n in names:
        rel = n[len(prefix):]
        if n.endswith("/") or not n.startswith(prefix):
            continue
        if rel == "level.dat" or rel.startswith(("datapacks/onion-bench/", "serverconfig/")):
            keep.append((n, rel))
    if not any(rel == "level.dat" for _, rel in keep):
        raise ValueError(f"{zip_path}: no level.dat")
    return keep


def build_world(game, journal, zip_path, files):
    root = Path(game) / "saves" / WORLD
    if root.exists():
        raise ValueError(f"{root} exists (stale B14 world?) - run `b14.py restore` or remove it")
    members = world_members(zip_path)
    journal.record({"op": "tree", "path": f"saves/{WORLD}"})
    with zipfile.ZipFile(zip_path) as z:
        for n, rel in members:
            dst = root / rel
            dst.parent.mkdir(parents=True, exist_ok=True)
            dst.write_bytes(z.read(n))
    write_pack(root, files)
    return root


def write_pack(world_root, files):
    d = Path(world_root) / "datapacks" / PACK
    for rel, text in files.items():
        p = d / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(text)


# ---------------------------------------------------------------- locate (offline)

# Chunk statuses whose biome palette is final (before "biomes" it is a placeholder).
BIOME_STATUSES = {"biomes", "noise", "surface", "carvers", "features", "initialize_light",
                  "light", "spawn", "full"}


def _chunk_info(chunk):
    """Surface samples of one chunk, or None. Full chunks use MOTION_BLOCKING_NO_LEAVES
    (ground under trees, water surface); proto chunks the worldgen WORLD_SURFACE_WG."""
    status = str(chunk.get("Status", "")).split(":")[-1]
    if status not in BIOME_STATUSES:
        return None
    hm = nbt.heightmap(chunk) or nbt.heightmap(chunk, "WORLD_SURFACE_WG")
    if hm is None:
        return None
    ys = {}
    for y in hm:
        ys[y] = ys.get(y, 0) + 1
    mode_y = max(ys, key=lambda k: (ys[k], -k))
    samples = []
    for lx, lz in ((2, 2), (2, 10), (10, 2), (10, 10), (8, 8)):
        y = hm[lz * 16 + lx]
        samples.append(nbt.biome_at(chunk, lx, y - 1, lz))
    col = None  # a column at the modal height (ground, not a trunk or hole), near centre
    best = None
    for lz in range(16):
        for lx in range(16):
            if hm[lz * 16 + lx] == mode_y:
                d = (lx - 8) ** 2 + (lz - 8) ** 2
                if best is None or d < best:
                    best, col = d, (lx, lz)
    centre = nbt.biome_at(chunk, col[0], mode_y - 1, col[1])
    return {"samples": samples, "y": mode_y, "col": col, "centre": centre,
            "full": status == "full"}


def scan_world(zip_path, stride=1):
    """{(cx, cz): info} for every fully generated overworld chunk in the zip."""
    prefix = bench.zip_world_root(zip_path)
    out = {}
    with zipfile.ZipFile(zip_path) as z:
        regions = [i for i in z.infolist()
                   if i.filename.startswith(prefix + "region/") and i.filename.endswith(".mca")
                   and i.file_size > 8192]
        for info in regions:
            rx, rz = (int(v) for v in info.filename.rsplit("/", 1)[1].split(".")[1:3])
            for lx, lz, chunk in nbt.region_chunks(z.read(info.filename)):
                if stride > 1 and (lx % stride or lz % stride):
                    continue
                ci = _chunk_info(chunk)
                if ci:
                    out[(rx * 32 + lx, rz * 32 + lz)] = ci
    return out


def pick_sites(chunks, avoid, min_gap=320, avoid_gap=192, radius=2):
    """Best chunk per target biome: centre column in the biome, maximal share of the
    biome in the (2r+1)^2 neighbourhood; far from `avoid` points and from each other."""
    chosen = {}
    for biome, spec in BIOMES.items():
        match = spec["match"]
        cands = []
        for (cx, cz), ci in chunks.items():
            if ci["centre"] not in match:
                continue
            hit = tot = forest = 0
            for dx in range(-radius, radius + 1):
                for dz in range(-radius, radius + 1):
                    n = chunks.get((cx + dx, cz + dz))
                    if not n:
                        continue
                    for b in n["samples"]:
                        tot += 1
                        hit += b in match
                        forest += bool(b) and "forest" in b
            if tot < 5 * 9:
                continue  # fewer than 9 known chunks around it
            score = hit / tot + (0.5 * forest / tot if biome == "river" else 0.0) \
                + (0.05 if ci["full"] else 0.0)
            x, z = cx * 16 + ci["col"][0], cz * 16 + ci["col"][1]
            cands.append((score, x, ci["y"], z, hit / tot, ci["full"], ci["centre"]))
        for score, x, y, z, frac, full, centre in sorted(cands, reverse=True):
            pts = list(avoid) + [(s["x"], s["z"]) for s in chosen.values()]
            gaps = [avoid_gap] * len(avoid) + [min_gap] * len(chosen)
            if all((x - px) ** 2 + (z - pz) ** 2 >= g * g for (px, pz), g in zip(pts, gaps)):
                chosen[biome] = {"x": x, "y": y, "z": z, "biome": centre,
                                 "frac": round(frac, 3), "score": round(score, 3),
                                 "full_chunk": full, "candidates": len(cands)}
                break
    return chosen


def avoid_points(zip_path):
    prefix = bench.zip_world_root(zip_path)
    with zipfile.ZipFile(zip_path) as z:
        _, ld = nbt.load_gzip(z.read(prefix + "level.dat"))
    d = ld.get("Data", {})
    pts = [(d.get("SpawnX", 0), d.get("SpawnZ", 0))]
    pos = (d.get("Player") or {}).get("Pos")
    if pos:
        pts.append((int(pos[0]), int(pos[2])))
    seed = (d.get("WorldGenSettings") or {}).get("seed")
    return pts, seed


def cmd_locate(a):
    t0 = time.time()
    chunks = scan_world(a.zip, a.stride)
    avoid, seed = avoid_points(a.zip)
    sites = pick_sites(chunks, avoid)
    hist = {}
    for ci in chunks.values():
        hist[ci["centre"]] = hist.get(ci["centre"], 0) + 1
    log(f"scanned {len(chunks)} chunks with final biomes in {time.time() - t0:.0f}s; avoid {avoid}")
    for b, n in sorted(hist.items(), key=lambda kv: -kv[1])[:25]:
        print(f"  {n:6d} {b}")
    missing = sorted(set(BIOMES) - set(sites))
    doc = {"source_zip": str(a.zip), "source_sha256": bench.sha256_file(a.zip),
           "seed": seed, "avoid": avoid, "stride": a.stride,
           "note": "y = first free block (full chunks: MOTION_BLOCKING_NO_LEAVES, proto "
                   "chunks: WORLD_SURFACE_WG); the player stands on a barrier at y-1. "
                   "Valid only for this seed + mod set (biomes are seed-deterministic).",
           "sites": sites}
    print(json.dumps(doc, indent=1))
    if a.out:
        Path(a.out).write_text(json.dumps(doc, indent=1) + "\n")
        log(f"wrote {a.out}")
    if missing:
        log(f"NO SITE for {missing}: not in the generated chunks; see README (manual locate)")
        return 1
    return 0


# ---------------------------------------------------------------- parse / summary

def parse_count_line(line):
    m = COUNT_RE.search(line)
    if not m:
        return None
    vals = {}
    for tok in m.group("rest").split():
        k, sep, v = tok.partition("=")
        if sep and re.fullmatch(r"-?\d+", v):
            vals[k] = int(v)
    return {"sid": m.group("sid"), "biome": m.group("biome"), "t": int(m.group("t")),
            "biome_ok": vals.pop("biome_ok", None), "total": vals.pop("total", None),
            "nonitem": vals.pop("nonitem", None), "tuned": vals.pop("tuned", None),
            "counts": vals}


def parse_lines(lines, sid=None):
    """Count rows from log lines; the client logs each tellraw once as [CHAT]."""
    rows, seen = [], set()
    for ln in lines:
        r = parse_count_line(ln)
        if not r or (sid and r["sid"] != sid):
            continue
        key = (r["sid"], r["biome"], r["t"])
        if key in seen:
            continue
        seen.add(key)
        rows.append(r)
    return rows


def load_rows(path):
    p = Path(path)
    if not p.exists():
        return [], []
    rows = [json.loads(ln) for ln in p.read_text().splitlines() if ln.strip()]
    return ([r for r in rows if r.get("kind") == "b14_count"],
            [r for r in rows if r.get("kind") == "b14_session"])


def summarize(counts, sessions):
    """{biome: {config: stats}} plus verdict per SPAWN-TUNING §B14."""
    bad = {s["sid"] for s in sessions
           if s.get("outcome") != "ok" or not (s.get("config_check") or {}).get("ok")}
    runs = {}
    for r in counts:
        if r["sid"] in bad:
            continue
        runs.setdefault((r["biome"], r["config"], r["sid"]), []).append(r)
    out = {}
    for (biome, config, sid), rs in runs.items():
        rs.sort(key=lambda r: r["t"])
        final = rs[-1]
        st = out.setdefault(biome, {}).setdefault(config, {
            "runs": 0, "final_total": [], "final_nonitem": [], "final_tuned": [],
            "seen": {}, "final": {}, "biome_ok": True, "minutes": []})
        st["runs"] += 1
        st["minutes"].append(final["t"])
        st["final_total"].append(final["total"])
        st["final_nonitem"].append(final["nonitem"])
        st["final_tuned"].append(final["tuned"])
        st["biome_ok"] &= all(r.get("biome_ok") == 1 for r in rs)
        for eid in set().union(*(r["counts"] for r in rs)):
            mx = max(r["counts"].get(eid, 0) for r in rs)
            st["seen"][eid] = max(st["seen"].get(eid, 0), mx)
            st["final"].setdefault(eid, []).append(final["counts"].get(eid, 0))
    tuned_ids = {eid for _, eid in TUNED}
    verdicts = {}
    for biome, by in out.items():
        d, s = by.get("defaults"), by.get("shipped")
        if not (d and s):
            verdicts[biome] = {"verdict": "incomplete",
                               "have": sorted(by)}
            continue
        dt, stt = statistics.median(d["final_total"]), statistics.median(s["final_total"])
        gone = sorted(e for e in tuned_ids if d["seen"].get(e, 0) > 0 and s["seen"].get(e, 0) == 0)
        rare = sorted(e for e in tuned_ids
                      if d["seen"].get(e, 0) >= 4 and 0 < s["seen"].get(e, 0) < 0.25 * d["seen"][e])
        vanilla_gone = sorted(e for _, e in VANILLA if e in ("minecraft:cow", "minecraft:sheep")
                              and d["seen"].get(e, 0) > 0 and s["seen"].get(e, 0) == 0)
        ok_biome = d["biome_ok"] and s["biome_ok"]
        v = "keep" if stt < dt and not gone else ("back off" if gone else "no drop")
        verdicts[biome] = {"verdict": v if ok_biome else "invalid (biome check failed)",
                           "total_defaults": dt, "total_shipped": stt, "disappeared": gone,
                           "rare_warn": rare, "vanilla_gone": vanilla_gone}
    vs = [v["verdict"] for v in verdicts.values()]
    overall = ("incomplete" if not vs or len(verdicts) < len(BIOMES)
               or any(v not in ("keep", "back off", "no drop") for v in vs)
               else "back off" if "back off" in vs else "keep" if all(v == "keep" for v in vs)
               else "review")
    return out, verdicts, overall


def render_summary(out, verdicts, overall):
    lines = ["| biome | config | runs | final total | nonitem | tuned | focus (max seen) |",
             "|---|---|---|---|---|---|---|"]
    for biome in sorted(out):
        for config in ("defaults", "shipped"):
            st = out[biome].get(config)
            if not st:
                continue
            focus = ", ".join(f"{e.split(':')[1]}={st['seen'].get(e, 0)}"
                              for e in FOCUS.get(biome, []))
            lines.append(f"| {biome} | {config} | {st['runs']} | "
                         f"{statistics.median(st['final_total'])} | "
                         f"{statistics.median(st['final_nonitem'])} | "
                         f"{statistics.median(st['final_tuned'])} | {focus} |")
    lines.append("")
    for biome, v in sorted(verdicts.items()):
        lines.append(f"- **{biome}**: {v['verdict']}"
                     + (f" (total {v['total_defaults']} -> {v['total_shipped']}"
                        f"{'; disappeared ' + ', '.join(v['disappeared']) if v['disappeared'] else ''}"
                        f"{'; rare ' + ', '.join(v['rare_warn']) if v['rare_warn'] else ''}"
                        f"{'; vanilla gone ' + ', '.join(v['vanilla_gone']) if v['vanilla_gone'] else ''})"
                        if "total_defaults" in v else f" (have {v.get('have')})"))
    lines.append(f"\n**Overall: {overall}** (keep = totals drop in every biome and no tuned "
                 f"species that appears under defaults vanishes; SPAWN-TUNING §B14)")
    return "\n".join(lines)


def cmd_summary(a):
    counts, sessions = load_rows(a.file)
    if not counts:
        log(f"no B14 rows in {a.file}")
        return 1
    out, verdicts, overall = summarize(counts, sessions)
    print(render_summary(out, verdicts, overall))
    return 0


def cmd_parse(a):
    rows = parse_lines(Path(a.log).read_text(errors="replace").splitlines())
    for r in rows:
        print(json.dumps(r, sort_keys=True))
    return 0 if rows else 1


# ---------------------------------------------------------------- run

def _ancestors(pid):
    out = set()
    while pid > 1 and pid not in out:
        out.add(pid)
        try:
            stat = Path(f"/proc/{pid}/stat").read_text()
            pid = int(stat.rsplit(")", 1)[1].split()[1])
        except (OSError, ValueError, IndexError):
            break
    return out


def bench_busy():
    """Other harness processes (matrix runner / bench.py session / another B14 run) that
    share the instance; our own process and its parent shells are excluded."""
    mine = _ancestors(os.getpid())
    busy = []
    for d in Path("/proc").iterdir():
        if not d.name.isdigit() or int(d.name) in mine:
            continue
        try:
            args = (d / "cmdline").read_bytes().split(b"\0")
        except OSError:
            continue
        s = " ".join(x.decode(errors="replace") for x in args)
        if re.search(r"\bmatrix\.py run\b|\bbench\.py session\b|\bb14\.py run\b", s) \
                and "--dry-run" not in s:
            busy.append(int(d.name))
    return busy


def load_sites(path, biomes):
    doc = json.loads(Path(path).read_text())
    sites = doc.get("sites", {})
    missing = [b for b in biomes if b not in sites]
    if missing:
        raise ValueError(f"{path}: no site for {missing} (run `b14.py locate` or add by hand)")
    return [(b, sites[b]) for b in biomes], doc


def cmd_run(a):
    biomes = list(BIOMES) if a.biome == "all" else bench.split_list(a.biome)
    for b in biomes:
        if b not in BIOMES:
            sys.exit(f"unknown biome {b!r}; choose from {sorted(BIOMES)}")
    try:
        sites, sites_doc = load_sites(a.sites, biomes)
        shipped = shipped_texts(a.rev)
        files = render_variant(a.config, shipped)
        world_members(a.source_zip)
    except (ValueError, OSError, RuntimeError, zipfile.BadZipFile) as e:
        sys.exit(str(e))
    sid = "b14-" + time.strftime("%Y%m%d%H%M%S")
    pack = generate_datapack(sid, sites, a.minutes, a.settle)
    budget = plan_seconds(len(sites), a.minutes, a.settle)
    timeout = a.timeout or int(a.boot_timeout + budget + 180)
    game = bench.GAME
    if a.dry_run:
        log(f"dry run {sid}: config={a.config} rev={a.rev} biomes={biomes} "
            f"minutes={a.minutes} plan {budget}s timeout {timeout}s dh={a.dh}")
        for b, s in sites:
            log(f"  site {b}: {s}")
        log(f"world saves/{WORLD} <- {a.source_zip}: "
            f"{[rel for _, rel in world_members(a.source_zip)][:6]}...")
        print(config_diff(game, files, shipped))
        if bench.pending_exists():
            log(f"WARNING: {game / bench.PENDING} exists; a real run would refuse")
        busy = bench_busy()
        if busy:
            log(f"WARNING: harness processes running {busy}; a real run would refuse")
        for rel, text in pack.items():
            print(f"--- {PACK}/{rel}")
            print(text, end="")
        return 0
    bench.refuse_pending()
    busy = bench_busy()
    if busy:
        sys.exit(f"harness processes running (pids {busy}); B14 shares the bench instance")
    if bench.server_procs() and not a.allow_server:
        log("WARNING: a dedicated server is running (noise only; spawning is client-side)")
    env = bench.bench_env()
    bench.preflight(a.kill_stale)
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(143))
    journal = bench.Journal.begin(game, sid)
    prism, state, tail = None, {"done": False, "fail": None}, None
    outcome, detail, check, meta, failed = "error", None, None, None, []
    t_launch = time.time()
    try:
        bench.reset_resource_packs(game)
        bench.apply_pins()
        bench.apply_dh(game, journal, a.dh == "on")
        apply_spawn_config(game, journal, files)
        build_world(game, journal, a.source_zip, pack)
        meta = {"git": bench.git_state(), "mods": bench.mods_hash(game / "mods"),
                "variant": bench.read_text(game / "BENCH_VARIANT.txt")}
        before = bench.crash_files()
        log(f"{sid}: config={a.config} biomes={biomes} {a.minutes} min, timeout {timeout}s")
        t_launch = time.time()
        prism = bench.launch(env, WORLD)
        tail = bench.LogTail(game / "logs" / "latest.log")

        def on_line(line):
            r = parse_count_line(line)
            if r and r["sid"] == sid and "[CHAT]" in line:
                log(f"  {r['biome']} t={r['t']} total={r['total']} tuned={r['tuned']} "
                    f"biome_ok={r['biome_ok']}")
            if DONE_RE.search(line):
                state["done"] = True
            if FUNC_FAIL_RE.search(line):
                state["fail"] = line.strip()[:300]

        outcome, detail, _ = bench.monitor(
            tail, prism, before, lambda: state["done"] or state["fail"], timeout, on_line)
        if state["fail"]:
            outcome, detail = "datapack_error", state["fail"]
    finally:
        if prism is not None:
            bench.kill_client(prism)
        if tail is not None:
            tail.poll()
        try:
            check = check_config(game, a.config, shipped)  # mods' re-saved files, pre-restore
        except OSError as e:
            check = {"ok": False, "mismatch": [f"read failed: {e}"], "values": {}}
        failed = bench.restore_pending(game)
        for e, err in failed:
            log(f"RESTORE FAILED {e}: {err}")
        if failed:
            log(f"journal kept at {game / bench.PENDING}; fix and run `b14.py restore`")
    log_path = bench.archive_log(f"session-{sid}")
    rows = parse_lines(tail.lines if tail else [], sid)
    base = {"sid": sid, "config": a.config, "rev": a.rev, "minutes": a.minutes,
            "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(t_launch)),
            "machine": bench.MACHINE}
    for r in rows:
        bench.append_jsonl(RESULTS_FILE, {**base, **r, "kind": "b14_count",
                                          "site": dict(sites)[r["biome"]]})
    bench.append_jsonl(RESULTS_FILE, {
        **base, "kind": "b14_session", "outcome": outcome, "detail": detail,
        "biomes": biomes, "sites": dict(sites), "sites_sha256": sites_doc.get("source_sha256"),
        "config_check": check, "rows": len(rows), "log": log_path,
        "git_sha": meta["git"][0] if meta else None,
        "git_dirty": meta["git"][1] if meta else None,
        "mods_hash": meta["mods"][0] if meta else None,
        "mods_count": meta["mods"][1] if meta else None,
        "bench_variant": meta["variant"] if meta else None,
        "restore_failed": [e for e, _ in failed]})
    log(f"{sid} {outcome}{': ' + detail if detail else ''}; {len(rows)} count rows; "
        f"config check {'ok' if check and check['ok'] else check and check['mismatch']}")
    want = len(sites) * (a.minutes + 1)
    return 0 if outcome == "ok" and len(rows) == want and check and check["ok"] else 1


def cmd_restore(a):
    if bench.client_pids():
        sys.exit("bench client is running; close it before restoring")
    if not bench.pending_exists():
        log("nothing pending")
        return 0
    failed = bench.restore_pending(bench.GAME)
    for e, err in failed:
        log(f"FAILED {e}: {err}")
    return 1 if failed else 0


def build_parser():
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    sub = ap.add_subparsers(dest="cmd", required=True)
    r = sub.add_parser("run", help="one client launch: config variant x biome sites")
    r.add_argument("--config", required=True, choices=["defaults", "shipped"])
    r.add_argument("--biome", default="all",
                   help="comma list of river,jungle,warm_ocean (in this order), or all")
    r.add_argument("--minutes", type=int, default=10)
    r.add_argument("--settle", type=int, default=20, help="seconds after join before site 1")
    r.add_argument("--sites", default=str(SITES_FILE))
    r.add_argument("--source-zip", default=str(SOURCE_ZIP),
                   help="world whose level.dat (seed, gamerules) seeds the throwaway world")
    r.add_argument("--rev", default="HEAD", help="git rev of the shipped partial files")
    r.add_argument("--dh", choices=["on", "off"], default="off")
    r.add_argument("--boot-timeout", type=int, default=300)
    r.add_argument("--timeout", type=int, default=None)
    r.add_argument("--kill-stale", action="store_true")
    r.add_argument("--allow-server", action="store_true", help="silence the server warning")
    r.add_argument("--dry-run", action="store_true")
    r.set_defaults(fn=cmd_run)
    lo = sub.add_parser("locate", help="find biome sites offline from a world zip")
    lo.add_argument("--zip", default=str(SOURCE_ZIP))
    lo.add_argument("--out", default=None, help=f"write sites json (e.g. {SITES_FILE})")
    lo.add_argument("--stride", type=int, default=1, help="sample every Nth chunk")
    lo.set_defaults(fn=cmd_locate)
    s = sub.add_parser("summary", help="defaults vs shipped, SPAWN-TUNING §B14 rule")
    s.add_argument("--file", default=str(RESULTS_FILE))
    s.set_defaults(fn=cmd_summary)
    p = sub.add_parser("parse", help="print count rows found in a log file")
    p.add_argument("log")
    p.set_defaults(fn=cmd_parse)
    rs = sub.add_parser("restore", help="undo a hard-killed run (same journal as bench.py)")
    rs.set_defaults(fn=cmd_restore)
    return ap


def main(argv=None):
    a = build_parser().parse_args(argv)
    return a.fn(a)


if __name__ == "__main__":
    sys.exit(main())