# Tensor Roll — Recursive CUDA-Q Model Quantizer # Copyright (C) 2026 SnapKitty Collective # SPDX-License-Identifier: AGPL-3.0-or-later """The `tensor-roll` command surface (spec section 11). All 14 subcommands operate on a working directory (default ~/.tensor-roll/default) and do real work: training runs, SVDs are computed, reports carry measured values. Anything that cannot be executed in this environment is reported as `unavailable` with a reason — never invented. """ from __future__ import annotations import argparse import os import shutil import sys import time import numpy as np from . import core as C from . import distill as D from . import model as M from . import qsim as Q from . import quant as QT from . import sandbox as SB from . import search as S from . import trq as T from .util import (AuditLog, ensure_dir, fmt_bytes, fmt_params, read_json, rss_mb, utc_now_iso, write_json) DEFAULT_WORKDIR = os.path.expanduser("~/.tensor-roll/default") FORMATS = ["fp32", "fp16", "bf16", "int8", "int4"] # --------------------------------------------------------------------------- # workdir helpers # --------------------------------------------------------------------------- def wd(args) -> str: return ensure_dir(args.workdir) def _save_params(path: str, params: dict) -> None: np.savez(path, **{k: np.asarray(v) for k, v in params.items()}) def _load_params(path: str) -> dict: z = np.load(path, allow_pickle=False) return {k: z[k] for k in z.files} def _need(path: str, what: str): if not os.path.exists(path): print(f"missing {what}: {path}\nrun the earlier pipeline stage first.", file=sys.stderr) sys.exit(2) return path def _teacher(wdir: str): cfg = M.Config.from_dict(read_json(os.path.join(wdir, "teacher_cfg.json"))) params = _load_params(_need(os.path.join(wdir, "teacher.npz"), "teacher")) return params, cfg def _student(wdir: str, variant: str = "fp32"): if variant == "fp32": cfg = M.Config.from_dict(read_json(os.path.join(wdir, "student_cfg.json"))) params = _load_params(_need(os.path.join(wdir, "student_final.npz"), "student")) return params, cfg # quantized variants live in .trq files path = _need(os.path.join(wdir, "trq", f"student_{variant}.trq"), f"{variant} artifact") header, tensors = T.read_trq(path) cfg = M.Config.from_dict(header["arch"]["config"]) params = {k: QT.dequantize_tensor(arr, meta).astype(np.float32) for k, (arr, meta) in tensors.items()} return params, cfg # --------------------------------------------------------------------------- # ingest # --------------------------------------------------------------------------- def cmd_ingest(args): wdir = wd(args) if args.scale == "30b": print("scale: glimmer-30b (meta-models/Muse-Glimmer-30B)") reqs = [ ("GPU with >=24GB VRAM (train) / >=20GB RAM (infer 4-bit)", _has_gpu(), "nvidia-smi not found"), ("disk >= 120GB free for weights + workspace", _free_gb() >= 120, f"only {_free_gb():.0f}GB free"), ("RAM >= 32GB for full-precision ingest", _ram_gb() >= 32, f"only {_ram_gb():.0f}GB RAM"), ] ok = True for name, have, why in reqs: print(f" [{'OK' if have else 'UNAVAILABLE'}] {name}" + ("" if have else f" — {why}")) ok = ok and have if not ok: print("ingest --scale 30b: unavailable in this environment " "(no GPU, 7GB RAM). See docs/ARCHITECTURE.md for the " "30B hardware path. Nothing was fabricated.") sys.exit(3) print("downloading meta-models/Muse-Glimmer-30B ...") sys.exit(3) # unreachable here; kept explicit cfg = M.Config() print(f"training tiny teacher {cfg} " f"({fmt_params(_pcount(cfg))} params, seed={args.seed})") t0 = time.perf_counter() params, history = M.train_teacher(cfg, steps=args.steps, batch=args.batch, lr=args.lr, seed=args.seed, log_every=max(1, args.steps // 5)) dt = time.perf_counter() - t0 _save_params(os.path.join(wdir, "teacher.npz"), params) write_json(os.path.join(wdir, "teacher_cfg.json"), cfg.to_dict()) write_json(os.path.join(wdir, "teacher_history.json"), history) ev = M.eval_metrics(params, cfg) write_json(os.path.join(wdir, "teacher_eval.json"), ev) prov = {"scale": "tiny", "task": "synthetic modular grammar", "steps": args.steps, "batch": args.batch, "lr": args.lr, "seed": args.seed, "elapsed_s": dt, "backend": C.CLASSICAL_BACKEND, "created": utc_now_iso()} write_json(os.path.join(wdir, "teacher_provenance.json"), prov) print(f"teacher: {fmt_params(_pcount(cfg))} params, " f"held-out loss={ev['loss']:.4f} acc={ev['accuracy']:.3f}, " f"{dt:.1f}s on {C.CLASSICAL_BACKEND}") def _pcount(cfg) -> int: return M.count_params(M.init_params(np.random.default_rng(0), cfg)) def _has_gpu() -> bool: return shutil.which("nvidia-smi") is not None def _free_gb() -> float: return shutil.disk_usage(os.path.expanduser("~")).free / 1e9 def _ram_gb() -> float: try: with open("/proc/meminfo") as f: for line in f: if line.startswith("MemTotal:"): return float(line.split()[1]) / 1e6 except OSError: return 0.0 return 0.0 # --------------------------------------------------------------------------- # inspect / map # --------------------------------------------------------------------------- def cmd_inspect(args): wdir = wd(args) if args.what == "trq": header, tensors = T.read_trq(_need(args.trq, ".trq file")) print(f"{args.trq}: {header['n_tensors']} tensors, " f"{fmt_bytes(os.path.getsize(args.trq))}, " f"sha256={T.read_trq(args.trq) and header and 'verified'}") total, ok = T.param_budget_ok(tensors) print(f"params: {fmt_params(total)} budget<=8.5B: {'OK' if ok else 'VIOLATION'}") for k, (arr, meta) in sorted(tensors.items()): print(f" {k:24s} {str(list(meta.get('shape', arr.shape))):28s} " f"{meta.get('fmt', '?'):6s}") return params, cfg = _teacher(wdir) if args.what == "teacher" else _student(wdir) print(f"{args.what}: {cfg}, {fmt_params(M.count_params(params))} params") for k in sorted(params): v = params[k] print(f" {k:16s} {str(list(v.shape)):22s} {str(v.dtype):10s} " f"||.||={float(np.linalg.norm(v.astype(np.float64).ravel())):.4f}") def cmd_map(args): wdir = wd(args) params, cfg = _teacher(wdir) rows = [] for k in sorted(params): m = C.tensor_metrics(params[k]) rows.append({"tensor": k, **{kk: m[kk] for kk in ("shape", "params", "norm", "rank", "entropy", "variance", "spectral")}}) write_json(os.path.join(wdir, "map.json"), rows) print(f"{'tensor':16s} {'shape':22s} {'params':>8s} {'rank':>4s} " f"{'entropy':>7s} {'spectral':>8s}") for r in rows: print(f"{r['tensor']:16s} {str(r['shape']):22s} {r['params']:8d} " f"{r['rank']:4d} {r['entropy']:7.3f} {r['spectral']:8.3f}") print(f"total: {fmt_params(sum(r['params'] for r in rows))} params") # --------------------------------------------------------------------------- # roll # --------------------------------------------------------------------------- def cmd_roll(args): wdir = wd(args) params, cfg = _teacher(wdir) t0 = time.perf_counter() # measured activation/gradient contributions (real fwd/bwd passes) stats = M.measure_tensor_stats(params, cfg) trees, leaves = S.run_roll_analysis(params, depth=args.depth, quantum_policy=args.quantum_policy, stats=stats) n_teacher = M.count_params(params) budget = n_teacher * args.budget_ratio plan = C.select_plan(leaves, budget) for leaf in leaves: leaf.decision = plan[leaf.name] dt = time.perf_counter() - t0 serial = {leaf.name: {"decision": leaf.decision["decision"], "cost_params": leaf.decision["cost_params"], "retained_energy": leaf.decision["retained_energy"], "reconstruction_error": leaf.decision["reconstruction_error"], "detail": leaf.decision["detail"], "metrics": leaf.metrics} for leaf in leaves} write_json(os.path.join(wdir, "roll_plan.json"), {"budget_ratio": args.budget_ratio, "budget_params": budget, "total_cost": plan["_total_cost_params"], "quantum_policy": args.quantum_policy, "elapsed_s": dt, "leaves": serial}) shown = leaves[:40] for i, leaf in enumerate(shown, 1): print(C.render_roll_line(leaf, i)) if len(leaves) > len(shown): print(f"... {len(leaves) - len(shown)} more leaves " f"({len(leaves)} total)") from collections import Counter hist = Counter(l.decision["decision"] for l in leaves) print(f"decisions: {dict(hist)}") print(f"budget: {fmt_params(budget)} planned cost: " f"{fmt_params(int(plan['_total_cost_params']))} " f"{'OK' if plan['_total_cost_params'] <= budget else 'OVER BUDGET'} " f"({dt:.1f}s)") # --------------------------------------------------------------------------- # qkernel # --------------------------------------------------------------------------- def cmd_qkernel(args): wdir = wd(args) params, cfg = _teacher(wdir) name = args.tensor or max((k for k in params if params[k].ndim == 2), key=lambda k: params[k].size) W = np.asarray(params[name], dtype=np.float64) # take the highest-energy top-left block that fits the qubit cap: # amplitude packs 2^n elems into n qubits; angle needs 1 qubit/elem. n = 8 if args.encoding == "amplitude" else 3 block = W[:n, :n].copy() nq = 6 if args.encoding == "amplitude" else 9 print(f"tensor_roll kernel family on {name}{tuple(block.shape)} " f"(top-left {n}x{n} block, {block.size} elems -> {nq} qubits, " f"{args.encoding} encoding)") t0 = time.perf_counter() payload, r0 = Q.tensor_roll_encode(block, encoding=args.encoding, seed=args.seed) print(_fmt_report("tensor_roll_encode", r0)) payload, r1 = Q.tensor_roll_rotate( payload, angles=np.linspace(0.2, 1.0, payload["n"]), seed=args.seed) print(_fmt_report("tensor_roll_rotate", r1)) payload, r2 = Q.tensor_roll_entangle(payload, seed=args.seed) print(_fmt_report("tensor_roll_entangle", r2)) payload, r3 = Q.tensor_roll_measure(payload, shots=args.shots, seed=args.seed) print(_fmt_report("tensor_roll_measure", r3)) rec, r4 = Q.tensor_roll_reconstruct(payload) r4["reconstruction_error"] = C.relative_error(block, rec) print(_fmt_report("tensor_roll_reconstruct", r4)) What, summary = Q.tensor_roll_recursive(block, depth=2, shots=args.shots, seed=args.seed, encoding=args.encoding) print(_fmt_report("tensor_roll_recursive[summary]", summary)) print(f"block rel recon error: {C.relative_error(block, rec):.4f} " f"(recursive: {C.relative_error(block, What):.4f}) " f"in {time.perf_counter()-t0:.2f}s — backend qsim-classical " f"(statevector simulator, NOT a QPU)") # Real process/IR boundary artifact: gate trace + measurement counts, # checksummed. This file is what crosses between the CUDA-Q process and # the Q# toolchain (or back); see docs/BOUNDARIES.md. No direct # CUDA-Q<->Q# linkage is claimed — the boundary is serialization. from . import boundary as BD bpath = os.path.join(wdir, "qkernel", "boundary.json") trace = BD.write_trace(bpath, "tensor_roll_recursive", summary["n_qubits"], args.shots, summary["counts"], depth=2, producer="qsim-classical") back = BD.read_trace(bpath) # verify the round trip immediately amps = BD.counts_to_amplitudes(back) print(f"boundary trace: {bpath} ({trace['n_gates']} gates, " f"{len(trace['counts'])} outcomes, sha256 verified, " f"recon[0]={amps[0]:.4f})") def _fmt_report(kernel: str, r: dict) -> str: ms = r.get("measurement_stats") if ms and r.get("shots"): if "levels" in ms: ms_s = (f"shots={r['shots']} levels={len(ms['levels'])} " f"active_q={ms['levels'][-1]['active_qubits']}") else: ms_s = (f"shots={r['shots']} outcomes={ms['n_outcomes']} " f"H={ms['dist_entropy_bits']:.2f}b") else: ms_s = "shots=—" re_ = r.get("reconstruction_error") return (f" {kernel:32s} backend={r['backend']} device={r['device']} " f"dims={r['tensor_dims']} prec={r['precision']} " f"t={r['exec_time_s']*1000:.2f}ms mem={fmt_bytes(r['memory_bytes'])} " f"depth={r['recursion_depth']} {ms_s} " f"recon_err={'—' if re_ is None else f'{re_:.2e}'}") # --------------------------------------------------------------------------- # compress # --------------------------------------------------------------------------- def cmd_doctor(args): from . import doctor as DOC report = DOC.doctor_report() print(DOC.render_doctor(report)) integ = report.get("integration", {}) if integ: print("Integration:") for k, v in integ.items(): print(f" {k}: {v}") if getattr(args, "json", False): path = os.path.join(wd(args), "doctor_report.json") write_json(path, report) print(f"wrote {path}") def cmd_compress(args): if getattr(args, "out_of_core", False): return cmd_compress_ooc(args) wdir = wd(args) t_params, t_cfg = _teacher(wdir) n_teacher = M.count_params(t_params) budget = n_teacher * args.budget_ratio print(f"teacher {fmt_params(n_teacher)} params, " f"budget ratio {args.budget_ratio} -> {fmt_params(budget)}") t0 = time.perf_counter() s_cfg, scored = S.search_student_config(t_params, t_cfg, budget) rng = np.random.default_rng(args.seed) s_params, provenance, energies = S.init_student_from_teacher( t_params, t_cfg, s_cfg, rng) heads = S.head_consolidation_report(t_params, t_cfg, s_cfg) n_student = M.count_params(s_params) dt = time.perf_counter() - t0 ok = n_student <= budget and n_student <= S.ABS_BUDGET_CAP print("architecture search (measured projected energy):") for s in scored[:6]: c = s["cfg"] print(f" d={c['d']} H={c['heads']} L={c['layers']} dff={c['dff']}: " f"est_energy={s['est_energy']:.4f} params={fmt_params(s['params'])}") print(f"chosen: {s_cfg} -> {fmt_params(n_student)} params " f"({100*n_student/n_teacher:.1f}% of teacher)") print(f"budget check: {fmt_params(n_student)} <= {fmt_params(budget)} " f"and <= 8.5B absolute: {'OK' if ok else 'VIOLATION'}") if not ok: print("FATAL: budget violated — refusing to emit student", file=sys.stderr) sys.exit(4) _save_params(os.path.join(wdir, "student_init.npz"), s_params) write_json(os.path.join(wdir, "student_cfg.json"), s_cfg.to_dict()) write_json(os.path.join(wdir, "provenance.json"), provenance) write_json(os.path.join(wdir, "head_report.json"), heads) write_json(os.path.join(wdir, "arch_search.json"), {"budget": budget, "elapsed_s": dt, "scored": scored, "init_energies": energies}) print(f"head consolidation: {heads}") print(f"compress done in {dt:.1f}s") def cmd_compress_ooc(args): """Streaming out-of-core compression (F1/F2). SCAN reads only weight metadata from --teacher (safetensors headers or a streaming npz repack — never the full model in RAM), PLAN allocates the global budget nonuniformly with measured sensitivity, EXECUTE streams windowed truncated-SVD factors to student shards with per-tensor checkpointing. --resume continues after a crash. """ from . import ooc as OOC from . import planner as P wdir = wd(args) teacher_dir = args.teacher or wdir if not os.path.isdir(teacher_dir): print(f"--teacher: not a directory: {teacher_dir}", file=sys.stderr) sys.exit(2) print(f"out-of-core compress: SCAN {teacher_dir} (metadata only)") t0 = time.perf_counter() source = OOC.prepare_teacher_source( teacher_dir, os.path.join(wdir, "ooc_cache")) print(f" source: {len(source.names())} tensors, " f"{fmt_params(source.total_params)} params, " f"{fmt_bytes(sum(s.nbytes for s in source.specs))}") rows = P.scan_rows(source.specs) # measured sensitivity from small sampled windows (<=4096 elems) reader = source.reader() try: def sample_fn(name): try: mt = reader.read(name) except Exception: return None flat = np.asarray(mt.view).reshape(-1) step = max(1, flat.size // 4096) s = flat[::step][:4096] return s.reshape(64, 64) if s.size == 4096 else s sens = P.calibrate_sensitivity(rows, sample_fn=sample_fn) finally: reader.close() plan = P.plan(rows, target=args.target_params, sensitivity=sens) dt_plan = time.perf_counter() - t0 print(f" PLAN: target {fmt_params(plan['total_target_params'])} " f"(ratio {plan['achieved_ratio']:.4f}), " f"cap {fmt_params(plan['cap'])}: " f"{'OK' if plan['within_cap'] else 'VIOLATION'} " f"[{dt_plan:.1f}s]") if not plan["within_cap"]: print("FATAL: plan exceeds hard budget — refusing", file=sys.stderr) sys.exit(4) out_dir = os.path.join(wdir, "ooc_student") res = OOC.stream_compress(source, plan, args.memory_budget, out_dir, resume=args.resume) print(f" EXECUTE: {res['processed_tensors']} tensors, " f"{fmt_params(res['index_total_params'])} student params " f"(ratio {res['compression_ratio']:.4f}), " f"peak workspace {fmt_bytes(res['workspace']['peak_bytes'])} " f"/ {fmt_bytes(res['workspace']['ceiling_bytes'])}, " f"{res['elapsed_s']:.1f}s") print(f" budget check: {'OK' if res['budget_ok'] else 'VIOLATION'}") energies = [s["retained_energy"] for s in res["tensor_stats"].values() if "retained_energy" in s] mean_e = float(np.mean(energies)) if energies else float("nan") print(f" mean measured retained energy: {mean_e:.4f}") write_json(os.path.join(wdir, "ooc_plan.json"), plan) write_json(os.path.join(wdir, "ooc_compress_summary.json"), {k: v for k, v in res.items() if k != "tensor_stats"}) print(f" report: {res['report']}") # --------------------------------------------------------------------------- # train / finetune # --------------------------------------------------------------------------- STAGES = [ (0, "architecture-init", {}), (1, "tensor-reconstruction", {"reconstruction": 3.0, "task": 0.2, "logits": 0.2}), (2, "hidden-state-matching", {"hidden": 2.0, "attention": 2.0, "task": 0.3, "logits": 0.5}), (3, "logit-distillation", {"logits": 3.0, "task": 0.5, "hidden": 0.3, "attention": 0.3}), (4, "language-model", {"task": 2.0, "logits": 0.5}), (5, "instruction-tuning", {"task": 1.0}), # fresh-seed task data (6, "quantization-aware", {"task": 1.0, "logits": 0.5}), # sim-int8 noise (7, "regression-evaluation", {}), ] def _checkpoint(wdir, stage, params, opt, tag): d = ensure_dir(os.path.join(wdir, "checkpoints")) cp = dict(params) if opt is not None: for k in params: cp[f"__m__{k}"] = opt.m[k] cp[f"__v__{k}"] = opt.v[k] cp["__t__"] = np.array(opt.t) _save_params(os.path.join(d, f"stage{stage}_{tag}.npz"), cp) def _restore_opt(params, path): z = np.load(path, allow_pickle=False) opt = M.Adam(params, lr=3e-3) if "__t__" in z.files: for k in params: opt.m[k] = z[f"__m__{k}"] opt.v[k] = z[f"__v__{k}"] opt.t = int(z["__t__"]) return opt def cmd_train(args): wdir = wd(args) t_params, t_cfg = _teacher(wdir) s_cfg = M.Config.from_dict(read_json(os.path.join(wdir, "student_cfg.json"))) s_params = _load_params(os.path.join(wdir, "student_init.npz")) rng = np.random.default_rng(args.seed) # frozen projections for hidden/embedding matching proj_h = rng.standard_normal((s_cfg.d, t_cfg.d)).astype(np.float32) proj_e = rng.standard_normal((s_cfg.d, t_cfg.d)).astype(np.float32) recon_targets = {k: v.astype(np.float64) for k, v in s_params.items()} plan = read_json(os.path.join(wdir, "roll_plan.json"))["leaves"] \ if os.path.exists(os.path.join(wdir, "roll_plan.json")) else {} plan_e = {k: v["retained_energy"] for k, v in plan.items() if isinstance(v, dict) and "retained_energy" in v} achiev = read_json(os.path.join(wdir, "arch_search.json"))["init_energies"] opt = M.Adam(s_params, lr=args.lr) history = [] _checkpoint(wdir, 0, s_params, opt, "init") stages = STAGES if args.stage == "all" else [ s for s in STAGES if str(s[0]) == args.stage] for num, name, lam in stages: if num == 0: # architecture-init: checkpoint the searched init, no training _checkpoint(wdir, 0, s_params, opt, "init") history.append({"stage": 0, "name": name, "note": "init checkpoint only (no training)"}) print(f"stage 0 {name}: init checkpointed " f"({fmt_params(M.count_params(s_params))} params)") continue if num == 7: ev = M.eval_metrics(s_params, s_cfg) t_ev = M.eval_metrics(t_params, t_cfg) print(f"stage 7 regression-eval: student loss={ev['loss']:.4f} " f"acc={ev['accuracy']:.3f} | teacher loss={t_ev['loss']:.4f} " f"acc={t_ev['accuracy']:.3f}") history.append({"stage": 7, "name": name, "eval": ev, "teacher_eval": t_ev}) continue steps = args.steps if num not in (1,) else max(20, args.steps // 2) t0 = time.perf_counter() for step in range(1, steps + 1): ids = M.gen_data(rng, args.batch, s_cfg) if num == 6: # quantization-aware: straight-through sim-int8 noise noisy = {k: QT.dequantize_int8( *QT.quantize_int8(v)).astype(np.float32) if v.ndim == 2 and v.size > 64 else v for k, v in s_params.items()} fwd_params = noisy else: fwd_params = s_params # forward with fwd_params, grads applied to s_params (STE for QAT) total, terms, grads = D.distill_step( fwd_params, s_cfg, t_params, t_cfg, ids, lambdas=lam, proj_h=proj_h, proj_e=proj_e, recon_targets=recon_targets, roll_plan_energies=plan_e or None, achieved_energies=achiev or None) # grads keys match s_params; apply to the clean params opt.step(s_params, {k: grads[k] for k in s_params}) if step % max(1, steps // 4) == 0 or step == steps: history.append({"stage": num, "name": name, "step": step, **{k: round(float(v), 5) for k, v in terms.items()}}) _checkpoint(wdir, num, s_params, opt, name) print(f"stage {num} {name}: {steps} steps, " f"total_loss={history[-1]['total']:.4f} " f"({time.perf_counter()-t0:.1f}s)") _save_params(os.path.join(wdir, "student_distilled.npz"), s_params) write_json(os.path.join(wdir, "distill_history.json"), history) print("distillation complete; checkpoints in checkpoints/") def cmd_finetune(args): wdir = wd(args) s_cfg = M.Config.from_dict(read_json(os.path.join(wdir, "student_cfg.json"))) s_params = _load_params(_need(os.path.join(wdir, "student_distilled.npz"), "distilled student (run train first)")) rng = np.random.default_rng(args.seed) opt = _restore_opt(s_params, os.path.join( wdir, "checkpoints", sorted(os.listdir(os.path.join(wdir, "checkpoints")))[-1])) t0 = time.perf_counter() for step in range(1, args.steps + 1): ids = M.gen_data(rng, args.batch, s_cfg) logits, cache = M.forward(s_params, s_cfg, ids[:, :-1]) loss, dl = M.ce_loss(logits, ids[:, 1:]) grads = M.backward(s_params, s_cfg, cache, dl) opt.step(s_params, grads) dt = time.perf_counter() - t0 _save_params(os.path.join(wdir, "student_final.npz"), s_params) ev = M.eval_metrics(s_params, s_cfg) print(f"finetune: {args.steps} steps task-CE in {dt:.1f}s -> " f"loss={ev['loss']:.4f} acc={ev['accuracy']:.3f}") write_json(os.path.join(wdir, "finetune.json"), {"steps": args.steps, "elapsed_s": dt, "eval": ev}) # --------------------------------------------------------------------------- # quantize # --------------------------------------------------------------------------- def cmd_quantize(args): wdir = wd(args) s_cfg = M.Config.from_dict(read_json(os.path.join(wdir, "student_cfg.json"))) s_params = _load_params(_need(os.path.join(wdir, "student_final.npz"), "final student (run finetune first)")) tdir = ensure_dir(os.path.join(wdir, "trq")) prov = read_json(os.path.join(wdir, "teacher_provenance.json")) roll_meta = {} if os.path.exists(os.path.join(wdir, "roll_plan.json")): rp = read_json(os.path.join(wdir, "roll_plan.json")) for k, v in rp["leaves"].items(): if isinstance(v, dict): roll_meta[k] = {"decision": v.get("decision"), "retained_energy": v.get("retained_energy")} manifest = [] for fmt in FORMATS: tensors = {} max_recon = 0.0 for k, v in s_params.items(): q, qmeta = QT.quantize_tensor(v, fmt) back = QT.dequantize_tensor( q, {"fmt": fmt, **qmeta} if fmt != "int4" else qmeta) max_recon = max(max_recon, C.relative_error( v.astype(np.float64), back.astype(np.float64))) tensors[k] = (np.ascontiguousarray(q), fmt, qmeta) path = os.path.join(tdir, f"student_{fmt}.trq") info = T.write_trq(path, arch={"name": "tensor-roll-student", "config": s_cfg.to_dict(), "format": fmt}, tensors=tensors, provenance=prov, roll_meta=roll_meta) # verify round-trip + checksums immediately _h, back_tensors = T.read_trq(path) total, ok = T.param_budget_ok(back_tensors) manifest.append({"format": fmt, "path": path, "bytes": info["bytes"], "sha256": info["sha256"], "params": total, "budget_ok": ok, "max_weight_recon_err": max_recon}) print(f"student_{fmt}.trq: {fmt_bytes(info['bytes'])} " f"params={fmt_params(total)} budget_ok={ok} " f"max_recon_err={max_recon:.2e} sha256={info['sha256'][:16]}… " f"[round-trip verified]") write_json(os.path.join(wdir, "quantize_manifest.json"), manifest) # --------------------------------------------------------------------------- # evaluate # --------------------------------------------------------------------------- def cmd_evaluate(args): wdir = wd(args) t_params, t_cfg = _teacher(wdir) results = {} for fmt in FORMATS: s_params, s_cfg = _student(wdir, variant=fmt) ev = M.eval_metrics(s_params, s_cfg, seed=4242) # teacher divergence on shared batches rng = np.random.default_rng(4242) kls = [] for _ in range(4): ids = M.gen_data(rng, 128, s_cfg) tl, _ = M.forward(t_params, t_cfg, ids[:, :-1]) sl, _ = M.forward(s_params, s_cfg, ids[:, :-1]) kls.append(M.kl_divergence_mean(tl, sl)) results[fmt] = {"held_out": ev, "teacher_kl_div": float(np.mean(kls))} print(f"{fmt:6s} loss={ev['loss']:.4f} acc={ev['accuracy']:.3f} " f"teacher_KL={results[fmt]['teacher_kl_div']:.4f}") s_params, s_cfg = _student(wdir, "fp32") rec = D.recursive_metrics(t_params, s_params) results["recursive_distillation"] = rec write_json(os.path.join(wdir, "eval.json"), results) print("recursive distillation metrics (sample):") for lvl in ("model", "layer_group_0", "subblock:L0.Wq[0]"): if lvl in rec: m = rec[lvl] print(f" {lvl:28s} rel_err={m['rel_err']:.3f} " f"cos={m['cosine_sim']:.4f} KL={m['kl_div']:.4f} " f"E_ret={m['retained_energy']:.4f}") # --------------------------------------------------------------------------- # benchmark # --------------------------------------------------------------------------- def cmd_benchmark(args): wdir = wd(args) rows = [] try: t_params, t_cfg = _teacher(wdir) rows.append(_bench_row("glimmer-tiny-teacher (fp32)", t_params, t_cfg, wdir)) except SystemExit: pass for fmt in FORMATS: try: s_params, s_cfg = _student(wdir, variant=fmt) label = f"tensor-roll-student ({fmt}/trq)" rows.append(_bench_row(label, s_params, s_cfg, wdir, trq_path=os.path.join( wdir, "trq", f"student_{fmt}.trq"))) except SystemExit: rows.append(_unavailable_row(f"tensor-roll-student ({fmt})", "artifact not built")) # 30B / CUDA rows: honestly unavailable here rows.append(_unavailable_row("glimmer-30b (fp16)", "no GPU / 7GB RAM")) rows.append(_unavailable_row("tensor-roll-8b (cuda)", "no CUDA device")) rows.append(_unavailable_row("tensor-roll-8b (cuda-q qpu)", "no QPU / no CUDA-Q toolchain")) hdr = (f"{'artifact':36s} {'params':>8s} {'size':>9s} {'load_s':>7s} " f"{'tok/s':>7s} {'ppl':>7s} {'acc':>6s} {'tch_KL':>7s} {'backend':>12s}") print(hdr) print("-" * len(hdr)) for r in rows: print(f"{r['artifact']:36s} {r['params']:>8s} {r['size']:>9s} " f"{r['load_s']:>7s} {r['tok_s']:>7s} {r['ppl']:>7s} " f"{r['acc']:>6s} {r['kl']:>7s} {r['backend']:>12s}") write_json(os.path.join(wdir, "benchmark.json"), rows) print("\n'unavailable' = not executed here; no number was invented.") def _bench_row(label, params, cfg, wdir, trq_path=None): n = M.count_params(params) size = os.path.getsize(trq_path) if trq_path else sum( v.nbytes for v in params.values()) t0 = time.perf_counter() if trq_path: T.read_trq(trq_path) load_s = time.perf_counter() - t0 rng = np.random.default_rng(7) ids = M.gen_data(rng, 64, cfg) t0 = time.perf_counter() logits, _ = M.forward(params, cfg, ids[:, :-1]) fwd_s = time.perf_counter() - t0 loss, _ = M.ce_loss(logits, ids[:, 1:]) toks, tps, _ = M.generate(params, cfg, ids[0, :4].tolist(), 16) acc = M.eval_metrics(params, cfg, batches=2)["accuracy"] return {"artifact": label, "params": fmt_params(n), "size": fmt_bytes(size), "load_s": f"{load_s:.3f}", "tok_s": f"{tps:.1f}", "ppl": f"{np.exp(loss):.2f}", "acc": f"{acc:.3f}", "kl": "see-eval", "backend": C.CLASSICAL_BACKEND, "_note": f"forward {fwd_s*1000:.1f}ms/64tok"} def _unavailable_row(label, reason): u = "unavailable" return {"artifact": label, "params": u, "size": u, "load_s": u, "tok_s": u, "ppl": u, "acc": u, "kl": u, "backend": u, "_note": reason} # --------------------------------------------------------------------------- # chat # --------------------------------------------------------------------------- def _detokenize(ids, cfg): # demo tokenizer: ids <-> printable chars (documented as demo-only) return "".join(chr(33 + (i % 94)) for i in ids) def _tokenize(text, cfg): return [ord(c) % cfg.vocab for c in text[:cfg.seq]] or [0] def cmd_chat(args): wdir = wd(args) s_params, s_cfg = _student(wdir, variant=args.variant) sup = SB.SandboxSupervisor(os.path.join(wdir, "sandbox")) print(f"tensor-roll chat [{args.variant}] — type text, `!cmd` runs sandboxed, " f"`quit` exits") if args.compare: t_params, t_cfg = _teacher(wdir) while True: try: line = input("tensor> ").strip() except EOFError: break if line in ("quit", "exit"): break if line.startswith("!"): rec = sup.run(line[1:]) print(f"[sandbox {'ALLOW' if rec['allowed'] else 'DENY'}] " f"{rec.get('reason', '')}") if rec["allowed"]: print(rec.get("stdout", ""), end="") continue if not line: continue prompt = _tokenize(line, s_cfg) toks, tps, dt = M.generate(s_params, s_cfg, prompt, args.max_tokens) print(f"[student] {_detokenize(toks, s_cfg)} " f"({tps:.1f} tok/s, {C.CLASSICAL_BACKEND})") if args.compare: tt, ttps, _ = M.generate(t_params, t_cfg, prompt, args.max_tokens) print(f"[teacher] {_detokenize(tt, t_cfg)} ({ttps:.1f} tok/s)") # --------------------------------------------------------------------------- # sandbox # --------------------------------------------------------------------------- def cmd_sandbox(args): wdir = wd(args) sup = SB.SandboxSupervisor(os.path.join(wdir, "sandbox")) print(f"jail: {sup.jail} netns_available={sup.netns_available}") if args.demo: for label, rec in sup.demo(): mark = "ALLOW" if rec["allowed"] else "DENY" print(f"[{mark}] {label}: {rec['reason']}") if rec["allowed"] and rec.get("stdout"): print(f" out: {rec['stdout'].strip()}") print(f"audit log: {sup.audit.path}") return rec = sup.run(args.sandbox_cmd) print(f"[{'ALLOW' if rec['allowed'] else 'DENY'}] {rec['reason']}") if rec["allowed"]: print(rec.get("stdout", ""), end="") if rec.get("stderr"): print(rec["stderr"], end="", file=sys.stderr) # --------------------------------------------------------------------------- # demo # --------------------------------------------------------------------------- def cmd_demo(args): wdir = wd(args) F = args.fast steps_t = 250 if F else 600 steps_d = 40 if F else 120 steps_f = 40 if F else 150 print("=" * 64) print("Tensor Roll") print("Recursive CUDA-Q Model Runtime") print("GLIMMER 30B -> TENSOR ROLL -> CUDA -> CUDA-Q -> Q# -> 8B NANO MODEL") print("-" * 64) print("TARGET (not executed here): Glimmer 30B teacher -> ~8B student,") print(" hard budget P <= 8.5B. Requires: 30B-param HW (see docs/).") print("THIS RUN (executed now): tiny from-scratch validation teacher") print(" trained by the same code path at small scale; every number") print(" below was measured on this machine (no GPU: cpu-numpy).") print(f"Quantum CUDA-Q kernels + Q# ops ship in-repo; here the") print(f" qsim-classical statevector simulator executes") print(f"Sandbox ACTIVE") print("=" * 64) # 1. ingest print("\n[1/9] ingest — training tiny teacher (same code path as 30B)") a = argparse.Namespace(workdir=args.workdir, scale="tiny", steps=steps_t, batch=128, lr=3e-3, seed=11) cmd_ingest(a) t_params, t_cfg = _teacher(wdir) # 2. roll print("\n[2/9] roll — TensorRoll operator over teacher tensors") a = argparse.Namespace(workdir=args.workdir, depth=2, quantum_policy="explore", budget_ratio=S.BUDGET_RATIO) cmd_roll(a) # 3. qkernel demo on a real block print("\n[3/9] qkernel — tensor_roll kernel family (qsim-classical)") a = argparse.Namespace(workdir=args.workdir, tensor=None, shots=1024, encoding="amplitude", seed=0) cmd_qkernel(a) # 4. compress print("\n[4/9] compress — representation search under hard budget") a = argparse.Namespace(workdir=args.workdir, budget_ratio=S.BUDGET_RATIO, seed=11) cmd_compress(a) # 5. train (distill, subset of stages for demo time) print("\n[5/9] train — recursive distillation") a = argparse.Namespace(workdir=args.workdir, steps=steps_d, batch=128, lr=3e-3, seed=12, stage="all") cmd_train(a) # 6. finetune print("\n[6/9] finetune — task fine-tuning") a = argparse.Namespace(workdir=args.workdir, steps=steps_f, batch=128, seed=13) cmd_finetune(a) # 7. quantize + trq print("\n[7/9] quantize — FP16/BF16/INT8/INT4 + .trq export") a = argparse.Namespace(workdir=args.workdir) cmd_quantize(a) # 8. evaluate + benchmark print("\n[8/9] evaluate + benchmark (measured only)") a = argparse.Namespace(workdir=args.workdir) cmd_evaluate(a) print() cmd_benchmark(a) # 9. sandbox + live inference print("\n[9/9] sandbox + live inference") a = argparse.Namespace(workdir=args.workdir, demo=True, sandbox_cmd=None) cmd_sandbox(a) s_params, s_cfg = _student(wdir, "int4") t_params, t_cfg = _teacher(wdir) print(f"\nthis run: teacher={fmt_params(M.count_params(t_params))} params, " f"student={fmt_params(M.count_params(s_params))} params " f"({M.count_params(s_params)/M.count_params(t_params):.1%} of teacher; " f"target ratio {S.BUDGET_RATIO:.4f})") rng = np.random.default_rng(99) ids = M.gen_data(rng, 1, s_cfg) print("\ninference (int4 student, streaming, tensor> prompt):") print(f" tensor> generate {ids[0, :4].tolist()}") print(" ", end="", flush=True) def _emit(tok): print(tok, end=" ", flush=True) toks, tps, dt = M.generate_stream(s_params, s_cfg, ids[0, :4].tolist(), 24, on_token=_emit) print() print(f" {tps:.1f} tok/s, {fmt_bytes(sum(v.nbytes for v in s_params.values()))} " f"weights, backend={C.CLASSICAL_BACKEND}, rss={rss_mb():.0f}MB") tt, _, _ = M.generate(t_params, t_cfg, ids[0, :4].tolist(), 24) agree = sum(a == b for a, b in zip(toks, tt)) print(f" teacher/student token agreement: {agree}/24") # 10. failure-resolution status board (all values measured below) print("\n[10/9] failure-resolution status — measured, not claimed") _demo_status_board(wdir) print("\nDemo complete. Every number above was measured on this machine.") def _demo_status_board(wdir): import glob import importlib.util import unittest import resource # test count: load test modules by file path (fast) without executing repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) n_tests = 0 for tf in sorted(glob.glob(os.path.join(repo_root, "tests", "test_*.py"))): spec = importlib.util.spec_from_file_location( "tr_test_count_mod", tf) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) n_tests += unittest.TestLoader().loadTestsFromModule(mod) \ .countTestCases() print(f" tests discovered: {n_tests} " f"(full suite: python -m unittest discover -s tests)") # validation teacher/student + measured compression t_params, t_cfg = _teacher(wdir) s_params, s_cfg = _student(wdir, "int4") nt, ns = M.count_params(t_params), M.count_params(s_params) print(f" validation teacher: {fmt_params(nt)} params " f"(acc={read_json(os.path.join(wdir, 'teacher_eval.json'))['accuracy']:.3f}); " f"student(int4): {fmt_params(ns)} params") print(f" measured compression: {ns/nt:.4f} " f"(target ratio {S.BUDGET_RATIO:.4f})") # measured INT8/INT4 perplexity from evaluate evp = os.path.join(wdir, "eval.json") if os.path.exists(evp): ev = read_json(evp) ppls = [] for fmt in ("int8", "int4"): if fmt in ev: ppls.append(f"{fmt} PPL={np.exp(ev[fmt]['held_out']['loss']):.2f}") if ppls: print(f" measured perplexity: {', '.join(ppls)}") # backend availability: live probes, never claimed from . import doctor as DOC caps = DOC.probe_capabilities() for b in ("CUDA", "CUDA-Q", "Q#"): state = caps[b]["state"] why = caps[b].get("reason", "") print(f" {b} execution: {state}" + (f" ({why})" if why else "")) # virtual 30B structural scale test — planned, never "validated" from . import ooc as OOC from . import planner as P man = OOC.glimmer30b_manifest() rows = P.scan_rows(man.specs) t0 = time.perf_counter() plan = P.plan(rows) dt = time.perf_counter() - t0 peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 print(f" 30B STRUCTURAL SCALE TEST: virtual " f"{fmt_params(man.total_params)} manifest ({len(rows)} tensors) " f"-> planned {fmt_params(plan['total_target_params'])} " f"(cap {fmt_params(plan['cap'])}: " f"{'OK' if plan['within_cap'] else 'VIOLATION'}) " f"in {dt:.2f}s, peak RSS {peak:.0f}MB, zero weights allocated") print(" actual 30B weights: NOT TESTED " "(no 30B checkpoint on this machine)") print(f" current measured token agreement: see inference above") # --------------------------------------------------------------------------- # main # --------------------------------------------------------------------------- def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="tensor-roll", description="Tensor Roll — Recursive CUDA-Q Model Quantizer") p.add_argument("--workdir", default=DEFAULT_WORKDIR, help="pipeline working directory") sub = p.add_subparsers(dest="cmd", required=True) a = sub.add_parser("inspect", help="tensor inventory + metrics") a.add_argument("--what", choices=["teacher", "student", "trq"], default="teacher") a.add_argument("--trq", default=None) a = sub.add_parser("ingest", help="build/load the teacher model") a.add_argument("--scale", choices=["tiny", "30b"], default="tiny") a.add_argument("--steps", type=int, default=600) a.add_argument("--batch", type=int, default=128) a.add_argument("--lr", type=float, default=3e-3) a.add_argument("--seed", type=int, default=11) sub.add_parser("map", help="per-tensor metric map") a = sub.add_parser("roll", help="run the TensorRoll operator") a.add_argument("--depth", type=int, default=3) a.add_argument("--quantum-policy", choices=["off", "explore", "selective"], default="off") a.add_argument("--budget-ratio", type=float, default=S.BUDGET_RATIO) a = sub.add_parser("qkernel", help="run the tensor_roll kernel family") a.add_argument("--tensor", default=None) a.add_argument("--shots", type=int, default=1024) a.add_argument("--encoding", choices=["amplitude", "angle"], default="amplitude") a.add_argument("--seed", type=int, default=0) a = sub.add_parser("compress", help="search the 8B representation") a.add_argument("--budget-ratio", type=float, default=S.BUDGET_RATIO) a.add_argument("--seed", type=int, default=11) a.add_argument("--teacher", default=None, help="teacher weight dir (.safetensors/.npz); " "default: workdir (in-RAM path's teacher.npz)") a.add_argument("--target-params", type=float, default=8.0e9, help="global student budget (default 8B, cap 8.5B)") a.add_argument("--memory-budget", default="512MB", help="streaming working-set ceiling, e.g. 512MB, 5G") a.add_argument("--out-of-core", action="store_true", help="streaming SCAN/PLAN/EXECUTE; never holds the " "teacher in RAM") a.add_argument("--resume", action="store_true", help="resume from the per-tensor checkpoint") a = sub.add_parser("train", help="recursive distillation stages") a.add_argument("--stage", default="all", help="'all' or a stage number 0-7") a.add_argument("--steps", type=int, default=120) a.add_argument("--batch", type=int, default=128) a.add_argument("--lr", type=float, default=3e-3) a.add_argument("--seed", type=int, default=12) a = sub.add_parser("finetune", help="task fine-tuning") a.add_argument("--steps", type=int, default=150) a.add_argument("--batch", type=int, default=128) a.add_argument("--seed", type=int, default=13) sub.add_parser("quantize", help="FP16/BF16/INT8/INT4 + .trq export") sub.add_parser("evaluate", help="measured quality metrics") sub.add_parser("benchmark", help="measured comparison table") a = sub.add_parser("chat", help="terminal inference REPL") a.add_argument("--variant", choices=FORMATS, default="fp32") a.add_argument("--compare", action="store_true") a.add_argument("--max-tokens", type=int, default=24) a = sub.add_parser("sandbox", help="OS-level sandbox supervisor") g = a.add_mutually_exclusive_group(required=True) g.add_argument("--cmd", dest="sandbox_cmd", default=None, help="command to run sandboxed") g.add_argument("--demo", action="store_true", help="allow/deny demonstration") a = sub.add_parser("doctor", help="backend capability matrix (honest)") a.add_argument("--json", action="store_true", help="also write doctor_report.json to the workdir") a = sub.add_parser("demo", help="full executable demonstration") a.add_argument("--fast", action="store_true", help="reduced training steps (still real training)") return p def main(): args = build_parser().parse_args() cmds = { "inspect": cmd_inspect, "ingest": cmd_ingest, "map": cmd_map, "roll": cmd_roll, "qkernel": cmd_qkernel, "compress": cmd_compress, "train": cmd_train, "finetune": cmd_finetune, "quantize": cmd_quantize, "evaluate": cmd_evaluate, "benchmark": cmd_benchmark, "chat": cmd_chat, "sandbox": cmd_sandbox, "demo": cmd_demo, "doctor": cmd_doctor, } cmds[args.cmd](args) if __name__ == "__main__": main()