"""Run reproducible small-brain experiments without changing a published release.

Commands:
  generate --out DIR --teacher-url URL --teacher-model ID --chats N --diaries N
  semantic-generate --manifest FILE --out DIR --teacher-url URL --teacher-model ID
                    [--per-cell 32 --workers 4] [--offset N --pilot-audit DIR --prior-generation DIR]
  correction-data --manifest FILE --student-export EXPORT --teacher-url URL
                  --teacher-model ID --pilot-audit DIR --prior-generation DIR --out DIR
  corpus-audit --data FILE --requests JSONL [JSONL ...] --reviews JSONL [JSONL ...]
               --manifest FILE --engine-binary BINARY --out DIR [--pilot]
  freeze-corpus --audits DIR [DIR ...] --generations DIR --manifest FILE --out DIR
  paired-inference --reference REPORT --candidate REPORT --published-graph-sha256 SHA
                   --out DIR
  rollout-data --teacher-export EXPORT --teacher-id ID --out DIR --chats N --diaries N
  merge --data FILE.jsonl [FILE.jsonl ...] --out DIR
  run --data DATA.jsonl --out DIR [--models ID,ID] [--max-steps N]
  compare --run DIR [--baseline REPORT.json] [--review REVIEW.json]
  benchmark --run DIR --out DIR --url BENCH_URL --export-python PYTHON
            --native-probe BINARY --ort-library LIBRARY [--models ID,ID]

Benchmark outputs benchmark.json/benchmark.md and fresh per-variant behavior,
browser and native receipts. Rankings are descriptive screening measurements,
not acceptance, inherited-v6 parity or publication permission.

Outputs must be outside the repository. Training artifacts, data and logs are
never committed. --python and --export-python select isolated tool environments;
--device and --threads control compute. No teacher endpoint, deployment path or
GPU host is hard-coded. A result is eligible only after complete comparable
behavior reports, acceptable quantization loss, actual runtime measurements and
transcript review. Smoke/screening runs are labelled, never publication evidence.
"""
import argparse
import hashlib
import json
import math
import os
from pathlib import Path
import platform
import statistics
import subprocess
import sys
import time

ROOT = Path(__file__).resolve().parent.parent
DEFAULT_PLAN = Path(__file__).with_name("experiments.json")


def digest(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def load_json(path):
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def save_json(path, value):
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(value, f, indent=2, sort_keys=True, allow_nan=False)
        f.write("\n")


def fresh_directory(path):
    path = Path(path).expanduser().resolve()
    if path == ROOT or ROOT in path.parents:
        raise ValueError("experiment outputs must be outside the repository")
    if path.exists() and any(path.iterdir()):
        raise ValueError(f"output directory is not empty: {path}")
    path.mkdir(parents=True, exist_ok=True)
    return path


def command(argv, log, env=None):
    start = time.monotonic()
    print("RUN " + " ".join(map(str, argv)), flush=True)
    with open(log, "w", encoding="utf-8") as f:
        result = subprocess.run(list(map(str, argv)), cwd=ROOT,
                                stdout=f, stderr=subprocess.STDOUT, env=env)
    elapsed = time.monotonic() - start
    if result.returncode:
        raise RuntimeError(f"command failed ({result.returncode}); see {log}")
    return {"argv": list(map(str, argv)), "seconds": elapsed, "log": str(log)}


def verify_dataset(path):
    # Import without torch so dataset preflight works in the stdlib environment.
    import gen_data as G
    counts = {}
    rows = 0
    seen = set()
    with open(path, encoding="utf-8") as f:
        for line in f:
            row = json.loads(line)
            G.recheck(row)
            text_hash = hashlib.sha256(row["text"].encode()).hexdigest()
            if text_hash in seen:
                raise ValueError("duplicate canonical text in dataset")
            seen.add(text_hash)
            rows += 1
            key = row["kind"] + ":" + row.get("stage", "unspecified")
            counts[key] = counts.get(key, 0) + 1
    if rows < 2:
        raise ValueError("at least two validated rows are required")
    return {"path": str(Path(path).resolve()), "sha256": digest(path),
            "rows": rows, "counts": counts}


def generate(args):
    out = fresh_directory(args.out)
    argv = [args.python, ROOT / "train/gen_data.py", out / "data.jsonl",
            args.chats, args.diaries, "--endpoints", args.teacher_url,
            "--model", args.teacher_model, "--workers", args.workers,
            "--seed", args.seed]
    if args.focus:
        argv += ["--focus", args.focus]
    if args.stage:
        argv += ["--stage", args.stage]
    if args.curriculum:
        argv += ["--curriculum", "--receipt", out / "requests.jsonl"]
    receipt = {"kind": "synthetic-data", "seed": args.seed,
               "teacher": {"url": args.teacher_url, "model": args.teacher_model},
               "requested": {"chats": args.chats, "diaries": args.diaries,
                             "focus": args.focus, "curriculum": args.curriculum},
               "generator_sha256": digest(ROOT / "train/gen_data.py"),
               "status": "running"}
    if args.teacher_artifact:
        receipt["teacher"]["artifact_sha256"] = digest(args.teacher_artifact)
    save_json(out / "generation.json", receipt)
    try:
        receipt["command"] = command(argv, out / "generation.log")
        receipt["dataset"] = verify_dataset(out / "data.jsonl")
        receipt["status"] = "complete"
    except Exception as e:
        receipt.update(status="failed", error=str(e))
        raise
    finally:
        save_json(out / "generation.json", receipt)
    print(json.dumps(receipt, indent=2))


def rollout_data(args):
    """Sequence distillation on independently sampled scenes, never eval cases."""
    import copy
    import collections
    import re
    import torch
    from transformers import AutoConfig, AutoTokenizer, GenerationConfig
    import gen_data as G
    import eval_pet as V
    from onnx_utils import CausalSession, model_path
    if min(args.chats, args.diaries) < 0 or args.chats + args.diaries < 1 or args.threads < 1:
        raise ValueError("positive rollout count and threads are required")
    out = fresh_directory(args.out)
    export = Path(args.teacher_export).expanduser().resolve()
    config = AutoConfig.from_pretrained(export, local_files_only=True)
    tokenizer = AutoTokenizer.from_pretrained(export, local_files_only=True)
    generation = GenerationConfig.from_pretrained(export, local_files_only=True)
    graph = model_path(export, args.dtype)
    torch.set_num_threads(args.threads)
    session = CausalSession(graph, config.to_dict(), threads=args.threads)
    evaluator = V.Evaluator(None, tokenizer, torch.device("cpu"), args.seed, session, config, generation)
    records, current = [], {}

    def chat_writer(brain, st, scene):
        initial, live, lines = copy.deepcopy(st), copy.deepcopy(st), []
        for turn, slot in enumerate(scene["slots"]):
            lines.extend(slot["events"])
            if f"* {st['pet']} woke up *" in slot["events"]:
                live["asleep"] = False
            if f"* {st['owner']} cleaned up the poop *" in slot["events"]:
                live["poop"] = 0
            result = brain.pet_turn(live, lines, "react" if slot["events"] else "idle",
                                    f"rollout/{current['index']}/{turn}")
            records.append(result)
            if result["failure"] or not result["delivered"] or not result["contract_valid"]:
                raise ValueError("teacher rollout failed the generation/lookup contract")
            text = result["text"]
            if "*" in text or G.fix_tags(text) != text:
                raise ValueError("teacher rollout requires text repair")
            tags = list(G.W.TAG_RE.finditer(text))
            saved = [tag.group(2) for tag in tags if tag.group(1) == "remember"]
            if slot.get("remember"):
                key = scene["target"]["key"].lower()
                pattern = r"(?<![a-z0-9])" + re.escape(key) + r"(?![a-z0-9])"
                if not any(re.search(pattern, fact.lower()) for fact in saved):
                    raise ValueError("teacher rollout missed the disclosed fact")
            elif saved:
                raise ValueError("teacher rollout invented an unsolicited memory")
            if slot.get("lookup") and not any(tag.group(1) == slot["lookup"] for tag in tags):
                raise ValueError("teacher rollout missed its requested lookup")
            if slot["mode"] == "NAP" and "[sleep]" not in text:
                raise ValueError("teacher rollout missed the nap")
            if slot["mode"] == "AWAKE" and "[sleep]" in text:
                raise ValueError("teacher rollout slept in an awake sequence")
            lines.append(st["pet"] + ": " + text)
            if slot.get("recall"):
                G.check_recall(slot["recall"], lines[-1], st["pet"])
            for fact in saved:
                if fact.lower() not in [value.lower() for value in live["memories"]]:
                    live["memories"].append(fact)
            live["shelf"].extend(tag.group(2) for tag in tags if tag.group(1) == "gift")
            if slot["mode"] == "OFFER" and V.offer_answer(text) == "yes":
                if live["asleep"]:
                    lines.append(f"* {st['pet']} woke up *")
                    live["asleep"] = False
                if "catch_stars" in slot:
                    lines.append(f"* {st['owner']} played catch with {st['pet']} ({slot['catch_stars']} stars) *")
            if "[sleep]" in text:
                live["asleep"] = True
        text, spans = G.check_chat(initial, "\n".join(lines))
        if text != G.W.header("chat", initial) + "\n".join(lines) + "\n":
            raise ValueError("teacher rollout text changed during validation")
        row = {"kind": "chat", "stage": st["stage"], "text": text, "spans": spans,
               "mask_version": 2, "curriculum": {"focus": scene["focus"], "variant": scene["variant"]}}
        G.recheck(row)
        return row

    def diary_writer(brain, st):
        result = brain.diary(st, f"rollout/{current['index']}/diary")
        records.append(result)
        if result["failure"] or result["unsafe_raw"] or not result["format_valid"]:
            raise ValueError("teacher rollout diary failed raw format/safety")
        text, spans = G.check_diary(st, result["text"])
        row = {"kind": "diary", "stage": st["stage"], "text": text, "spans": spans, "mask_version": 2}
        G.recheck(row)
        return row

    receipt = {"status": "running", "kind": "pet-brain-sequence-distillation",
               "teacher": {"id": args.teacher_id, "export": str(export), "dtype": args.dtype,
                           "graph_sha256": digest(graph), "graph_bytes": graph.stat().st_size,
                           "files_sha256": {p.name: digest(p) for p in export.glob("*.json")},
                           "session": session.metadata(), "providers": session.session.get_providers(),
                           "disable_quant_qdq": True, "disable_prepacking": True},
               "hardware": {"platform": platform.platform(), "torch": torch.__version__,
                            "torch_threads": torch.get_num_threads()},
               "seed": args.seed, "threads": args.threads, "requested": {"chats": args.chats, "diaries": args.diaries, "focus": args.focus},
               "source_sha256": {str(p.relative_to(ROOT)): digest(p) for p in
                                  (ROOT / "train/experiments.py", ROOT / "train/gen_data.py", ROOT / "train/eval_pet.py")},
               "scope": "Independently sampled scripted snapshots, not evaluator cases or timed body simulation; every pet line is actual teacher inference. No publication or same-corpus parity claim."}
    save_json(out / "generation.json", receipt)
    accepted, rejected, reasons = 0, 0, collections.Counter()
    started = time.monotonic()
    jobs = G.curriculum_jobs(args.chats, args.diaries, args.seed, focus=args.focus, with_receipts=True,
                             chat_writer=chat_writer, diary_writer=diary_writer)
    with open(out / "data.jsonl", "x") as data, open(out / "requests.jsonl", "x") as transcript:
        for fn, metadata in jobs:
            current["index"] = metadata["index"]
            records.clear()
            try:
                row = fn(evaluator)
                text_hash = hashlib.sha256(row["text"].encode()).hexdigest()
                data.write(json.dumps(row, ensure_ascii=False) + "\n")
                data.flush()
                accepted += 1
                error = None
            except Exception as exc:
                rejected += 1
                text_hash = None
                error = {"type": type(exc).__name__, "reason": str(exc)}
                reasons[str(exc)] += 1
            transcript.write(json.dumps({**metadata, "accepted": error is None, "error": error,
                                         "text_sha256": text_hash, "turns": records}, ensure_ascii=False, allow_nan=False) + "\n")
            transcript.flush()
            if (accepted + rejected) % 50 == 0:
                print(f"ROLLOUT accepted={accepted} rejected={rejected} {dict(reasons)}", flush=True)
    receipt.update(status="complete" if accepted else "failed", accepted=accepted, rejected=rejected,
                   rejections=dict(reasons), seconds=time.monotonic() - started)
    if accepted:
        receipt["dataset"] = verify_dataset(out / "data.jsonl")
    if digest(graph) != receipt["teacher"]["graph_sha256"]:
        raise ValueError("teacher graph changed during rollouts")
    save_json(out / "generation.json", receipt)
    print(json.dumps(receipt, indent=2))
    return 0 if accepted else 1


def merge(args):
    import gen_data as G
    out = fresh_directory(args.out)
    seen, rows, sources = set(), [], []
    for path in args.data:
        source = {"path": str(Path(path).resolve()), "sha256": digest(path), "rows": 0, "duplicates": 0}
        with open(path, encoding="utf-8") as f:
            for line in f:
                row = json.loads(line)
                G.recheck(row)
                source["rows"] += 1
                if row["text"] in seen:
                    source["duplicates"] += 1
                    continue
                seen.add(row["text"])
                rows.append(row)
        if digest(path) != source["sha256"]:
            raise ValueError(f"source changed during merge: {path}")
        sources.append(source)
    path = out / "data.jsonl"
    with open(path, "w", encoding="utf-8") as f:
        for row in rows:
            f.write(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n")
    receipt = {"status": "complete", "sources": sources, "dataset": verify_dataset(path)}
    save_json(out / "merge.json", receipt)
    print(json.dumps(receipt, indent=2))


def jsonl_rows(path):
    """Keep physical line numbers, including interrupted producer receipts."""
    with open(path, encoding="utf-8") as stream:
        for number, line in enumerate(stream, 1):
            if not line.strip():
                continue
            try:
                yield number, json.loads(line)
            except json.JSONDecodeError as exc:
                raise ValueError(f"invalid JSON receipt {path}:{number}: {exc}") from exc


def value_hash(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False,
                                     separators=(",", ":"), allow_nan=False).encode()).hexdigest()


def text_hash(row):
    return hashlib.sha256(row["text"].encode()).hexdigest()


def source_identity(path):
    path = Path(path).expanduser().resolve()
    return {"path": str(path), "sha256": digest(path)}


def write_jsonl(stream, value):
    stream.write(json.dumps(value, ensure_ascii=False, sort_keys=True, allow_nan=False) + "\n")
    stream.flush()


def request_history(directories):
    """Follow the immutable receipt chain, counting starts even after a crash."""
    seen, calls, groups = {}, set(), set()
    def visit(directory):
        directory = Path(directory).expanduser().resolve()
        if str(directory) in seen:
            return
        receipt_path = directory / "generation.json"
        receipt = load_json(receipt_path)
        seen[str(directory)] = source_identity(receipt_path)
        for previous in receipt.get("prior_generations", []):
            path = Path(previous["path"])
            if digest(path) != previous["sha256"]:
                raise ValueError("prior generation receipt changed")
            visit(path.parent)
        completed_groups = {}
        requests = directory / "requests.jsonl"
        if requests.is_file():
            for _, request in jsonl_rows(requests):
                if request.get("group_id"):
                    groups.add(request["group_id"])
                    if request.get("case_id"):
                        completed_groups[request["case_id"]] = request["group_id"]
        for artifact in receipt.get("artifacts", {}).values():
            if digest(artifact["path"]) != artifact["sha256"]:
                raise ValueError("prior generation artifact changed")
        ledger = directory / "api-calls.jsonl"
        if ledger.is_file():
            for _, event in jsonl_rows(ledger):
                if event["event"] == "start":
                    calls.add(event["request_id"])
                    # Complete older receipts can recover reservations through
                    # their matching case. An interrupted unreserved legacy
                    # start is not safe to resume or silently request again.
                    group = event.get("group_id") or completed_groups.get(event.get("case_id"))
                    if not group:
                        raise ValueError("started teacher request lacks a recoverable training group reservation")
                    groups.add(group)
    for directory in directories:
        visit(directory)
    return list(seen.values()), calls, groups


def bounded_scene_results(jobs, worker, workers):
    """At most workers pending scenes; HTTP calls never run under ledger locks."""
    import concurrent.futures as futures
    if type(workers) is not int or workers < 1:
        raise ValueError("workers must be positive")
    jobs = iter(jobs)
    if workers == 1:
        for job in jobs:
            yield worker(*job)
        return
    with futures.ThreadPoolExecutor(max_workers=workers) as pool:
        pending = set()
        for _ in range(workers):
            job = next(jobs, None)
            if job is None:
                break
            pending.add(pool.submit(worker, *job))
        while pending:
            completed, pending = futures.wait(pending, return_when=futures.FIRST_COMPLETED)
            for future in completed:
                yield future.result()
                job = next(jobs, None)
                if job is not None:
                    pending.add(pool.submit(worker, *job))


def semantic_generate(args):
    """Bounded teacher generation; acceptance is decided by the separate audit."""
    import copy
    import random
    import threading
    import gen_data as G
    import semantic_data as S
    if args.workers < 1:
        raise ValueError("workers must be positive")
    if getattr(args, "student_export", None) and args.workers != 1:
        raise ValueError("student correction inference requires --workers 1; model sessions are not shared")
    manifest = S.load_manifest(args.manifest)
    if args.per_cell < 0 or min(args.curriculum_chats, args.diaries, args.offset) < 0:
        raise ValueError("request counts/offset must be nonnegative")
    prior, previous_calls, previous_groups = request_history(args.prior_generation)
    pilot_evidence = None
    if args.prior_generation or args.offset or args.curriculum_chats or args.diaries or getattr(args, "student_export", None):
        if not args.pilot_audit or not args.prior_generation:
            raise ValueError("expansion/corrections require --pilot-audit and --prior-generation accounting")
        pilot_path = Path(args.pilot_audit)
        if pilot_path.is_dir():
            pilot_path /= "audit.json"
        pilot = load_json(pilot_path)
        if (pilot.get("status") != "complete" or not pilot.get("pilot_gate", {}).get("passed") or
                pilot.get("manifest", {}).get("sha256") != digest(args.manifest)):
            raise ValueError("checked pilot gate/manifest mismatch")
        for artifact in pilot["artifacts"].values():
            if digest(artifact["path"]) != artifact["sha256"]:
                raise ValueError("checked pilot artifact changed")
        counted_receipts = {str(Path(item["path"]).parent / "requests.jsonl") for item in prior}
        if not pilot.get("request_sources") or any(source["path"] not in counted_receipts for source in pilot["request_sources"]):
            raise ValueError("pilot requests are absent from --prior-generation accounting chain")
        pilot_evidence = source_identity(pilot_path)
    cases = S.training_cases(manifest, args.seed, per_cell=args.per_cell, offset=args.offset) if args.per_cell else []
    # Extra coverage is drawn once. Isolation failures are recorded, not resampled.
    for index, (_, metadata) in enumerate(G.curriculum_jobs(args.curriculum_chats, args.diaries,
                                                          args.seed + 1, with_receipts=True)):
        rng = random.Random(metadata["scenario_seed"])
        st = copy.deepcopy(metadata["state"])
        scene = metadata["scenario"]
        if scene["focus"] != "diary":
            scene = G.curriculum_scenario(st, scene["focus"], scene["variant"], rng)
        identity = value_hash({"seed": args.seed, "offset": args.offset, "index": index, "state": st, "scene": scene})
        cases.append({"case_id": "expansion/" + identity, "group_id": "train/" + identity,
                      "template_id": "train/curriculum/" + scene["focus"] + "/" + scene["variant"],
                      "family": scene["focus"], "stage": st["stage"], "state": st,
                      "scene": scene, "target": scene.get("target"), "gold": {}})
    if not cases or len(previous_calls) + len(cases) > 4096:
        raise ValueError("new-request cap exceeded (4096 includes pilot, errors and corrections)")
    if any(case["group_id"] in previous_groups for case in cases):
        raise ValueError("training group already requested; use a new offset/seed, never retry rejected scenes")
    if len({case["group_id"] for case in cases}) != len(cases):
        raise ValueError("duplicate requested training group")
    out = fresh_directory(args.out)
    teacher = G.Teacher(args.teacher_url.split(","), args.teacher_model)
    student = None
    if getattr(args, "student_export", None):
        student = load_correction_student(args)
    receipt = {"kind": "student-grounded-corrections" if student else "semantic-teacher-generation",
               "status": "running", "manifest": source_identity(args.manifest), "seed": args.seed,
               "offset": args.offset, "requested_scenes": len(cases), "prior_generations": prior,
               "pilot_audit": pilot_evidence,
               "semantic_grid": {"per_cell": args.per_cell, "stages": 4, "families": 4},
               "extra_requests": {"curriculum_chats": args.curriculum_chats, "diaries": args.diaries},
               "prior_api_calls": len(previous_calls), "api_calls": 0, "request_cap": 4096,
               "workers": args.workers, "completed_scenes": 0,
               "teacher": {"url": args.teacher_url, "model": args.teacher_model},
               "source_sha256": {name: digest(ROOT / "train" / name) for name in
                                  ("gen_data.py", "experiments.py", "semantic_data.py")},
               "scope": "Scripted initial snapshots, not timed-body verification; no semantic approval from grammar alone."}
    if args.teacher_artifact:
        receipt["teacher"]["artifact"] = source_identity(args.teacher_artifact)
    if student:
        receipt["student"] = student[1]
    save_json(out / "generation.json", receipt)
    accepted = 0
    ledger_lock = threading.Lock()
    with open(out / "api-calls.jsonl", "x") as ledger, open(out / "requests.jsonl", "x") as requests, open(out / "data.jsonl", "x") as data:
        def process_scene(index, original):
            case = copy.deepcopy(original)
            st, scene = case["state"], case["scene"]
            metadata = {"index": index, "case": case, "case_id": case["case_id"],
                        "group_id": case["group_id"], "template_id": case["template_id"],
                        "family": case["family"], "stage": case["stage"],
                        "state": copy.deepcopy(st), "scenario": copy.deepcopy(scene),
                        "state_sha256": value_hash(st), "scene_sha256": value_hash(scene)}
            scene_calls = 0
            def recorded_teacher(prompt, **kwargs):
                nonlocal scene_calls
                request_id = value_hash({"out": str(out), "case_id": case["case_id"], "call": scene_calls})
                scene_calls += 1
                with ledger_lock:
                    if len(previous_calls) + receipt["api_calls"] >= 4096:
                        raise ValueError("4096 total API request cap reached")
                    write_jsonl(ledger, {"event": "start", "request_id": request_id, "case_id": case["case_id"],
                                         "group_id": case["group_id"], "template_id": case["template_id"],
                                         "index": index, "prompt": prompt, "parameters": kwargs})
                    receipt["api_calls"] += 1
                    save_json(out / "generation.json", receipt)
                try:
                    raw = teacher(prompt, **kwargs)
                    with ledger_lock:
                        write_jsonl(ledger, {"event": "finish", "request_id": request_id, "raw": raw})
                    return raw
                except Exception as exc:
                    with ledger_lock:
                        write_jsonl(ledger, {"event": "finish", "request_id": request_id,
                                             "error": {"type": type(exc).__name__, "reason": str(exc)}})
                    raise
            try:
                issues = S.isolation_check({"text": json.dumps(case, ensure_ascii=False),
                                            "group_id": case["group_id"], "template_id": case["template_id"]}, manifest)
                if issues:
                    raise ValueError("reserved evaluation overlap: " + str(issues))
                if student:
                    attempt = student_scene_attempt(student[0], st, scene, case["case_id"])
                    metadata["student_attempt"] = attempt
                    if S.isolation_check(json.dumps(attempt, ensure_ascii=False), manifest):
                        raise ValueError("student output overlaps reserved evaluation values; not sent to teacher")
                    # The full replacement is teacher-authored and rendered from the
                    # untouched snapshot. No old APP result/event is ever spliced.
                    metadata["correction_scope"] = "full continuation replay from initial snapshot"
                    prefix = ("Correct the NEW training scene below. Original student attempt is fallible, not truth. "
                              "Return a FULL replacement using only the fixed scene truth.\nSTUDENT ATTEMPT\n" +
                              json.dumps(attempt, ensure_ascii=False) + "\nFIXED TASK\n")
                    def correction_teacher(prompt, **kwargs):
                        return recorded_teacher(prefix + prompt, **kwargs)
                    author = correction_teacher
                else:
                    author = recorded_teacher
                fn = (lambda source: G.make_curriculum_diary(source, copy.deepcopy(st))) if scene["focus"] == "diary" else (
                    lambda source: G.make_curriculum_chat(source, copy.deepcopy(st), scene))
                row, error, record = G.run_generation(author, fn, metadata)
            except Exception as exc:
                row, error = None, {"type": type(exc).__name__, "reason": str(exc)}
                record = {**metadata, "requests": [], "accepted": False, "error": error}
            if row is not None:
                row["semantic_source"] = {"case_id": case["case_id"], "group_id": case["group_id"],
                                           "template_id": case["template_id"], "manifest_sha256": receipt["manifest"]["sha256"]}
                # Structurally retained != semantically accepted. corpus-audit owns approval.
                record["structurally_valid"] = True
                record["semantic_status"] = "pending_grounded_audit"
            record["accepted"] = False
            return row, record
        for row, record in bounded_scene_results(enumerate(cases), process_scene, args.workers):
            if row is not None:
                write_jsonl(data, row)
                accepted += 1
            write_jsonl(requests, record)
            with ledger_lock:
                receipt["completed_scenes"] += 1
                receipt["structurally_valid"] = accepted
                save_json(out / "generation.json", receipt)
                print(f"SEMANTIC completed={receipt['completed_scenes']}/{len(cases)} "
                      f"api_calls={receipt['api_calls']} structurally_valid={accepted}", flush=True)
    receipt.update(status="complete_pending_audit", structurally_valid=accepted,
                   rejected=len(cases) - accepted, total_api_calls=len(previous_calls) + receipt["api_calls"])
    receipt["artifacts"] = {name: source_identity(out / name) for name in ("data.jsonl", "requests.jsonl", "api-calls.jsonl")}
    save_json(out / "generation.json", receipt)
    print(json.dumps(receipt, indent=2))


def load_correction_student(args):
    import torch
    from transformers import AutoConfig, AutoTokenizer, GenerationConfig
    from onnx_utils import CausalSession, model_path
    import eval_pet as V
    export = Path(args.student_export).expanduser().resolve()
    config = AutoConfig.from_pretrained(export, local_files_only=True)
    tokenizer = AutoTokenizer.from_pretrained(export, local_files_only=True)
    generation = GenerationConfig.from_pretrained(export, local_files_only=True)
    graph = model_path(export, args.dtype)
    torch.set_num_threads(args.threads)
    session = CausalSession(graph, config.to_dict(), threads=args.threads)
    return (V.Evaluator(None, tokenizer, torch.device("cpu"), args.seed, session, config, generation),
            {"export": str(export), "dtype": args.dtype, "graph": source_identity(graph),
             "files_sha256": {p.name: digest(p) for p in export.glob("*.json")},
             "session": session.metadata(), "seed": args.seed, "threads": args.threads})


def student_scene_attempt(evaluator, initial, scene, case_id):
    import copy
    import gen_data as G
    live, lines, turns = copy.deepcopy(initial), [], []
    if scene["focus"] == "diary":
        return {"initial_state": copy.deepcopy(initial), "diary": evaluator.diary(live, case_id + "/student")}
    for index, slot in enumerate(scene["slots"]):
        lines.extend(slot["events"])
        if f"* {live['pet']} woke up *" in slot["events"]:
            live["asleep"] = False
        if f"* {live['owner']} cleaned up the poop *" in slot["events"]:
            live["poop"] = 0
        try:
            turn = evaluator.pet_turn(live, lines, "react" if slot["events"] else "idle", f"{case_id}/student/{index}")
        except Exception as exc:
            turns.append({"slot_index": index, "failure": {"type": type(exc).__name__, "reason": str(exc)},
                          "state": copy.deepcopy(live), "lines": lines[:]})
            break
        turns.append(copy.deepcopy(turn))
        if turn["failure"] or not turn["delivered"] or not turn["contract_valid"]:
            break
        lines.append(live["pet"] + ": " + turn["text"])
        # Lookups above used pre-line state. Effects now commit together.
        for tag in G.W.TAG_RE.finditer(turn["text"]):
            if tag.group(1) == "remember" and tag.group(2) not in live["memories"]:
                live["memories"].append(tag.group(2))
            if tag.group(1) == "gift":
                live["shelf"].append(tag.group(2))
        if slot["mode"] == "OFFER" and "[yes]" in turn["text"]:
            if live["asleep"]:
                lines.append(f"* {live['pet']} woke up *")
                live["asleep"] = False
            if "catch_stars" in slot:
                lines.append(f"* {live['owner']} played catch with {live['pet']} ({slot['catch_stars']} stars) *")
        if "[sleep]" in turn["text"]:
            live["asleep"] = True
    return {"initial_state": copy.deepcopy(initial), "turns": turns, "lines": lines,
            "final_scripted_state": live}


def recover_producer(row, producer):
    """Validate retained snapshot and raw model output without rewriting a row."""
    import copy
    import gen_data as G
    st = producer.get("state")
    if not isinstance(st, dict):
        raise ValueError("missing producer snapshot")
    header = G.W.header(row["kind"], st)
    if not row["text"].startswith(header):
        raise ValueError("producer snapshot/header mismatch")
    if row["kind"] == "diary":
        reproduced = False
        for request in producer.get("requests", []):
            raw = request.get("raw")
            if not isinstance(raw, str):
                continue
            try:
                if isinstance(json.loads(raw), dict):
                    replay = G.make_curriculum_diary(lambda *args, **kwargs: raw, copy.deepcopy(st))
                    reproduced = replay["text"] == row["text"]
            except (ValueError, KeyError, TypeError):
                pass
            if not reproduced:
                try:
                    reproduced = G.check_diary(st, raw)[0] == row["text"]
                except ValueError:
                    pass
            if reproduced:
                break
        if not reproduced:
            for turn in producer.get("turns", []):
                if turn.get("text"):
                    try:
                        reproduced = G.check_diary(st, turn["text"])[0] == row["text"]
                    except ValueError:
                        continue
                    if reproduced:
                        break
        if not reproduced:
            raise ValueError("missing exact raw teacher diary provenance")
        return {"state_sha256": value_hash(st), "replay": "exact teacher diary text and canonical mask checked"}
    scene = producer.get("scenario")
    if not isinstance(scene, dict) or not scene.get("slots"):
        raise ValueError("missing producer scene slots")
    replies = None
    for request in producer.get("requests", []):
        try:
            raw = json.loads(request.get("raw", ""))
        except (ValueError, TypeError):
            continue
        if isinstance(raw, list) and all(isinstance(reply, str) for reply in raw):
            replies = raw
            break
    if replies is None and producer.get("turns"):
        replies = [G.W.TAG_RE.sub(lambda match: match.group(0).removesuffix(match.group(3) or ""), turn["text"])
                   for turn in producer["turns"]]
    if replies is None:
        raise ValueError("missing raw teacher continuation")
    replay = G.replay_curriculum(copy.deepcopy(st), scene, replies)
    if replay["text"] != row["text"]:
        raise ValueError("producer replay text/APP result mismatch")
    if G._span_coverage(replay["spans"], len(row["text"])) != G._span_coverage(row["spans"], len(row["text"])):
        raise ValueError("producer replay mask mismatch")
    return {"state_sha256": value_hash(st), "scene_sha256": value_hash(scene),
            "replies_sha256": value_hash(replies), "replay": "exact text and mask-v2 coverage"}


def shared_producer_replay(row, producer, engine):
    """Authenticate pre-line lookups and post-line effects with the real core."""
    import copy
    import gen_data as G
    st = copy.deepcopy(producer["state"])
    save = engine.save_for_state(st)
    query = next((tag.group(2) for tag in G.W.TAG_RE.finditer(row["text"]) if tag.group(1) == "recall"), "")
    parity = engine.parity(st, query)
    evidence = []
    prefix = st["pet"] + ": "
    for line in row["text"].partition("---\n")[2].splitlines():
        if line == f"* {st['owner']} cleaned up the poop *":
            st["poop"] = save["poop"] = 0
        if line == f"* {st['pet']} woke up *":
            st["asleep"] = save["asleep"] = False
        if not line.startswith(prefix):
            continue
        body = line[len(prefix):]
        tags = list(G.W.TAG_RE.finditer(body))
        for tag in tags:
            name, argument, result = tag.groups()
            if name in G.W.LOOKUP_TOOLS and result != "{" + G.W.tool_result(st, name, argument) + "}":
                raise ValueError("shared producer replay stale pre-line lookup result")
        committed = engine.replay(save, [body])
        for tag in tags:
            if tag.group(1) == "remember" and tag.group(2).casefold() not in [fact.casefold() for fact in committed["after"]["memories"]]:
                raise ValueError("shared engine did not persist the complete remember effect")
        gift_writes = [tag.group(2) for tag in tags if tag.group(1) == "gift"]
        gift_effects = [effect["arg"] for step in committed["steps"]
                        for effect in step["engine"]["effects"] if effect["kind"] == "gift"]
        if gift_effects != gift_writes:
            raise ValueError("shared engine did not commit every gift with its exact multiplicity")
        if gift_writes:
            shelf = committed["after"]["shelf"]
            expected_shelf = save["shelf"] + gift_writes
            if not shelf or shelf != expected_shelf[-len(shelf):]:
                raise ValueError("shared engine gift shelf transition differs from committed gifts")
        evidence.append(committed)
        save = committed["after"]
        st["memories"], st["shelf"] = save["memories"][:], save["shelf"][:]
    return {"binary_sha256": engine.sha256, "parity": parity, "turns": evidence,
            "scope": "actual core line effects; fixed scene cleanup/wake events are scripted, not timed-body verification"}


def grounded_review(review, row, producer):
    """An explicit source-linked judgment, never a field-presence pseudo-oracle."""
    if not review or review.get("decision") != "accept":
        raise ValueError("missing/rejected/uncertain independent grounding review")
    if not review.get("reviewer") or not review.get("evidence"):
        raise ValueError("grounding review lacks reviewer/evidence provenance")
    if review.get("state_sha256") != value_hash(producer["state"]):
        raise ValueError("grounding review is not linked to producer slot truth")
    required = ("events", "actors", "gift_direction", "game_counts", "unsupported_claims") if row["kind"] == "diary" else (
        "owner_attribution", "values", "visible_answers", "unsupported_claims", "persistent_notes")
    if any(review.get("checks", {}).get(key) is not True for key in required):
        raise ValueError("grounding review incomplete or ambiguous: " + ",".join(required))
    return review


def corpus_audit(args):
    import gen_data as G
    import semantic_data as S
    manifest = S.load_manifest(args.manifest)
    engine = S.EngineAdapter(args.engine_binary)
    out = fresh_directory(args.out)
    identities = [source_identity(path) for path in [args.data, args.manifest, *args.requests, *args.reviews]]
    producers, reviews = {}, {}
    for path in args.requests:
        identity = source_identity(path)
        for number, producer in jsonl_rows(path):
            key = producer.get("text_sha256")
            if key:
                producers.setdefault(key, []).append((producer, {**identity, "line": number}))
    for path in args.reviews:
        identity = source_identity(path)
        for number, review in jsonl_rows(path):
            key = review.get("text_sha256")
            if key:
                reviews.setdefault(key, []).append((review, {**identity, "line": number}))
    admitted, excluded, cells = 0, 0, {}
    with open(out / "data.jsonl", "x") as data, open(out / "audit.jsonl", "x") as audit, open(out / "review-inputs.jsonl", "x") as pending:
        for number, row in jsonl_rows(args.data):
            key = text_hash(row)
            record = {"text_sha256": key, "source": identities[0], "line": number,
                      "row_sha256": value_hash(row), "accepted": False}
            candidates = producers.get(key, [])
            try:
                G.recheck(row)
                issues = S.isolation_check(row, manifest)
                if issues:
                    raise ValueError("reserved evaluation overlap: " + str(issues))
                if not candidates:
                    raise ValueError("no source-linked producer receipt")
                valid = []
                errors = []
                for producer, identity in candidates:
                    try:
                        replay = recover_producer(row, producer)
                        valid.append((producer, identity, replay))
                    except (ValueError, KeyError, TypeError) as exc:
                        errors.append({"source": identity, "reason": str(exc)})
                if not valid:
                    record["producer_errors"] = errors
                    raise ValueError("no verifiable producer snapshot/replay")
                producer, identity, replay = valid[0]
                legacy_scene = producer.get("scenario", {})
                group_id = producer.get("group_id") or "legacy/" + value_hash({"state": producer["state"], "scene": legacy_scene})
                template_id = producer.get("template_id") or "legacy/" + legacy_scene.get("focus", row["kind"]) + "/" + legacy_scene.get("variant", "unclassified")
                if S.isolation_check({**row, "group_id": group_id, "template_id": template_id}, manifest):
                    raise ValueError("reserved evaluation group/template identity")
                record.update(producer=identity, replay=replay, group_id=group_id, template_id=template_id)
                write_jsonl(pending, {"text_sha256": key, "row": row, "state": producer["state"],
                                     "state_sha256": value_hash(producer["state"]), "scene": producer.get("scenario"),
                                     "case": producer.get("case"), "producer": identity})
                # Full exact semantic checks are sufficient only when every slot is
                # decidable. Open prose/diaries require independent grounded review.
                semantic = None
                if row["kind"] == "chat":
                    record["shared_engine_replay"] = shared_producer_replay(row, producer, engine)
                if producer.get("case", {}).get("gold", {}).get("slots") and row["kind"] == "chat":
                    semantic = S.check_scene(producer["case"], row, engine)
                    record["semantic"] = semantic
                    if semantic.get("status") == "fail":
                        raise ValueError("shared-engine semantic scene failure")
                decisions = semantic.get("judgments", []) if isinstance(semantic, dict) else []
                exact_pass = bool(decisions) and all(item.get("status") == "pass" for item in decisions)
                review_list = reviews.get(key, [])
                if not exact_pass:
                    approved = []
                    for review, review_identity in review_list:
                        try:
                            grounded_review(review, row, producer)
                            approved.append(review_identity)
                        except ValueError:
                            continue
                    if not approved:
                        raise ValueError("requires source-linked independent grounding review")
                    record["grounding_reviews"] = approved
                if any(review.get("decision") in ("reject", "uncertain") for review, _ in review_list):
                    raise ValueError("conflicting/rejected/uncertain grounding review")
                # A hard deterministic failure cannot be overruled by prose review.
                if any(item.get("status") == "fail" for item in decisions):
                    raise ValueError("shared-engine semantic scene failure")
                record["accepted"] = True
                write_jsonl(data, row)
                admitted += 1
                family = producer.get("family", producer.get("case", {}).get("family"))
                if family:
                    cell = row.get("stage", producer.get("stage", "unknown")) + "/" + family
                    cells[cell] = cells.get(cell, 0) + 1
            except (ValueError, KeyError, TypeError) as exc:
                record["reason"] = str(exc)
                excluded += 1
            write_jsonl(audit, record)
    receipt = {"status": "complete", "kind": "source-linked-corpus-audit", "sources": identities,
               "admitted": admitted, "excluded": excluded, "accepted_cells": cells,
               "manifest": source_identity(args.manifest), "no_text_or_mask_repair": True,
               "request_sources": [source_identity(path) for path in args.requests],
               "engine": {"path": str(engine.binary), "sha256": engine.sha256},
               "source_sha256": {name: digest(ROOT / "train" / name) for name in
                                  ("gen_data.py", "experiments.py", "semantic_data.py")},
               "artifacts": {name: source_identity(out / name) for name in ("data.jsonl", "audit.jsonl", "review-inputs.jsonl")}}
    if args.pilot:
        expected = [stage + "/" + family for stage in G.W.STAGES for family in
                    ("capture", "header_recall", "older_recall", "abstention")]
        requested_cells = {}
        pilot_errors = []
        if len(args.requests) != 1:
            pilot_errors.append("pilot requires one original 512-request generation receipt")
        for path in args.requests:
            generation_path = Path(path).parent / "generation.json"
            generation = load_json(generation_path) if generation_path.is_file() else {}
            if (generation.get("kind") != "semantic-teacher-generation" or generation.get("requested_scenes") != 512 or
                    generation.get("semantic_grid", {}).get("per_cell") != 32 or generation.get("offset") != 0 or
                    generation.get("prior_api_calls") != 0 or generation.get("api_calls") != 512 or
                    generation.get("manifest", {}).get("sha256") != digest(args.manifest)):
                pilot_errors.append("pilot is not the original complete 4x4x32 teacher request grid")
            for _, request in jsonl_rows(path):
                cell = request.get("stage", "unknown") + "/" + request.get("family", "unknown")
                requested_cells[cell] = requested_cells.get(cell, 0) + 1
        if set(requested_cells) != set(expected) or any(requested_cells.get(cell) != 32 for cell in expected):
            pilot_errors.append("pilot requested cell counts are not exactly32")
        receipt["pilot_gate"] = {"passed": not pilot_errors and all(cells.get(cell, 0) >= 16 for cell in expected),
                                 "minimum_per_cell": 16, "counts": {cell: cells.get(cell, 0) for cell in expected},
                                 "requested_cells": requested_cells, "errors": pilot_errors}
    for identity in identities:
        if digest(identity["path"]) != identity["sha256"]:
            raise ValueError("audit source changed while reading")
    if digest(engine.binary) != engine.sha256:
        raise ValueError("shared engine binary changed during source audit")
    save_json(out / "audit.json", receipt)
    print(json.dumps(receipt, indent=2))
    return 1 if args.pilot and not receipt["pilot_gate"]["passed"] else 0


def freeze_corpus(args):
    import gen_data as G
    import semantic_data as S
    from model_data import heldout_split
    manifest = S.load_manifest(args.manifest)
    manifest_hash = digest(args.manifest)
    out = fresh_directory(args.out)
    rows, seen, sources, groups, exclusions = [], set(), [], [], []
    pilot_passed = False
    for directory in args.audits:
        directory = Path(directory).expanduser().resolve()
        audit = load_json(directory / "audit.json")
        if audit.get("status") != "complete" or audit.get("manifest", {}).get("sha256") != manifest_hash:
            raise ValueError("incomplete audit or evaluation isolation manifest mismatch")
        for artifact in audit["artifacts"].values():
            if digest(artifact["path"]) != artifact["sha256"]:
                raise ValueError("audited artifact changed")
        pilot_passed |= audit.get("pilot_gate", {}).get("passed", False)
        sources.append(source_identity(directory / "audit.json"))
        decisions = {record["text_sha256"]: record for _, record in jsonl_rows(directory / "audit.jsonl")}
        for record in decisions.values():
            if not record["accepted"]:
                exclusions.append(record)
        for _, row in jsonl_rows(directory / "data.jsonl"):
            G.recheck(row)
            key = text_hash(row)
            if not decisions.get(key, {}).get("accepted"):
                raise ValueError("row is not in the accepted source-linked audit")
            if S.isolation_check(row, manifest):
                raise ValueError("reserved evaluation leakage at freeze")
            if key in seen:
                exclusions.append({"text_sha256": key, "reason": "duplicate canonical text", "source": sources[-1]})
                continue
            seen.add(key)
            rows.append(row)
            groups.append({"text_sha256": key, "source": decisions[key], "semantic_source": row.get("semantic_source")})
    if not pilot_passed:
        raise ValueError("C1 requires the checked pilot's >=16 accepted chats in every cell")
    prior, calls, requested_groups = request_history(args.generations)
    if not calls or len(calls) > 4096:
        raise ValueError("missing generation request evidence or 4096 cap exceeded")
    for row in rows:
        source = row.get("semantic_source")
        if source and source.get("group_id") not in requested_groups:
            raise ValueError("admitted new row has no request in the generation accounting chain")
    train, heldout = heldout_split(rows, args.split_seed)
    with open(out / "data.jsonl", "x") as stream:
        for row in rows:
            write_jsonl(stream, row)
    for name, values in (("groups.jsonl", groups), ("exclusions.jsonl", exclusions)):
        with open(out / name, "x") as stream:
            for value in values:
                write_jsonl(stream, value)
    receipt = {"status": "frozen", "name": "C1", "audits": sources,
               "manifest": source_identity(args.manifest), "generation_chain": prior,
               "total_new_api_calls": len(calls), "request_cap": 4096,
               "dataset": verify_dataset(out / "data.jsonl"),
               "artifacts": {name: source_identity(out / name) for name in ("data.jsonl", "groups.jsonl", "exclusions.jsonl")},
               "ce_split": {"seed": args.split_seed, "algorithm": "model_data.heldout_split unchanged",
                            "train_text_sha256": [text_hash(row) for row in train],
                            "heldout_text_sha256": [text_hash(row) for row in heldout]},
               "semantic_evaluation_is_not_ce_split": True}
    save_json(out / "corpus.json", receipt)
    print(json.dumps(receipt, indent=2))


def stage(args):
    """Expose private export trees to a static preview without copying weights."""
    out = fresh_directory(args.out)
    manifest = load_json(Path(args.run) / "experiment.json")
    exports = []
    for row in manifest["models"]:
        if row.get("status") != "complete":
            continue
        model_id = row["id"]
        if not model_id or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" for c in model_id):
            raise ValueError("unsafe model ID in experiment manifest")
        for dtype, export in row["exports"].items():
            if dtype not in {"q4", "q8", "fp32"}:
                raise ValueError("unsupported export dtype")
            source = Path(export["path"]).resolve()
            graph = source / "onnx" / {"q4": "model_q4.onnx", "q8": "model_quantized.onnx", "fp32": "model.onnx"}[dtype]
            # HF registry IDs support at most namespace/name, not three levels.
            # A caller can use experiments/<id>-<dtype> under the preview root.
            target = out / f"{model_id}-{dtype}"
            target.parent.mkdir(parents=True, exist_ok=True)
            graph_hash = digest(graph)
            target.symlink_to(source, target_is_directory=True)
            exports.append({"id": model_id, "dtype": dtype, "path": str(source),
                            "model_suffix": f"{model_id}-{dtype}", "graph_sha256": graph_hash})
    if not exports:
        raise ValueError("no completed model exports to stage")
    receipt = {"status": "complete", "run": str(Path(args.run).resolve()), "exports": exports}
    save_json(out / "stage.json", receipt)
    print(json.dumps(receipt, indent=2))


def require_rate(report, name):
    counts = report.get("rate_counts", {}).get(name)
    value = report.get("rates", {}).get(name)
    if not isinstance(counts, dict):
        raise ValueError(f"missing denominator for {name}")
    numerator, denominator = counts.get("numerator"), counts.get("denominator")
    if (isinstance(numerator, bool) or isinstance(denominator, bool)
            or not isinstance(numerator, (int, float))
            or not isinstance(denominator, (int, float))
            or not math.isfinite(numerator) or not math.isfinite(denominator)
            or denominator <= 0 or not 0 <= numerator <= denominator):
        raise ValueError(f"invalid counts for {name}")
    if (isinstance(value, bool) or not isinstance(value, (int, float))
            or not math.isfinite(value) or not 0 <= value <= 1
            or not math.isclose(value, numerator / denominator, abs_tol=1e-9)):
        raise ValueError(f"invalid rate for {name}")
    return value, denominator


def quality_gate(reference, candidate, policy):
    """Conservative finite-suite comparison, not a global quality guarantee."""
    reasons = []
    for label, report in (("reference", reference), ("candidate", candidate)):
        if report.get("status") != "complete":
            reasons.append(f"{label} behavior report is not complete")
    # Runner embeds identical explicit evaluation arguments. Model and backend
    # identities intentionally differ, but scenario budgets may not.
    ref_contract = reference.get("experiment_contract")
    can_contract = candidate.get("experiment_contract")
    if not ref_contract or ref_contract != can_contract:
        reasons.append("different or missing evaluation experiment contracts")
    if not reference.get("scenario_sha256") or reference.get("scenario_sha256") != candidate.get("scenario_sha256"):
        reasons.append("different or missing actual scenario hashes")
    if not reference.get("dataset_sha256") or reference.get("dataset_sha256") != candidate.get("dataset_sha256"):
        reasons.append("different or missing training dataset hashes")
    if reference.get("config", {}).get("device") != candidate.get("config", {}).get("device"):
        reasons.append("different sampling devices")
    margin = policy.get("margin", 0)
    if not isinstance(margin, (int, float)) or not math.isfinite(margin) or not 0 <= margin <= 1:
        raise ValueError("quality margin must be finite and between zero and one")
    for direction in ("minimum", "maximum"):
        for name in policy[direction]:
            try:
                base, base_n = require_rate(reference, name)
                value, count = require_rate(candidate, name)
                # Raw attempt count varies with the real three-retry policy;
                # scenario identity/budgets, not retry count, establish fairness.
                if count != base_n and name != "cute_filter_rejections":
                    reasons.append(f"{name}: unequal denominator")
                if direction == "minimum" and value + margin < base:
                    reasons.append(f"{name}: {value:.4f} below reference {base:.4f}")
                if direction == "maximum" and value - margin > base:
                    reasons.append(f"{name}: {value:.4f} above reference {base:.4f}")
            except ValueError as e:
                reasons.append(str(e))
    return {"passed": not reasons, "reasons": reasons}


def published_inference_gate(reference, candidate, policy, published_graph_sha256):
    """Paired inference only. Never manufacture a published training-data hash."""
    reasons = []
    if len(published_graph_sha256) != 64 or any(c not in "0123456789abcdef" for c in published_graph_sha256):
        raise ValueError("published graph SHA-256 must be explicitly pinned")
    identity = reference.get("onnx_artifact", {})
    if identity.get("sha256", identity.get("graph_sha256")) != published_graph_sha256:
        reasons.append("reference is not the pinned actual published graph")
    for label, report in (("reference", reference), ("candidate", candidate)):
        if report.get("status") != "complete" or not report.get("onnx_artifact"):
            reasons.append(label + " is not complete actual ONNX inference evidence")
    for key in ("experiment_contract", "scenario_sha256", "scenario_suite_sha256", "scenario_counts"):
        if not reference.get(key) or reference.get(key) != candidate.get(key):
            reasons.append("different or missing paired " + key)
    # Lock generation/sampling, but do not equate graph/model identities or the
    # historical v6 corpus (which is unavailable) with C1's training receipt.
    ref_config, can_config = reference.get("config", {}), candidate.get("config", {})
    for key in ("seed", "device", "threads", "generation"):
        if key not in ref_config or ref_config.get(key) != can_config.get(key):
            reasons.append("different or missing paired sampling " + key)
    for key in ("temperature", "top_k", "top_p", "repetition_penalty", "seed_strategy"):
        sampling = ref_config.get("sampling", {})
        if key not in sampling or sampling.get(key) != can_config.get("sampling", {}).get(key):
            reasons.append("different or missing paired sampling " + key)
    for label, config in (("reference", ref_config), ("candidate", can_config)):
        if not config.get("sampling", {}).get("eos_token_ids"):
            reasons.append(label + " lacks actual EOS identity")
    margin = policy.get("margin", 0)
    if margin != 0:
        raise ValueError("published paired inference preserves zero-margin behavior policy")
    for direction in ("minimum", "maximum"):
        for name in policy[direction]:
            try:
                base, base_n = require_rate(reference, name)
                value, count = require_rate(candidate, name)
                if count != base_n and name != "cute_filter_rejections":
                    reasons.append(name + ": unequal denominator")
                if direction == "minimum" and value < base or direction == "maximum" and value > base:
                    reasons.append(name + ": regression against actual published inference")
            except ValueError as exc:
                reasons.append(str(exc))
    return {"passed": not reasons, "reasons": reasons, "kind": "published-paired-inference",
            "published_graph_sha256": published_graph_sha256, "same_training_corpus_claim": False}


def paired_inference(args):
    out = fresh_directory(args.out)
    reference, candidate = load_json(args.reference), load_json(args.candidate)
    result = published_inference_gate(reference, candidate, load_json(args.plan)["quality"], args.published_graph_sha256)
    result["sources"] = {"reference": source_identity(args.reference), "candidate": source_identity(args.candidate),
                         "policy": source_identity(args.plan)}
    save_json(out / "paired-inference.json", result)
    print(json.dumps(result, indent=2))
    return 0 if result["passed"] else 1


def evaluation_command(args, checkpoint, report, samples, diaries, cells, seed, export=None, dtype=None):
    # Use CPU sampling for both backends; CUDA/ROCm generators have different
    # random streams and would confound model/quantization comparisons.
    argv = [args.export_python,
            ROOT / "train/eval_pet.py", checkpoint, samples, report,
            "--device", "cpu",
            "--threads", args.threads, "--seed", seed,
            "--diary-count", diaries, "--cell-count", cells]
    if export is not None:
        argv += ["--onnx", export, "--onnx-dtype", dtype]
    else:
        argv += ["--dtype", "float32"]
    return argv


def stamp_behavior(report_path, data_hash, contract):
    report = load_json(report_path)
    report["dataset_sha256"] = data_hash
    report["experiment_contract"] = contract
    save_json(report_path, report)


def run(args):
    plan = load_json(args.plan)
    if plan.get("version") != 1:
        raise ValueError("unsupported experiment plan version")
    dataset = verify_dataset(args.data)
    out = fresh_directory(args.out)
    save_json(out / "plan.json", plan)
    save_json(out / "dataset.json", dataset)
    selected = set(args.models.split(",")) if args.models else {m["id"] for m in plan["models"]}
    unknown = selected - {m["id"] for m in plan["models"]}
    if unknown:
        raise ValueError(f"unknown model IDs: {sorted(unknown)}")
    training, evaluation = plan["training"], plan["evaluation"]
    samples = args.samples if args.samples is not None else evaluation["samples"]
    diaries = args.diaries if args.diaries is not None else evaluation["diaries"]
    cells = args.cell_samples if args.cell_samples is not None else evaluation["cell_samples"]
    from model_data import evaluation_contract
    contract = evaluation_contract(plan["seed"], samples, diaries, cells)
    manifest = {"status": "running", "dataset": dataset, "evaluation": contract,
                "screening": args.max_steps is not None,
                "plan_sha256": digest(args.plan), "models": []}
    save_json(out / "experiment.json", manifest)
    for spec in plan["models"]:
        if spec["id"] not in selected:
            continue
        work = out / spec["id"]
        work.mkdir()
        save_json(work / "spec.json", spec)
        row = {"id": spec["id"], "role": spec["role"], "status": "running",
               "spec": spec, "commands": [], "artifacts": {}}
        manifest["models"].append(row)
        save_json(out / "experiment.json", manifest)
        try:
            if digest(args.data) != dataset["sha256"]:
                raise ValueError("input dataset changed during the experiment")
            prepared = work / "prepared"
            prepare = [args.python, ROOT / "train/prepare_model.py", "--spec", work / "spec.json",
                       "--data", args.data, "--out", prepared, "--seed", plan["seed"]]
            if spec["kind"] == "llama" and args.pretrain_data:
                prepare += ["--tokenizer-data", args.pretrain_data]
            row["commands"].append(command(prepare, work / "prepare.log"))
            row["artifacts"]["preparation"] = str(prepared / "preparation.json")
            base = prepared
            if spec["kind"] == "llama" or args.pretrain_warm:
                if not args.pretrain_data:
                    raise ValueError("language pretraining requires --pretrain-data (scratch language pretraining is never silently skipped)")
                pretrain = work / "pretrained"
                row["commands"].append(command(
                    [args.python, ROOT / "train/train.py", "--pretrain-data", args.pretrain_data,
                     "--base", base, "--out", pretrain, "--device", args.device,
                     "--threads", args.threads, "--seed", plan["seed"],
                     "--split-seed", plan["split_seed"], "--max-steps", args.pretrain_steps,
                     "--lr", args.pretrain_lr, "--bs", training["batch_size"],
                     "--max-len", training["max_length"], "--eval-every", args.eval_every], work / "pretrain.log"))
                base = pretrain
                row["artifacts"]["pretraining"] = str(pretrain / "training.json")
            checkpoint = work / "checkpoint"
            train = [args.python, ROOT / "train/train.py", "--data", args.data, "--base", base,
                     "--out", checkpoint, "--device", args.device, "--threads", args.threads,
                     "--seed", plan["seed"], "--split-seed", plan["split_seed"],
                     "--epochs", training["epochs"], "--lr", training["lr"],
                     "--bs", training["batch_size"], "--max-len", training["max_length"],
                     "--history-aug", training["history_aug"], "--late-answer", training["late_answer"],
                     "--eval-every", args.eval_every]
            if args.max_steps is not None:
                train += ["--max-steps", args.max_steps]
            row["commands"].append(command(train, work / "train.log"))
            row["artifacts"]["training"] = str(checkpoint / "training.json")
            if load_json(checkpoint / "training.json")["data"]["sha256"] != dataset["sha256"]:
                raise ValueError("trainer consumed a different dataset")
            report = work / "behavior-torch.json"
            row["commands"].append(command(evaluation_command(args, checkpoint, report, samples, diaries, cells, plan["seed"]), work / "behavior-torch.log"))
            stamp_behavior(report, dataset["sha256"], contract)
            row["artifacts"]["behavior"] = str(report)
            row["exports"] = {}
            for dtype in args.dtypes.split(","):
                if dtype not in ("q4", "q8", "fp32"):
                    raise ValueError(f"unsupported dtype: {dtype}")
                exported = work / dtype
                env = dict(os.environ)
                env["EXP"] = str(Path(args.export_python).expanduser().absolute().parent.parent)
                row["commands"].append(command([ROOT / "train/export.sh", checkpoint, exported, dtype], work / f"export-{dtype}.log", env))
                with open(args.data, encoding="utf-8") as source:
                    probe_row = json.loads(next(source))
                header = probe_row["text"].partition("---\n")[0] + "---\n"
                pet = next(line[5:].split(" · ")[0] for line in header.splitlines() if line.startswith("pet: "))
                probe = header + ("dear diary," if probe_row["kind"] == "diary" else pet + ":")
                row["commands"].append(command(
                    [args.export_python, ROOT / "train/check_onnx.py", exported, probe, checkpoint,
                     "--dtype", dtype, "--threads", args.threads, "--tokens", 16,
                     "--report", work / f"cache-{dtype}.json"], work / f"cache-{dtype}.log"))
                quant_report = work / f"quant-{dtype}.json"
                row["commands"].append(command(
                    [args.export_python, ROOT / "train/quant_eval.py", checkpoint, exported, args.data,
                     args.quant_rows, "--dtype", dtype, "--threads", args.threads,
                     "--report", quant_report], work / f"quant-{dtype}.log"))
                quant_behavior = work / f"behavior-{dtype}.json"
                row["commands"].append(command(evaluation_command(args, checkpoint, quant_behavior, samples, diaries, cells, plan["seed"], exported, dtype), work / f"behavior-{dtype}.log"))
                stamp_behavior(quant_behavior, dataset["sha256"], contract)
                row["exports"][dtype] = {"path": str(exported), "behavior": str(quant_behavior),
                                         "quantization": str(quant_report), "cache": str(work / f"cache-{dtype}.json")}
            row["status"] = "complete"
        except Exception as e:
            row.update(status="failed", error=str(e))
            print(f"FAILED {spec['id']}: {e}", file=sys.stderr, flush=True)
        save_json(out / "experiment.json", manifest)
    manifest["status"] = "complete" if all(m["status"] == "complete" for m in manifest["models"]) else "failed"
    save_json(out / "experiment.json", manifest)
    print(json.dumps({"status": manifest["status"], "run": str(out), "models": [{"id": m["id"], "status": m["status"]} for m in manifest["models"]]}, indent=2))
    return 0 if manifest["status"] == "complete" else 1


def reviewed(reviews, key, path):
    review = reviews.get(key, {})
    return (review.get("approved") is True and review.get("report_sha256") == digest(path)
            and isinstance(review.get("notes"), str) and bool(review["notes"].strip()))


def runtime_score(report, required_flags=None):
    if report.get("status") != "complete" or not report.get("graph_sha256"):
        raise ValueError("runtime report is incomplete or lacks a measured graph hash")
    scores = []
    for cell in report.get("cases", []):
        if cell.get("status") != "complete":
            continue
        if required_flags and any(cell.get("config", {}).get(k) != v for k, v in required_flags.items()):
            continue
        if not cell.get("cross_origin_isolated"):
            raise ValueError("runtime cell was not cross-origin isolated")
        result = cell.get("result", {})
        cfg = result.get("config", {})
        if (type(cfg.get("seed")) is not int or cfg["seed"] != cell.get("config", {}).get("seed")
                or cfg.get("prompts") != "both" or cfg.get("mode") != "both" or cfg.get("repeats", 0) < 3):
            raise ValueError("runtime selection requires a recorded matching seed, both scenarios/modes and at least three repetitions")
        expected = {(scenario, mode) for scenario in ("short_idle", "long_react") for mode in ("greedy", "sampled")}
        if {(r.get("scenario"), r.get("mode")) for r in result.get("results", [])} != expected:
            raise ValueError("runtime scenario coverage is incomplete")
        value = result.get("summary", {}).get("sampled_turn_ms")
        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:
            raise ValueError("runtime cell lacks finite complete sampled-turn timing")
        hardware = cell.get("result", {}).get("hardware", {})
        gpu = hardware.get("webgpu") or {}
        if report.get("device") == "webgpu" and (gpu.get("software_adapter_detected")
                                                  or not gpu.get("adapter_info_available")):
            raise ValueError("software or unidentified WebGPU is not a verified hardware-runtime result")
        scores.append((value, cell["config"]))
    if not scores:
        raise ValueError("runtime report has no successful timing cells")
    score, config = min(scores, key=lambda pair: pair[0])
    return {"sampled_turn_ms": score, "config": config,
            "graph_bytes": report["graph_bytes"], "graph_sha256": report["graph_sha256"],
            "browser_version": report["browser_version"]}


def pareto_frontier(variants):
    dimensions = ("parameters", "graph_bytes", "sampled_turn_ms")
    return [candidate for candidate in variants if not any(
        other is not candidate and all(other[k] <= candidate[k] for k in dimensions)
        and any(other[k] < candidate[k] for k in dimensions) for other in variants)]


def compare(args):
    root = Path(args.run).resolve()
    manifest, plan = load_json(root / "experiment.json"), load_json(root / "plan.json")
    reference_rows = [r for r in manifest["models"] if r["role"] == "reference" and r["status"] == "complete"]
    reference_path = args.baseline or (reference_rows[0]["artifacts"]["behavior"] if reference_rows else None)
    if reference_path is None:
        raise ValueError("a complete same-corpus GPT-2 control or --baseline report is required")
    reference = load_json(reference_path)
    reviews = load_json(args.review) if args.review else {}
    runtime_reports = load_json(args.runtime) if args.runtime else {}
    behavior_reports = load_json(args.behavior) if args.behavior else {}
    cache_reports = load_json(args.cache) if args.cache else {}
    results, eligible = [], []
    for row in manifest["models"]:
        if row["role"] == "reference":
            continue
        item = {"id": row["id"], "eligible": False, "reasons": []}
        if row["status"] != "complete":
            item["reasons"].append(row.get("error", "incomplete model run"))
        elif manifest.get("screening"):
            item["reasons"].append("screening budget: not full quality acceptance")
        else:
            gate = quality_gate(reference, load_json(row["artifacts"]["behavior"]), plan["quality"])
            item.update(quality=gate)
            item["reasons"].extend(gate["reasons"])
            if plan["quality"].get("transcript_review_required") and not reviewed(reviews, row["id"], row["artifacts"]["behavior"]):
                item["reasons"].append("matching documented transcript review is required")
            parameters = load_json(row["artifacts"]["preparation"])["parameters"]["unique_total"]
            item["variants"] = []
            for dtype, export in row.get("exports", {}).items():
                variant = {"id": row["id"], "dtype": dtype, "eligible": False, "reasons": []}
                item["variants"].append(variant)
                try:
                    quant = load_json(export["quantization"])
                    graph = Path(export["path"]) / "onnx" / {"q4": "model_q4.onnx", "q8": "model_quantized.onnx", "fp32": "model.onnx"}[dtype]
                    graph_hash = digest(graph)
                    cache_path = cache_reports.get(row["id"], {}).get(dtype, export.get("cache"))
                    cache = load_json(cache_path) if cache_path else {}
                    if (cache.get("status") != "success" or cache.get("dtype") != dtype
                            or cache.get("artifact", {}).get("graph_sha256") != graph_hash
                            or not cache.get("incremental_kv", {}).get("allclose")):
                        variant["reasons"].append("successful graph-specific numerical/cache check is required")
                    relative = quant.get("relative_increase")
                    fp_loss, quant_loss = quant.get("fp_loss"), quant.get("quant_loss")
                    finite_losses = all(isinstance(v, (int, float)) and not isinstance(v, bool)
                                        and math.isfinite(v) for v in (fp_loss, quant_loss))
                    loss_consistent = (finite_losses and fp_loss > 0 and quant_loss >= 0
                                       and isinstance(relative, (int, float)) and not isinstance(relative, bool)
                                       and math.isfinite(relative)
                                       and math.isclose(relative, (quant_loss - fp_loss) / fp_loss, abs_tol=1e-9))
                    if (not loss_consistent or quant.get("status") != "complete" or quant.get("dataset_sha256") != manifest["dataset"]["sha256"]
                            or quant.get("dtype") != dtype or quant.get("graph_sha256") != graph_hash
                            or isinstance(relative, bool) or not isinstance(relative, (int, float))
                            or not math.isfinite(relative) or relative > plan["quality"]["maximum_quantization_loss_increase"]):
                        variant["reasons"].append("quantization loss or provenance gate failed")
                    behavior_path = behavior_reports.get(row["id"], {}).get(dtype, export["behavior"])
                    quant_behavior = load_json(behavior_path)
                    actual_graph = quant_behavior.get("onnx_artifact") or {}
                    if (quant_behavior.get("backend") != "onnx" or actual_graph.get("graph_sha256") != graph_hash
                            or actual_graph.get("dtype") != dtype
                            or type(actual_graph.get("disable_quant_qdq")) is not bool
                            or type(actual_graph.get("disable_prepacking")) is not bool):
                        variant["reasons"].append("behavior report did not qualify this actual ONNX graph/session")
                    flags = {"qdq": int(not actual_graph.get("disable_quant_qdq", True)),
                             "prepack": int(not actual_graph.get("disable_prepacking", True))}
                    cache_config = cache.get("configuration", {})
                    if (cache_config.get("qdq_fusion") is not bool(flags["qdq"])
                            or cache_config.get("prepacking") is not bool(flags["prepack"])):
                        variant["reasons"].append("cache check and behavioral report session flags differ")
                    quant_gate = quality_gate(reference, quant_behavior, plan["quality"])
                    variant["reasons"].extend(quant_gate["reasons"])
                    variant["behavior_report"] = behavior_path
                    if plan["quality"].get("transcript_review_required") and not reviewed(reviews, row["id"] + ":" + dtype, behavior_path):
                        variant["reasons"].append("matching quantized transcript review is required")
                    runtime_path = runtime_reports.get(row["id"], {}).get(dtype)
                    if not runtime_path:
                        variant["reasons"].append("actual browser runtime report is required")
                    else:
                        runtime = runtime_score(load_json(runtime_path), flags)
                        if runtime["graph_sha256"] != graph_hash or runtime["graph_bytes"] != graph.stat().st_size:
                            variant["reasons"].append("runtime measured a different exported graph")
                        variant.update(runtime, parameters=parameters, runtime_report=runtime_path)
                    variant["eligible"] = not item["reasons"] and not variant["reasons"]
                    if variant["eligible"]:
                        eligible.append(variant)
                except (ValueError, KeyError, OSError) as e:
                    variant["reasons"].append(str(e))
            item["eligible"] = any(v["eligible"] for v in item["variants"])
            if not item["variants"]:
                item["reasons"].append("no exported variants measured")
        results.append(item)
    frontier = pareto_frontier(eligible)
    selected = min(frontier, key=lambda v: (v["parameters"], v["graph_bytes"], v["sampled_turn_ms"])) if frontier else None
    report = {"reference": str(reference_path), "results": results, "pareto_frontier": frontier,
              "selected": selected, "scope": "finite-suite experiments; no inherited-v6 parity or global-minimum claim"}
    save_json(root / "comparison.json", report)
    print(json.dumps(report, indent=2))
    return 0


CAPABILITY_METRICS = {
    "valid_lines": ("valid_lines",),
    "care": ("offer_answers", "late_answers", "sleeping_offer_valid", "medicine_answers"),
    "memory": ("memory_saved", "memory_header", "memory_recall"),
    "sleep": ("sleep_idle_valid", "sleep_lights_off_valid", "sleep_message_valid", "sleep_pat_valid", "woken_valid"),
    "idle": ("quiet_idle",),
    "diary": ("diary_format_valid",),
}
SAFETY_METRICS = ("cute_filter_rejections", "unsafe_diaries", "food_full_yes",
                  "healthy_medicine_yes", "exhausted_nonstarving_food_yes",
                  "stuffed_hunger_claims", "unresolved_lookup_turns",
                  "model_written_lookup_turns", "dropped_turns")
GRAPH_NAMES = {"q4": "model_q4.onnx", "q8": "model_quantized.onnx", "fp32": "model.onnx"}


def capability_index(report):
    """Descriptive balanced visible-capability index, never an acceptance gate."""
    if report.get("status") != "complete":
        raise ValueError("behavior report is not complete")
    categories = {}
    for category, names in CAPABILITY_METRICS.items():
        values = [require_rate(report, name)[0] for name in names]
        categories[category] = {"percent": 100 * statistics.mean(values),
                                "metrics": {name: report["rate_counts"][name] for name in names}}
    return {"quality": statistics.mean(c["percent"] for c in categories.values()),
            "categories": categories}


def validate_benchmark_behavior(report, graph_hash, dtype, dataset_hash, contract, reference=None):
    if report.get("status") != "complete" or report.get("backend") != "onnx":
        raise ValueError("fresh ONNX behavior evaluation is incomplete")
    artifact = report.get("onnx_artifact") or {}
    if (artifact.get("graph_sha256") != graph_hash or artifact.get("dtype") != dtype
            or artifact.get("disable_quant_qdq") is not True
            or artifact.get("disable_prepacking") is not True):
        raise ValueError("behavior graph/dtype/session flags mismatch")
    if report.get("dataset_sha256") != dataset_hash or report.get("experiment_contract") != contract:
        raise ValueError("behavior dataset/seed/evaluation contract mismatch")
    if not report.get("scenario_sha256") or not report.get("scenario_counts"):
        raise ValueError("behavior lacks actual scenario provenance")
    config = report.get("config", {})
    if config.get("seed") != contract["seed"] or config.get("device") != "cpu":
        raise ValueError("behavior seed or sampling device mismatch")
    if reference is not None:
        for key in ("scenario_sha256", "scenario_suite_sha256", "scenario_counts", "experiment_contract", "dataset_sha256"):
            if not reference.get(key) or report.get(key) != reference[key]:
                raise ValueError(f"behavior {key} differs across variants")
        for key in ("sampling", "generation"):
            if not config.get(key) or config[key] != reference.get("config", {}).get(key):
                # EOS IDs describe the model, not a changed sampling workload.
                a, b = dict(config.get(key) or {}), dict(reference.get("config", {}).get(key) or {})
                if key == "sampling":
                    a.pop("eos_token_ids", None)
                    b.pop("eos_token_ids", None)
                if not a or a != b:
                    raise ValueError(f"behavior {key} contract differs across variants")


def rank_benchmark(rows):
    """Missing/failed rows cannot receive either descriptive rank."""
    valid = []
    for row in rows:
        row.update(rank=None, speed_rank=None)
        quality, speed, size = row.get("quality"), row.get("browser_fixed_decode_tps"), row.get("graph_bytes")
        if (row.get("status") == "complete" and not row.get("failures")
                and all(not isinstance(v, bool) and isinstance(v, (int, float)) and math.isfinite(v)
                        for v in (quality, speed, size)) and 0 <= quality <= 100 and speed > 0 and size > 0):
            valid.append(row)
    for rank, row in enumerate(sorted(valid, key=lambda r: (-r["quality"], -r["browser_fixed_decode_tps"], r["graph_bytes"], r["id"], r["dtype"])), 1):
        row["rank"] = rank
    for rank, row in enumerate(sorted(valid, key=lambda r: (-r["browser_fixed_decode_tps"], -r["quality"], r["graph_bytes"], r["id"], r["dtype"])), 1):
        row["speed_rank"] = rank


def finite_timing(value, name, positive=False):
    if (isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value)
            or value < 0 or (positive and value == 0)):
        raise ValueError(f"missing or invalid {name}")
    return value


def browser_benchmark_metrics(report, graph_hash, graph_bytes, args):
    if (report.get("status") != "complete" or report.get("graph_sha256") != graph_hash
            or report.get("graph_bytes") != graph_bytes or len(report.get("cases", [])) != 1):
        raise ValueError("browser measured a different graph or incomplete workload")
    cell = report["cases"][0]
    if cell.get("status") != "complete" or not cell.get("cross_origin_isolated"):
        raise ValueError("browser cell failed or lacked cross-origin isolation")
    result = cell.get("result", {})
    config = result.get("config", {})
    for key, value in {"seed": args.seed, "repeats": args.repeats, "warmup": args.warmup,
                       "qdq": 0, "prepack": 0, "threads": args.threads,
                       "workload": "both", "prompts": "both", "mode": "both"}.items():
        if config.get(key) != value:
            raise ValueError(f"browser {key} contract mismatch")
    fixed, pet = result.get("fixed_results", []), result.get("results", [])
    scenarios = {"short_idle", "long_react"}
    if len(fixed) != 2 or {r.get("scenario") for r in fixed} != scenarios:
        raise ValueError("browser fixed scenario coverage mismatch")
    if len(pet) != 4 or {(r.get("scenario"), r.get("mode")) for r in pet} != {(s, m) for s in scenarios for m in ("greedy", "sampled")}:
        raise ValueError("browser pet scenario coverage mismatch")
    decode_tokens, decode_ms, summaries = 0, 0, {}
    for entry in fixed:
        repeats = entry.get("repeats", [])
        if len(repeats) != args.repeats or entry.get("max_new_tokens") != 32 or entry.get("mode") != "greedy":
            raise ValueError("browser fixed token/repeat contract mismatch")
        for repeat in repeats:
            if repeat.get("generated_tokens") != 32 or repeat.get("decode_tokens") != 31 or repeat.get("stop_reason") != "fixed_length":
                raise ValueError("browser fixed workload stopped early")
            decode_tokens += repeat["decode_tokens"]
            decode_ms += finite_timing(repeat.get("decode_ms"), "browser decode_ms", True)
            finite_timing(repeat.get("prefill_ms"), "browser prefill_ms")
            if not repeat.get("prompt") or repeat["prompt"] != entry.get("prompt"):
                raise ValueError("browser fixed fitted prompts differ")
        summaries[entry["scenario"]] = entry["medians"]
    pet_summary = {}
    for entry in pet:
        if len(entry.get("repeats", [])) != args.repeats:
            raise ValueError("browser pet repetition count mismatch")
        finite_timing(entry.get("medians", {}).get("full_turn_ms"), "browser pet latency", True)
        pet_summary[f"{entry['scenario']}/{entry['mode']}"] = entry["medians"]
    return {"browser_fixed_decode_tps": 1000 * decode_tokens / decode_ms,
            "browser_fixed": summaries, "browser_pet": pet_summary,
            "browser_pet_counts": {f"{e['scenario']}/{e['mode']}": [{k: v for k, v in r.items()
                if k in ("safety_attempts", "lookups_resolved", "lookups_unresolved", "stop_reason", "silent", "completed", "generated_tokens", "final_content_tokens")}
                for r in e["repeats"]] for e in pet},
            "browser_load_ms": finite_timing(result.get("load_ms"), "browser load_ms", True),
            "browser_version": report.get("browser_version"), "browser_worklist": result.get("worklist"),
            "browser_hardware": result.get("hardware")}, fixed


def native_benchmark(args, export, dtype, fixed, work):
    """Same fitted prompts; pet timings are raw greedy generations, not app turns."""
    env = dict(os.environ)
    env.pop("SUBTOKENLING_GPU_DEVICE", None)
    env.update(ORT_DYLIB_PATH=str(Path(args.ort_library).expanduser().resolve()),
               SUBTOKENLING_MODEL_DIR=str(export), SUBTOKENLING_MODEL_DTYPE=dtype,
               SUBTOKENLING_THREADS=str(args.threads))
    receipt = {"status": "running", "threads": args.threads, "device": "cpu", "qdq": 0, "prepack": 0,
               "fixed_tokens": 32, "cases": [], "timing_scope": "raw native greedy generation, no app filter/retry/lookup loop"}
    for workload in ("fixed", "pet"):
        requests, labels = [], []
        for entry in fixed:
            budget = 40 if entry["scenario"] == "short_idle" else 64
            request = {"header": "", "lines": [], "prefix": entry["prompt"], "mode": "line",
                       "maxNew": 32 if workload == "fixed" else budget,
                       "temperature": 0, "topK": 40, "topP": 0.92, "ctx": entry["context"]}
            for i in range(args.warmup + args.repeats):
                requests.append(request)
                labels.append({"scenario": entry["scenario"], "workload": workload, "warmup": i < args.warmup})
        argv = [str(Path(args.native_probe).expanduser().resolve())]
        if workload == "fixed":
            argv += ["--benchmark-tokens", "32"]
        save_json(work / f"native-{workload}-inputs.json", requests)
        print(f"RUN native {dtype} {workload}: {' '.join(argv)}", flush=True)
        start = time.monotonic()
        result = subprocess.run(argv, input=json.dumps(requests), text=True, capture_output=True, cwd=ROOT, env=env, timeout=1200)
        (work / f"native-{workload}.log").write_text(result.stderr, encoding="utf-8")
        (work / f"native-{workload}-stdout.json").write_text(result.stdout, encoding="utf-8")
        if result.returncode:
            raise RuntimeError(f"native {workload} failed ({result.returncode}); see {work / f'native-{workload}.log'}")
        loads = [json.loads(line.removeprefix("BRAIN_PROBE_LOAD ")) for line in result.stderr.splitlines()
                 if line.startswith("BRAIN_PROBE_LOAD ")]
        if len(loads) != 1 or loads[0].get("threads") != args.threads:
            raise ValueError("native probe lacks matching session load receipt")
        finite_timing(loads[0].get("load_ms"), "native load_ms", True)
        receipt.setdefault("loads", {})[workload] = loads[0]
        outputs = json.loads(result.stdout)
        if not isinstance(outputs, list) or len(outputs) != len(requests):
            raise ValueError("native output count differs from requested workload")
        for label, request, output in zip(labels, requests, outputs):
            entry = next(e for e in fixed if e["scenario"] == label["scenario"])
            if output.get("prompt") != request["prefix"] or output.get("promptTokens") != entry["prompt_tokens"]:
                raise ValueError("native/browser fitted prompt or token count mismatch")
            if workload == "fixed" and (output.get("generatedTokens") != 32 or output.get("decodeTokens") != 31):
                raise ValueError("native fixed workload stopped early")
            finite_timing(output.get("prefillMs"), "native prefillMs")
            finite_timing(output.get("totalMs"), "native totalMs", True)
            if workload == "fixed":
                finite_timing(output.get("decodeMs"), "native decodeMs", True)
            receipt["cases"].append(dict(label, request=request, result=output))
        receipt.setdefault("commands", []).append({"argv": argv, "wall_seconds": time.monotonic() - start})
        save_json(work / "native.json", receipt)
    receipt["status"] = "complete"
    measured = [c for c in receipt["cases"] if not c["warmup"]]
    fixed_cases = [c["result"] for c in measured if c["workload"] == "fixed"]
    summaries = {}
    for workload in ("fixed", "pet"):
        for scenario in ("short_idle", "long_react"):
            runs = [c["result"] for c in measured if c["workload"] == workload and c["scenario"] == scenario]
            summaries[f"{scenario}/{workload}"] = {key: statistics.median(r[key] for r in runs)
                                                   for key in ("prefillMs", "totalMs", "generatedTokens", "decodeMs", "decodeTokens")}
    receipt["summary"] = summaries
    receipt["fixed_decode_tps"] = 1000 * sum(r["decodeTokens"] for r in fixed_cases) / sum(r["decodeMs"] for r in fixed_cases)
    save_json(work / "native.json", receipt)
    return {"native_fixed_decode_tps": receipt["fixed_decode_tps"], "native_timings": summaries,
            "native_load_ms": {workload: load["load_ms"] for workload, load in receipt["loads"].items()},
            "native_load_scope": {workload: load.get("scope") for workload, load in receipt["loads"].items()}}


def historical_readiness(export, graph_hash, dtype, dataset_hash, policy):
    evidence, blockers = {}, []
    for kind in ("quantization", "cache"):
        path = export.get(kind)
        if not path or not Path(path).is_file():
            evidence[kind] = None
            blockers.append(f"historical {kind} evidence unavailable")
            continue
        report = load_json(path)
        evidence[kind] = {"path": str(path), "sha256": digest(path), "report": report, "historical": True}
        if kind == "cache":
            if (report.get("status") != "success" or report.get("dtype") != dtype
                    or report.get("artifact", {}).get("graph_sha256") != graph_hash
                    or not report.get("incremental_kv", {}).get("allclose")):
                blockers.append("historical numerical/KV-cache gate failed or graph provenance mismatched")
        else:
            fp, quant, relative = report.get("fp_loss"), report.get("quant_loss"), report.get("relative_increase")
            valid = all(not isinstance(v, bool) and isinstance(v, (int, float)) and math.isfinite(v) for v in (fp, quant, relative))
            if (not valid or fp <= 0 or quant < 0 or not math.isclose(relative, (quant - fp) / fp, abs_tol=1e-9)
                    or report.get("status") != "complete" or report.get("dtype") != dtype
                    or report.get("dataset_sha256") != dataset_hash or report.get("graph_sha256") != graph_hash
                    or relative > policy["maximum_quantization_loss_increase"]):
                blockers.append("historical quantization loss/provenance gate failed")
    return evidence, blockers


def benchmark_markdown(report):
    def number(value, decimals=2):
        return "—" if value is None else f"{value:.{decimals}f}"
    rows = sorted(report["rows"], key=lambda r: (r["rank"] is None, r["rank"] or 0, r["id"], r["dtype"]))
    lines = ["# Descriptive model benchmark", "", report["scope"], "", report["quality_formula"], "",
             "Quality rank: descending capability index, then pooled browser fixed decode tokens/s, then graph bytes. Speed rank: throughput, quality, size. Failed/missing rows unranked. No promotion implied.", "",
             "Fixed speed uses 32 greedy tokens ignoring EOS/wire stops, excludes first-token prefill, and pools both short/long repeats. Pet browser latency includes real stop/filter/retry/lookup behavior; native pet latency is raw greedy generation only.", "",
             "| Quality rank | Speed rank | Model / dtype | Parameters (M) | Quality /100 | Browser fixed tok/s | Native fixed tok/s | Browser short / long sampled ms | Native short / long raw ms | Graph MB | Status |",
             "|---:|---:|---|---:|---:|---:|---:|---|---|---:|---|"]
    for row in rows:
        pet, native = row.get("browser_pet") or {}, row.get("native_timings") or {}
        b = [pet.get(f"{s}/sampled", {}).get("full_turn_ms") for s in ("short_idle", "long_react")]
        n = [native.get(f"{s}/pet", {}).get("totalMs") for s in ("short_idle", "long_react")]
        lines.append(f"| {row['rank'] or '—'} | {row['speed_rank'] or '—'} | {row['id']} / {row['dtype']} | {number(row['parameters'] / 1e6 if row['parameters'] is not None else None)} | {number(row['quality'])} | {number(row['browser_fixed_decode_tps'])} | {number(row['native_fixed_decode_tps'])} | {' / '.join(map(number, b))} | {' / '.join(map(number, n))} | {number(row['graph_bytes'] / 1e6 if row['graph_bytes'] is not None else None)} | {row['status']} |")
    lines += ["", "## Load and fixed-workload prefill (milliseconds)", "",
              "Browser load includes tokenizer/session loading; native load includes Brain::load and its built-in one-token warmup, excluding ORT init. Native prefill includes cache binding setup and browser prefill excludes tokenization; scopes are not identical.", "",
              "| Model / dtype | Browser load | Native fixed / pet load | Browser short / long prefill | Native short / long prefill | Qualification |",
              "|---|---:|---|---|---|---|"]
    for row in rows:
        browser, native = row.get("browser_fixed") or {}, row.get("native_timings") or {}
        loads = row.get("native_load_ms") or {}
        b = [browser.get(s, {}).get("prefill_ms") for s in ("short_idle", "long_react")]
        n = [native.get(f"{s}/fixed", {}).get("prefillMs") for s in ("short_idle", "long_react")]
        lines.append(f"| {row['id']} / {row['dtype']} | {number(row.get('browser_load_ms'))} | {number(loads.get('fixed'))} / {number(loads.get('pet'))} | {' / '.join(map(number, b))} | {' / '.join(map(number, n))} | {row['qualification']} |")
    lines += ["", "## Equal-weight capability percentages and observed counts", "",
              "| Model / dtype | " + " | ".join(CAPABILITY_METRICS) + " |", "|---|" + "---|" * len(CAPABILITY_METRICS)]
    for row in rows:
        categories = row.get("categories") or {}
        cells = []
        for name in CAPABILITY_METRICS:
            category = categories.get(name)
            cells.append("—" if category is None else number(category["percent"]) + "% (" + "; ".join(f"{metric}: {count['numerator']}/{count['denominator']}" for metric, count in category["metrics"].items()) + ")")
        lines.append(f"| {row['id']} / {row['dtype']} | " + " | ".join(cells) + " |")
    lines += ["", "## Safety, failures and readiness (not part of capability score)", "",
              "Safety cells are raw numerator/denominator, not safety percentages; silence cannot earn safety points. Historical cache/quantization receipts are not fresh qualification.", "",
              "| Model / dtype | Raw unsafe / incorrect-offer / invalidity counts | Failures | Readiness blockers |", "|---|---|---|---|"]
    for row in rows:
        safety = row.get("safety")
        counts = "—" if safety is None else "; ".join(f"{k}: {v['numerator']}/{v['denominator']}" for k, v in safety.items())
        failures = "; ".join(f"{f['phase']}: {f['error']}" for f in row["failures"]) or "none"
        blockers = "; ".join(row["readiness_blockers"])
        lines.append(f"| {row['id']} / {row['dtype']} | {counts} | {failures.replace('|', '/').replace(chr(10), ' ')} | {blockers.replace('|', '/')} |")
    return "\n".join(lines) + "\n"


def benchmark(args):
    from model_data import evaluation_contract
    if (args.threads not in (1, 2, 4, 8, 16) or min(args.samples, args.diaries, args.cell_samples, args.repeats) < 1
            or args.warmup < 0 or not 0 <= args.seed <= 4294967295):
        raise ValueError("positive suite/repetition counts, supported threads, nonnegative warmup and uint32 seed are required")
    if not args.runtime_prefix or "/" in args.runtime_prefix or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" for c in args.runtime_prefix):
        raise ValueError("runtime prefix must be one safe registry namespace")
    from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
    url = urlsplit(args.url)
    if url.scheme not in ("http", "https"):
        raise ValueError("benchmark URL must use HTTP(S)")
    root = Path(args.run).expanduser().resolve()
    manifest, plan = load_json(root / "experiment.json"), load_json(root / "plan.json")
    selected = set(args.models.split(",")) if args.models else {r["id"] for r in manifest["models"]}
    unknown = selected - {r["id"] for r in manifest["models"]}
    dtypes = list(dict.fromkeys(args.dtypes.split(",")))
    if unknown or not dtypes or any(d not in GRAPH_NAMES for d in dtypes):
        raise ValueError(f"unknown models or dtypes: {sorted(unknown)}, {dtypes}")
    for path in (args.export_python, args.native_probe, args.ort_library):
        if not Path(path).expanduser().is_file():
            raise ValueError(f"explicit benchmark prerequisite is missing: {path}")
    dataset_path = Path(manifest["dataset"]["path"]).expanduser()
    if digest(dataset_path) != manifest["dataset"]["sha256"]:
        raise ValueError("current training dataset differs from the experiment manifest")
    out = fresh_directory(args.out)
    contract = evaluation_contract(args.seed, args.samples, args.diaries, args.cell_samples)
    cpu = platform.processor()
    if Path("/proc/cpuinfo").is_file():
        cpu = next((line.split(":", 1)[1].strip() for line in Path("/proc/cpuinfo").read_text().splitlines() if line.startswith("model name")), cpu)
    report = {"schema_version": 2, "status": "running", "screening": manifest.get("screening", True),
              "created_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
              "scope": "Fresh finite-suite measurements of experiment checkpoints; descriptive rankings only. No inherited-v6 parity, behavioral equivalence, publication approval or promotable winner.",
              "quality_formula": "quality = 100 * mean(valid_lines, mean(offer_answers, late_answers, sleeping_offer_valid, medicine_answers), mean(memory_saved, memory_header, memory_recall), mean(sleep_idle_valid, sleep_lights_off_valid, sleep_message_valid, sleep_pat_valid, woken_valid), quiet_idle, diary_format_valid). All six categories have equal weight; each metric uses validated observed numerator/denominator. Missing/zero denominators are not scores.",
              "inputs": dict(vars(args), func=None), "evaluation_contract": contract,
              "source_run": {"path": str(root), "manifest_sha256": digest(root / "experiment.json"), "plan_sha256": digest(root / "plan.json"), "screening": manifest.get("screening"), "dataset": manifest["dataset"]},
              "hardware": {"cpu_name": cpu, "platform": platform.platform(), "machine": platform.machine(), "python": platform.python_version(), "threads": args.threads, "device": "cpu", "native_probe_sha256": digest(Path(args.native_probe).expanduser()), "ort_library_sha256": digest(Path(args.ort_library).expanduser())},
              "source_sha256": {str(p.relative_to(ROOT)): digest(p) for p in (ROOT / "train/experiments.py", ROOT / "train/eval_pet.py", ROOT / "train/model_data.py", ROOT / "test/experiment_bench.py", ROOT / "test/bench.html", ROOT / "server/src/bin/brain-probe.rs", ROOT / "server/src/brain.rs")},
              "rows": []}
    reference_behavior, behavior_reports = None, {}
    for model in manifest["models"]:
        if model["id"] not in selected:
            continue
        model_id = model["id"]
        if not model_id or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" for c in model_id):
            raise ValueError("unsafe model ID in manifest")
        for dtype in dtypes:
            work = out / f"{model_id}-{dtype}"
            work.mkdir()
            row = {"id": model_id, "dtype": dtype, "role": model.get("role"), "status": "running",
                   "qualification": "Not qualified", "quality": None, "categories": None,
                   "safety": None, "parameters": None, "graph_bytes": None, "graph_sha256": None, "browser_fixed_decode_tps": None,
                   "native_fixed_decode_tps": None, "browser_fixed": None, "browser_pet": None, "native_timings": None,
                   "browser_load_ms": None, "native_load_ms": None,
                   "failures": [], "commands": [], "reports": {}, "readiness_blockers": ["matching transcript review required", "inherited-v6 equivalence not established by this benchmark"]}
            if manifest.get("screening", True):
                row["readiness_blockers"].append("screening checkpoint; no full training acceptance")
            report["rows"].append(row)
            save_json(out / "benchmark.json", report)
            export_spec = model.get("exports", {}).get(dtype)
            try:
                if not export_spec:
                    raise ValueError("selected export is missing")
                export = Path(export_spec["path"]).expanduser().resolve()
                graph = export / "onnx" / GRAPH_NAMES[dtype]
                graph_hash = digest(graph)
                row.update(graph_sha256=graph_hash, graph_bytes=graph.stat().st_size, export=str(export))
                training_path = Path(model["artifacts"]["training"])
                checkpoint = training_path.parent
                export_receipt = load_json(export / "export.json")
                training_receipt = load_json(training_path)
                parameters = training_receipt.get("model", {}).get("unique_parameters")
                if type(parameters) is not int or parameters < 1:
                    raise ValueError("training receipt lacks a valid unique parameter count")
                row["parameters"] = parameters
                dataset_hash = manifest["dataset"]["sha256"]
                if training_receipt.get("data", {}).get("sha256") != dataset_hash:
                    raise ValueError("checkpoint training dataset differs from source manifest")
                if digest(export / "training.json") != digest(training_path):
                    raise ValueError("export training receipt differs from checkpoint")
                for name, artifact in export_receipt["checkpoint"]["files"].items():
                    if digest(checkpoint / name) != artifact["sha256"]:
                        raise ValueError(f"current checkpoint file differs from export receipt: {name}")
                quantization = load_json(export / "quantization.json")
                identity = quantization.get("artifacts", {}).get(graph.name, {})
                if identity.get("sha256") != graph_hash or identity.get("bytes") != row["graph_bytes"]:
                    raise ValueError("current graph differs from export quantization receipt")
                row["provenance"] = {"training_path": str(training_path), "training_sha256": digest(training_path), "training": training_receipt,
                                     "checkpoint": export_receipt["checkpoint"], "export_receipt": export_receipt, "quantization_receipt_sha256": digest(export / "quantization.json")}
                row["historical_evidence"], blockers = historical_readiness(export_spec, graph_hash, dtype, dataset_hash, plan["quality"])
                row["readiness_blockers"].extend(blockers)
            except Exception as e:
                row["failures"].append({"phase": "artifact", "error": str(e)})
                row["status"] = "failed"
                save_json(out / "benchmark.json", report)
                continue
            print(f"BENCHMARK {model_id}/{dtype}: fresh behavior", flush=True)
            behavior_path = work / "behavior.json"
            row["reports"]["behavior"] = str(behavior_path)
            try:
                row["commands"].append(command(evaluation_command(args, checkpoint, behavior_path, args.samples, args.diaries, args.cell_samples, args.seed, export, dtype), work / "behavior.log"))
                behavior = load_json(behavior_path)
                validate_benchmark_behavior(behavior, graph_hash, dtype, dataset_hash, contract, reference_behavior)
                row.update(capability_index(behavior))
                for metric in SAFETY_METRICS:
                    require_rate(behavior, metric)
                row["safety"] = {name: behavior["rate_counts"][name] for name in SAFETY_METRICS}
                row["behavior_hardware"] = behavior["hardware"]
                row["behavior_provenance"] = {key: behavior[key] for key in ("schema_version", "scenario_sha256", "scenario_suite_sha256", "scenario_counts", "experiment_contract", "dataset_sha256", "config", "onnx_artifact")}
                behavior_reports[(model_id, dtype)] = behavior
                if reference_behavior is None:
                    reference_behavior = behavior
            except Exception as e:
                row["failures"].append({"phase": "behavior", "error": str(e)})
                row.update(quality=None, categories=None, safety=None)
            print(f"BENCHMARK {model_id}/{dtype}: browser pet+fixed", flush=True)
            browser_path = work / "browser.json"
            row["reports"]["browser"] = str(browser_path)
            fixed = None
            try:
                params = dict(parse_qsl(url.query))
                params.update(model=f"{args.runtime_prefix}/{model_id}-{dtype}", prompts="both", mode="both")
                served_url = urlunsplit(url._replace(query=urlencode(params)))
                row["commands"].append(command([args.export_python, ROOT / "test/experiment_bench.py", "--url", served_url,
                    "--export", export, "--out", browser_path, "--dtype", dtype, "--device", "wasm", "--threads", args.threads,
                    "--qdq", 0, "--prepack", 0, "--workload", "both", "--repeats", args.repeats, "--warmup", args.warmup,
                    "--seed", args.seed, "--timeout-ms", 3600000], work / "browser.log"))
                metrics, fixed = browser_benchmark_metrics(load_json(browser_path), graph_hash, row["graph_bytes"], args)
                row.update(metrics)
            except Exception as e:
                row["failures"].append({"phase": "browser", "error": str(e)})
            if fixed is not None:
                print(f"BENCHMARK {model_id}/{dtype}: native pet+fixed", flush=True)
                row["reports"]["native"] = str(work / "native.json")
                try:
                    row.update(native_benchmark(args, export, dtype, fixed, work))
                except Exception as e:
                    row["failures"].append({"phase": "native", "error": str(e)})
                    native_path = work / "native.json"
                    native_report = load_json(native_path) if native_path.is_file() else {}
                    native_report.update(status="failed", error=str(e))
                    save_json(native_path, native_report)
            else:
                row["failures"].append({"phase": "native", "error": "browser fitted prompt receipt unavailable; native workload not substituted"})
            if digest(graph) != graph_hash:
                row["failures"].append({"phase": "artifact", "error": "graph changed during measurement"})
            row["status"] = "failed" if row["failures"] else "complete"
            print(f"BENCHMARK {model_id}/{dtype}: {row['status']}", flush=True)
            save_json(out / "benchmark.json", report)
    controls = [r for r in report["rows"] if r["role"] == "reference" and (r["id"], r["dtype"]) in behavior_reports]
    control = next((r for r in controls if r["dtype"] == "fp32"), controls[0] if controls else None)
    for row in report["rows"]:
        behavior = behavior_reports.get((row["id"], row["dtype"]))
        if behavior is not None and control is not None:
            gate = quality_gate(behavior_reports[(control["id"], control["dtype"])], behavior, plan["quality"])
            row["behavior_gate"] = dict(gate, reference=f"{control['id']}/{control['dtype']}")
            row["readiness_blockers"].extend(gate["reasons"])
        else:
            row["readiness_blockers"].append("fresh complete control/variant behavior comparison unavailable")
    rank_benchmark(report["rows"])
    report["status"] = "complete" if all(r["status"] == "complete" for r in report["rows"]) else "complete_with_failures"
    report["successful_variants"] = sum(r["status"] == "complete" for r in report["rows"])
    report["failed_variants"] = len(report["rows"]) - report["successful_variants"]
    save_json(out / "benchmark.json", report)
    (out / "benchmark.md").write_text(benchmark_markdown(report), encoding="utf-8")
    print(json.dumps({"status": report["status"], "report": str(out / "benchmark.json"), "table": str(out / "benchmark.md"), "variants": len(report["rows"])}))
    return 0 if report["failed_variants"] == 0 else 1


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)
    gen = sub.add_parser("generate")
    gen.add_argument("--out", required=True)
    gen.add_argument("--teacher-url", required=True)
    gen.add_argument("--teacher-model", required=True)
    gen.add_argument("--teacher-artifact")
    gen.add_argument("--chats", type=int, required=True)
    gen.add_argument("--diaries", type=int, required=True)
    gen.add_argument("--workers", type=int, default=2)
    gen.add_argument("--seed", type=int, default=0)
    gen.add_argument("--focus", choices=["general", "memory", "recall", "sleep", "care", "lookup"])
    gen.add_argument("--curriculum", action="store_true", help="teacher-authored structured turn curriculum")
    gen.add_argument("--stage", choices=["baby", "kid", "teen", "adult"])
    gen.add_argument("--python", default=sys.executable)
    gen.set_defaults(func=generate)
    for name in ("semantic-generate", "correction-data"):
        semantic = sub.add_parser(name, help="bounded manifest-isolated teacher data; requires separate grounded audit")
        semantic.add_argument("--manifest", required=True)
        semantic.add_argument("--out", required=True)
        semantic.add_argument("--teacher-url", required=True)
        semantic.add_argument("--teacher-model", required=True)
        semantic.add_argument("--teacher-artifact")
        semantic.add_argument("--seed", type=int, default=85710)
        semantic.add_argument("--per-cell", type=int, default=32)
        semantic.add_argument("--workers", type=int, default=1 if name == "correction-data" else 4,
                              help="bounded teacher-only concurrency; correction inference requires 1")
        semantic.add_argument("--offset", type=int, default=0)
        semantic.add_argument("--curriculum-chats", type=int, default=0)
        semantic.add_argument("--diaries", type=int, default=0)
        semantic.add_argument("--prior-generation", nargs="*", default=[])
        semantic.add_argument("--pilot-audit", help="passing source-linked pilot audit required before expansion/correction")
        if name == "correction-data":
            semantic.add_argument("--student-export", required=True)
            semantic.add_argument("--dtype", choices=("q4", "q8", "fp32"), default="fp32")
            semantic.add_argument("--threads", type=int, default=4)
        semantic.set_defaults(func=semantic_generate)
    auditor = sub.add_parser("corpus-audit", help="exclude rows without immutable producer truth and semantic grounding")
    auditor.add_argument("--data", required=True)
    auditor.add_argument("--requests", nargs="+", required=True)
    auditor.add_argument("--reviews", nargs="*", default=[])
    auditor.add_argument("--manifest", required=True)
    auditor.add_argument("--engine-binary", required=True)
    auditor.add_argument("--out", required=True)
    auditor.add_argument("--pilot", action="store_true")
    auditor.set_defaults(func=corpus_audit)
    freezer = sub.add_parser("freeze-corpus", help="freeze checked C1, original rows, provenance and unchanged seeded CE split")
    freezer.add_argument("--audits", nargs="+", required=True)
    freezer.add_argument("--generations", nargs="+", required=True)
    freezer.add_argument("--manifest", required=True)
    freezer.add_argument("--split-seed", type=int, default=0)
    freezer.add_argument("--out", required=True)
    freezer.set_defaults(func=freeze_corpus)
    rollout = sub.add_parser("rollout-data", help="sequence distill an actual exported pet brain on new synthetic scenes")
    rollout.add_argument("--teacher-export", required=True)
    rollout.add_argument("--teacher-id", required=True)
    rollout.add_argument("--dtype", choices=("q4", "q8", "fp32"), default="q4")
    rollout.add_argument("--out", required=True)
    rollout.add_argument("--chats", type=int, required=True)
    rollout.add_argument("--diaries", type=int, required=True)
    rollout.add_argument("--threads", type=int, default=4)
    rollout.add_argument("--seed", type=int, default=0)
    rollout.add_argument("--focus", choices=("general", "memory", "recall", "sleep", "care", "lookup"))
    rollout.set_defaults(func=rollout_data)
    mixer = sub.add_parser("merge")
    mixer.add_argument("--data", nargs="+", required=True)
    mixer.add_argument("--out", required=True)
    mixer.set_defaults(func=merge)
    runner = sub.add_parser("run")
    runner.add_argument("--plan", default=str(DEFAULT_PLAN))
    runner.add_argument("--data", required=True)
    runner.add_argument("--out", required=True)
    runner.add_argument("--models")
    runner.add_argument("--python", default=sys.executable)
    runner.add_argument("--export-python", default=sys.executable)
    runner.add_argument("--device", choices=["auto", "cpu", "cuda"], default="auto")
    runner.add_argument("--threads", type=int, default=4)
    runner.add_argument("--max-steps", type=int)
    runner.add_argument("--eval-every", type=int, default=200)
    runner.add_argument("--samples", type=int)
    runner.add_argument("--diaries", type=int)
    runner.add_argument("--cell-samples", type=int)
    runner.add_argument("--quant-rows", type=int, default=40)
    runner.add_argument("--dtypes", default="q4,q8,fp32")
    runner.add_argument("--pretrain-data")
    runner.add_argument("--pretrain-warm", action="store_true", help="also run the identical language-adaptation stage for pretrained controls/candidates")
    runner.add_argument("--pretrain-steps", type=int, default=2000)
    runner.add_argument("--pretrain-lr", type=float, default=0.0006)
    runner.set_defaults(func=run)
    staging = sub.add_parser("stage")
    staging.add_argument("--run", required=True)
    staging.add_argument("--out", required=True, help="new private directory under the preview models tree")
    staging.set_defaults(func=stage)
    paired = sub.add_parser("paired-inference", help="compare pinned published-v6 actual inference without a fictional training receipt")
    paired.add_argument("--reference", required=True)
    paired.add_argument("--candidate", required=True)
    paired.add_argument("--published-graph-sha256", required=True)
    paired.add_argument("--plan", default=str(DEFAULT_PLAN))
    paired.add_argument("--out", required=True)
    paired.set_defaults(func=paired_inference)
    comp = sub.add_parser("compare")
    comp.add_argument("--run", required=True)
    comp.add_argument("--baseline")
    comp.add_argument("--review")
    comp.add_argument("--runtime", help="JSON mapping model IDs and dtypes to actual browser report paths")
    comp.add_argument("--behavior", help="same mapping for separately evaluated ONNX session flags; original receipts remain untouched")
    comp.add_argument("--cache", help="same mapping for separately checked numerical/cache session flags")
    comp.set_defaults(func=compare)
    bench = sub.add_parser("benchmark", help="fresh descriptive quality/speed measurements; never promotes screening models")
    bench.add_argument("--run", required=True)
    bench.add_argument("--out", required=True)
    bench.add_argument("--url", required=True)
    bench.add_argument("--export-python", required=True)
    bench.add_argument("--native-probe", required=True)
    bench.add_argument("--ort-library", required=True)
    bench.add_argument("--threads", type=int, default=4)
    bench.add_argument("--samples", type=int, default=40)
    bench.add_argument("--diaries", type=int, default=8)
    bench.add_argument("--cell-samples", type=int, default=10)
    bench.add_argument("--seed", type=int, default=0)
    bench.add_argument("--repeats", type=int, default=5)
    bench.add_argument("--warmup", type=int, default=1)
    bench.add_argument("--dtypes", default="q4,q8,fp32")
    bench.add_argument("--models")
    bench.add_argument("--runtime-prefix", default="experiments")
    bench.set_defaults(func=benchmark)
    args = parser.parse_args()
    try:
        result = args.func(args)
        return result or 0
    except (ValueError, RuntimeError, OSError) as e:
        parser.exit(1, f"experiment error: {e}\n")


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