"""Group damage sweep (battlefield teacher). For each tied group / singleton: baseline (everything Q5_K + pins) vs variant (group dropped to Q4_K). Label = PPL(variant) - PPL(baseline) — the measured damage used to train the tiny importance net. Usage: python group_damage_sweep.py --model M.gguf --imatrix I.gguf --out models/damage.jsonl [--resume] ~4.5 min/label on 1.7B (quant + PPL). ~130 labels ≈ 10h. Run overnight. """ import argparse import json import os import re import subprocess import sys import time from concurrent.futures import ThreadPoolExecutor sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from model_reader import read_model from imatrix_reader import read_imatrix, detect_tied_groups, build_importance_table from classifier import build_groups from config_generator import generate_flags from quantizer import run_quantization REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) BIN = os.environ.get("LLAMA_QUANTIZE", "/home/wepiqx/llama.cpp/build/bin/llama-quantize") PPL = os.environ.get("LLAMA_PPL", "/home/wepiqx/llama.cpp/build/bin/llama-perplexity") WIKI = os.environ.get("WIKITEXT_DATA", "/mnt/Vsio/wikitext-2-raw/wiki.test.raw") def run_ppl(model_path: str, ctx: int = 1024, n: int = 64, wiki: str | None = None, kld_base: str | None = None, timeout: int = 1200, ngl: int = 99, do_plain: bool = True): """Returns (ppl, kld|None, extra dict). NOTE (build >=11051 scar): --kl-divergence REPLACES perplexity instead of augmenting it (no 'Final estimate' line). So with kld_base we run TWICE: plain (PPL Final, protocol-comparable) + KLD flags (Mean KLD). Costs 2x PPL time per unit — honesty over speed. The KLD pass prints a whole statistics section; we used to read ONE line (Mean KLD). Now also parsed, same run, zero extra compute: Maximum KLD (tail/detonation), Median KLD (skew vs mean), RMS dp (behavioral shift), Same-top-p (top-1 agreement = decisiveness). do_plain=False (--kld-only): skip the PPL pass entirely — it lies (Perplexity Illusion), and on the forge it doubles wall time. ppl comes back None; callers must treat it as "not measured". """ cmd = [PPL, "-m", model_path, "-f", wiki or WIKI, "-ngl", str(ngl), "--load-mode", "mmap", "-c", str(ctx), "-n", str(n), "-b", os.environ.get("LLAMA_PPL_BATCH", "512"), "--seed", "7"] thr = os.environ.get("LLAMA_THREADS") # forge has 8 logical CPUs, the if thr: # default is half — KLD softmax cmd += ["-t", str(thr)] # and ubatch prep are CPU-side ppl, kld = None, None extra = {} if do_plain: out = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) for line in (out.stderr + out.stdout).splitlines(): if "Final estimate" in line: ppl = float(line.split("PPL =")[1].split("+/-")[0].strip()) if ppl is None: tail = "\n".join((out.stderr + out.stdout).splitlines()[-8:]) raise RuntimeError(f"no PPL in output for {model_path}:\n{tail}") if kld_base: kcmd = cmd + ["--kl-divergence", "--kl-divergence-base", kld_base] out2 = subprocess.run(kcmd, capture_output=True, text=True, timeout=timeout) for line in (out2.stderr + out2.stdout).splitlines(): s = line.strip() if "Mean" in s and "KLD" in s and "Δp" not in s: try: kld = float(s.split("KLD:")[1].split("±")[0].strip()) except (IndexError, ValueError): pass elif s.startswith("Maximum KLD:"): try: extra["kld_max"] = float(s.split(":")[1].strip()) except (IndexError, ValueError): pass elif s.startswith("Median") and "KLD" in s: try: extra["kld_median"] = float(s.split(":")[1].strip()) except (IndexError, ValueError): pass elif s.startswith("RMS"): try: extra["dp_rms"] = float(s.split(":")[1].split("±")[0].strip()) / 100.0 except (IndexError, ValueError): pass elif s.startswith("Same top p:"): try: extra["same_top"] = float(s.split(":")[1].split("±")[0].strip()) / 100.0 except (IndexError, ValueError): pass if kld is None: tail = "\n".join((out2.stderr + out2.stdout).splitlines()[-8:]) raise RuntimeError(f"no KLD in output for {model_path}:\n{tail}") return ppl, kld, extra def main(): ap = argparse.ArgumentParser() ap.add_argument("--model", required=True) ap.add_argument("--imatrix", required=True) ap.add_argument("--out", required=True) ap.add_argument("--base-tier", default="Q5_K") ap.add_argument("--drop-tier", default="Q4_K") ap.add_argument("--resume", action="store_true") ap.add_argument("--wiki-file", default=None, help="PPL corpus (default: $WIKITEXT_DATA or home path)") ap.add_argument("--per-tensor", action="store_true", help="v2: units are single tensors (finer Gnom labels), " "not tied groups") ap.add_argument("--ppl-ctx", type=int, default=1024) ap.add_argument("--timeout", type=int, default=1200, help="per-PPL subprocess timeout, seconds " "(slow backends: 3600+ for Vega)") ap.add_argument("--ngl", type=int, default=99, help="GPU layers for PPL (big BF16 on small VRAM: " "lower until context fits, e.g. 24-32 on 8GB)") ap.add_argument("--units", default=None, help="comma-separated unit tags to measure (subset mode " "for the active loop); default: all unfinished") ap.add_argument("--ppl-n", type=int, default=64, help="v2: short protocol (-n 32) if damage is ctx-invariant") ap.add_argument("--kld-base", default=None, help="KLD-teacher: path to ref logits .dat (from " "--save-all-logits on unquantized model). When set, " "each unit also scores KLD vs ref; damage_kld is " "stored alongside PPL damage.") ap.add_argument("--kld-only", action="store_true", help="skip the plain-PPL pass (PPL lies, and it doubles " "wall time on slow backends). Requires --kld-base. " "damage/ppl are written as null = not measured.") args = ap.parse_args() if args.kld_only and not args.kld_base: ap.error("--kld-only requires --kld-base (no ref = nothing to score)") model = read_model(args.model) im = read_imatrix(args.imatrix) tg = detect_tied_groups(im) imp_table = build_importance_table(im, model) ne_map = {k: v["n_elements"] for k, v in model.get("tensors", {}).items()} # imatrix built by another llama.cpp generation may carry row names # WITHOUT the current .weight suffix (Ling scar, 2026-10-23): those # ghosts also become anchored --tensor-type rules downstream, and # llama-quantize splits the rule at the FIRST '=' — the '=' hidden # inside the '(?=' lookahead then poisons the parse. Real units only: # ghosts are not in the model, so they cannot be measured anyway. for tname, info in imp_table.items(): if tname not in ne_map and tname in model.get("tensors", {}): ne_map[tname] = info["n_elements"] # pool = everything the queue would fight over (no MTP here by construction # on dense models; keep pins as fixed like production) from constants import get_tensor_type, EMBD_PIN_TYPES, ROUTER_PIN_TYPES pool = set(ne_map.keys()) for pin in (EMBD_PIN_TYPES, ROUTER_PIN_TYPES): pool = {n for n in pool if imp_table.get(n, {}).get("type", get_tensor_type(n)) not in pin} pool = {n for n in pool if "norm" not in n} os.makedirs(os.path.join(REPO_ROOT, "models"), exist_ok=True) if args.per_tensor: units = [[n] for n in sorted(pool)] # v2: finer labels, no sharing else: groups = build_groups(tg, pool, ne_map, dict(ne_map)) units = [g for _, (g, _, _, *_) in sorted(groups.items())] print(f"units: {len(units)}", flush=True) # Failure ledger (mailbox 2026-09-28): a failed unit is a LABEL, not a # hole — but it must never enter the labels file, or --resume would # skip retrying it. Separate file, trainer reads both, resume untouched. fail_path = os.path.splitext(args.out)[0] + ".failures.jsonl" def log_failure(tag, stage, err, tensors): # HOLE vs TERMINAL (mailbox 2026-09-28): first failure of a unit is # a hole — logged aside, retried by --resume. A unit that fails # TWICE (this attempt + a prior hole) is TERMINAL: the measurement # is physically untakeable at this tier (censored observation). # TERMINAL goes INTO the labels file as a null record so --resume # never retries it and the trainer can see it (never impute it). prior = 0 try: if os.path.exists(fail_path): with open(fail_path) as f: for line in f: try: if json.loads(line).get("unit") == tag: prior += 1 except (ValueError, AttributeError): pass except OSError: pass if prior >= 1: rec = {"unit": tag, "tensors": tensors, "tier": args.drop_tier, "damage": None, "terminated": f"{stage}:{err}"[:200], "retryable": False} if kb: rec["damage_kld"] = None try: with open(args.out, "a") as f: f.write(json.dumps(rec) + "\n") print(f"{tag} TERMINAL ({stage}) — censored label, " f"will not retry", flush=True) except OSError: pass return try: with open(fail_path, "a") as f: f.write(json.dumps({"unit": tag, "status": "failed", "stage": stage, "error": str(err)[:200]}) + "\n") except OSError: pass done = set() if args.resume and os.path.exists(args.out): with open(args.out) as f: for line in f: done.add(json.loads(line)["unit"]) from constants import ROUTER_PIN_TYPES def build_file_inner(drop: list | None, path: str): assignments = {} # base assignments(truth, not hope): droppable tensors at base tier, # norm/router/F32 pins stay UNTOUCHED — quantizing a router (the # constants.py 'verified disaster') into the BASELINE would shift # every row by the same detonation instead of leaving the pins out. for n in ne_map: if get_tensor_type(n) in ROUTER_PIN_TYPES: # router is NOT covered by the binary's F32 law — a plain # base_type=Q5_K positional would quantize it silently into # every copy (the constants.py 'verified disaster'). Pin F32 # by an explicit rule; a clean blk-pattern without lookaheads. assignments[n] = "F32" elif "norm" in n: continue # f32 map + binary law keep norms F32 else: assignments[n] = args.base_tier if drop: for n in drop: assignments[n] = args.drop_tier flags = generate_flags(assignments, model, args.base_tier, 100000) flags["imatrix"] = args.imatrix print("DBG rules:", len(flags["tensor_type_rules"]), "anchored:", sum(1 for r in flags["tensor_type_rules"] if "(?=" in r), flush=True) for r in flags["tensor_type_rules"]: if "(?=" in r: print("DBG anchored:", r, flush=True) ok = run_quantization(flags, args.model, path) if not ok: raise RuntimeError("quant failed") def _looks_valid(path: str) -> bool: """GGUF magic + sane size (>100MB for 1.7B quants).""" try: if os.path.getsize(path) < 100 * 1024 * 1024: return False with open(path, "rb") as f: return f.read(4) == b"GGUF" except OSError: return False def _wait_then_build(prev_fut, unit, path): # Serialization only (one quantize at a time: RAM). A dead # predecessor must NEVER poison this unit: 2026-10-03 scar — # one OOM-killed build cascaded RuntimeError through 148 # downstream units that never even ran. Catch, then build anyway; # own failures surface via own future below. try: prev_fut.result() except Exception: pass build_file_inner(unit, path) build_file = build_file_inner tmp = os.path.join(REPO_ROOT, "models", ".sweep_tmp.gguf") # crash orphans: stale .sweep_* buffers from killed runs (else disk death) for fn in os.listdir(os.path.join(REPO_ROOT, "models")): if fn.startswith(".sweep_") and fn.endswith(".gguf"): try: os.remove(os.path.join(REPO_ROOT, "models", fn)) print(f"orphan removed: {fn}", flush=True) except OSError: pass kb = args.kld_base kld_only = args.kld_only if "BASELINE" not in done: build_file(None, tmp) base_ppl, base_kld, base_extra = run_ppl(tmp, args.ppl_ctx, args.ppl_n, args.wiki_file, kb, timeout=args.timeout, ngl=args.ngl, do_plain=not kld_only) rec = {"unit": "BASELINE", "ppl": None if kld_only else base_ppl} if kb: rec["kld"] = base_kld rec.update(base_extra) with open(args.out, "a") as f: f.write(json.dumps(rec) + "\n") print("BASELINE " + (f"ppl={base_ppl:.4f}" if base_ppl is not None else "ppl=null (kld-only, not measured)") + (f" kld={base_kld:.4f}" if base_kld is not None else ""), flush=True) else: with open(args.out) as f: rows = [json.loads(l) for l in f if json.loads(l)["unit"] == "BASELINE"] base_ppl = rows[0].get("ppl") base_kld = rows[0].get("kld") base_extra = {k: rows[0].get(k) for k in ("kld_max", "kld_median", "dp_rms", "same_top")} def unit_tag(unit): return "+".join(n.replace(".weight", "").split(".")[-1] + "@" + n.split(".")[1] for n in unit if n.startswith("blk.")) or "global" # Pipelined: quant(unit i+1) on CPU overlaps PPL(unit i) on GPU. # UNIQUE tmp file per unit: rotating buffers raced when a slow PPL # (loaded system) let a later build overwrite the file PPL still read. # BOUNDED window (2026-10-03 scar): the old code submitted ALL builds # upfront — builders outran PPL, buffers + page cache grew all run, # and the tail died (OOM/pressure). At most MAX_INFLIGHT builds may # be ahead of consumption; the rest wait unsubmitted. ≤3 files alive # (~3.4GB per 1.7B unit), deleted right after use. MAX_INFLIGHT = 3 todo = [(unit_tag(u), u) for u in units if unit_tag(u) not in done] if args.units: want = {s.strip() for s in args.units.split(",") if s.strip()} todo = [(t, u) for t, u in todo if t in want] print(f"subset mode: {len(todo)} units requested", flush=True) safe = lambda tag, i: re.sub(r"[^A-Za-z0-9_@+-]", "_", tag)[:60] ex = ThreadPoolExecutor(max_workers=2) submitted = {} # todo_idx -> (tag, unit, buf, future) def ensure_submitted(idx): if idx < 0 or idx >= len(todo) or idx in submitted: return tag, unit = todo[idx] buf = os.path.join(REPO_ROOT, "models", f".sweep_{idx}_{safe(tag, idx)}.gguf") prev = submitted[idx - 1][3] if idx - 1 in submitted else None if prev is None: fut = ex.submit(build_file, unit, buf) else: fut = ex.submit(_wait_then_build, prev, unit, buf) submitted[idx] = (tag, unit, buf, fut) try: if todo: for i in range(min(MAX_INFLIGHT, len(todo))): ensure_submitted(i) n_total = len(todo) for idx in range(n_total): ensure_submitted(idx) tag, unit, buf, fq = submitted[idx] try: fq.result() # quant done (previous PPL overlapped it) except Exception as e: print(f"[{idx + 1}/{n_total}] {tag} QUANT failed " f"({type(e).__name__}), skipped", flush=True) log_failure(tag, "quant", f"{type(e).__name__}", unit) try: os.remove(buf) except OSError: pass ensure_submitted(idx + MAX_INFLIGHT) del submitted[idx] continue if not _looks_valid(buf): print(f"[{idx + 1}/{n_total}] {tag} CORRUPT file, " f"skipped", flush=True) log_failure(tag, "corrupt", "bad magic/size", unit) try: os.remove(buf) except OSError: pass ensure_submitted(idx + MAX_INFLIGHT) del submitted[idx] continue try: p, k, extra = run_ppl(buf, args.ppl_ctx, args.ppl_n, args.wiki_file, kb, timeout=args.timeout, ngl=args.ngl, do_plain=not kld_only) except Exception as e: print(f"[{idx + 1}/{n_total}] {tag} PPL failed " f"({type(e).__name__}: {str(e)[:100]}), skipped", flush=True) log_failure(tag, "ppl", f"{type(e).__name__}: {str(e)[:100]}", unit) ensure_submitted(idx + MAX_INFLIGHT) del submitted[idx] continue finally: try: os.remove(buf) except OSError: pass # null-safe: kld-only leaves ppl None, a censored baseline # leaves base_ppl None — "not measured", never a crash. dmg = (p - base_ppl if p is not None and base_ppl is not None else None) rec = {"unit": tag, "tensors": unit, "ppl": p, "damage": dmg} if kb: rec["kld"] = k # null-safe: a unit (or a resumed BASELINE) without KLD is # a censored observation, never a crash (2026-10-13 scar: # resumed PPL-only BASELINE fed None into this subtraction). rec["damage_kld"] = (k - base_kld if k is not None and base_kld is not None else None) for ek in ("kld_max", "kld_median", "dp_rms", "same_top"): if ek in extra and extra[ek] is not None: rec[ek] = extra[ek] b = (base_extra or {}).get(ek) if b is not None: rec["damage_" + ek] = extra[ek] - b with open(args.out, "a") as f: f.write(json.dumps(rec) + "\n") print(f"[{idx + 1}/{n_total}] {tag} " + (f"damage={dmg:+.4f}" if dmg is not None else "damage=null") + (f" kld={k - base_kld:+.4f}" if kb and k is not None and base_kld is not None else ""), flush=True) ensure_submitted(idx + MAX_INFLIGHT) del submitted[idx] finally: ex.shutdown(wait=False) if __name__ == "__main__": main()