Something went wrong. Try again.
a tiny pet with an even tinier brain, living in your browser
Something went wrong. Try again.
121 kB · 1961 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962"""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 arenever committed. --python and --export-python select isolated tool environments;--device and --threads control compute. No teacher endpoint, deployment path orGPU host is hard-coded. A result is eligible only after complete comparablebehavior reports, acceptable quantization loss, actual runtime measurements andtranscript review. Smoke/screening runs are labelled, never publication evidence."""import argparseimport hashlibimport jsonimport mathimport osfrom pathlib import Pathimport platformimport statisticsimport subprocessimport sysimport time
ROOT = Path(__file__).resolve().parent.parentDEFAULT_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())