refactor(quant): one corpus plumbing module, and the builders out of main()

The three corpus tools now share scripts/qwen38_common.py: tokenizer load with the
venv guidance, one n_tokens (so a KLD count cannot drift from the corpus it scores),
JSONL read/write, an atomic writer, and refuse(). make-calib-corpus.py's 170-line
main() with three nested closures became classify / split_pools / fragment_text /
build_frags / pack_class / emit_calib / emit_kld / manifest; its magic numbers are
named constants. rechunk-kld-corpus.py's main became validate / cut / rechunk around
a Window record, so the round-trip proof sits in one function that can be tested
without a tokenizer. calib-push.py grew parse_args / load_prompts / push_round, and
a malformed corpus now refuses with a line number instead of raising KeyError.

Parity, measured on dankserver against the pre-refactor scripts at the same seed:
calib.jsonl and kld.jsonl carry identical records in identical order, the manifest
is identical, and the re-chunked corpus is byte-identical (md5 e824e67d...). The one
byte-level change is JSON escaping — write_jsonl emits literal UTF-8, which is what
the re-chunker already wrote; the parsed records do not move.

scripts/calib-push.py

@@ -22,6 +22,9 @@ Mechanics notes:
   prompts do not bias the Hessians (the corpus is drawn round-robin over
   source classes, so loss is class-neutral as long as failures are rare;
   a systematic 4xx is loud in the log and must be fixed, not ignored).
+- a round that seals no tap bytes is a FAILURE, not a retry: the engine
+  answering 200 while the tap directory is wrong, full or read-only is
+  how a corpus gets pushed four times for nothing.
 
 Usage: calib-push.py [--port 7770] [--parallel 64] [--container arbench]
        [--corpus /home/dank/corpora/out/calib.jsonl] [--max-rounds 4]
@@ -29,11 +32,18 @@ Usage: calib-push.py [--port 7770] [--parallel 64] [--container arbench]
 import argparse
 import concurrent.futures
 import json
+import os
 import subprocess
 import sys
 import time
 import urllib.request
 
+_HERE = os.path.dirname(os.path.abspath(__file__)) if "__file__" in globals() else os.getcwd()
+if _HERE not in sys.path:
+    sys.path.insert(0, _HERE)
+
+import qwen38_common as qc  # noqa: E402 - sibling module, same directory
+
 DONE_MARK = "calibration wrote"
 
 
@@ -49,9 +59,9 @@ def tap_done(container):
     return DONE_MARK in (out.stdout + out.stderr)
 
 
-def push_one(url, prompt):
+def push_one(url, prompt, model):
     body = json.dumps({
-        "model": "frank-archon",
+        "model": model,
         "prompt": prompt,
         "max_tokens": 1,
         "temperature": 0,
@@ -68,38 +78,89 @@ def push_one(url, prompt):
         return False
 
 
-def main():
+def push_round(url, prompts, model, parallel):
+    """Push the whole corpus once; returns how many requests were accepted."""
+    ok = 0
+    with concurrent.futures.ThreadPoolExecutor(parallel) as ex:
+        for good in ex.map(lambda p: push_one(url, p, model), prompts):
+            ok += 1 if good else 0
+    return ok
+
+
+def tap_bytes(path):
+    """Bytes the tap has sealed so far; 0 when the directory is absent.
+
+    A tap whose directory is wrong, full or read-only stops writing while the
+    engine keeps answering 200 — the driver then re-pushes the whole corpus for
+    every round it is allowed and the loss is invisible until the hours are gone.
+    """
+    total = 0
+    try:
+        for root, _, files in os.walk(path):
+            for f in files:
+                try:
+                    total += os.path.getsize(os.path.join(root, f))
+                except OSError:
+                    pass
+    except OSError:
+        return 0
+    return total
+
+
+def parse_args():
     ap = argparse.ArgumentParser()
     ap.add_argument("--port", type=int, default=7770)
     ap.add_argument("--parallel", type=int, default=64)
     ap.add_argument("--container", default="arbench")
+    ap.add_argument("--model", default="frank-archon")
+    ap.add_argument("--tap-dir", default="/home/dank/campaign/qwen38-w5nl64/calib")
     ap.add_argument("--corpus", default="/home/dank/corpora/out/calib.jsonl")
     ap.add_argument("--max-rounds", type=int, default=4)
-    a = ap.parse_args()
-    url = f"http://127.0.0.1:{a.port}/v1/completions"
+    return ap.parse_args()
 
+
+def load_prompts(a) -> list[str]:
+    if not os.path.isfile(a.corpus):
+        qc.refuse(f"no corpus at {a.corpus}")
     prompts = []
-    with open(a.corpus) as f:
-        for line in f:
-            line = line.strip()
-            if line:
-                prompts.append(json.loads(line)["prompt"])
-    print(f"corpus: {len(prompts)} prompts", flush=True)
+    for lineno, rec in qc.iter_jsonl(a.corpus):
+        if not isinstance(rec.get("prompt"), str):
+            qc.refuse(f"{a.corpus}:{lineno} has no string 'prompt' field")
+        prompts.append(rec["prompt"])
+    if not prompts:
+        qc.refuse(f"the corpus at {a.corpus} holds no prompts")
+    return prompts
+
+
+def main():
+    a = parse_args()
+    url = f"http://127.0.0.1:{a.port}/v1/completions"
+    prompts = load_prompts(a)
+    chars = sum(len(p) for p in prompts)
+    print(f"corpus: {len(prompts)} prompts, ~{chars // 4} tokens (chars/4)",
+          flush=True)
 
     for rnd in range(1, a.max_rounds + 1):
         if tap_done(a.container):
             break
+        before = tap_bytes(a.tap_dir)
         t0 = time.time()
-        ok = 0
-        with concurrent.futures.ThreadPoolExecutor(a.parallel) as ex:
-            for good in ex.map(lambda p: push_one(url, p), prompts):
-                ok += 1 if good else 0
-        dt = time.time() - t0
+        ok = push_round(url, prompts, a.model, a.parallel)
+        dt = max(time.time() - t0, 1e-9)
         print(f"round {rnd}: {ok}/{len(prompts)} ok in {dt:.0f}s "
-              f"({sum(len(p) for p in prompts)/4/dt:.0f} chars/s est)",
-              flush=True)
+              f"({chars / 4 / dt:.0f} tokens/s est)", flush=True)
+        if ok == 0:
+            print("FAIL: not one prompt was accepted; fix the engine or the argv "
+                  "before spending another round", flush=True)
+            sys.exit(1)
         if tap_done(a.container):
             break
+        after = tap_bytes(a.tap_dir)
+        if after <= before:
+            print(f"FAIL: the tap at {a.tap_dir} sealed nothing this round "
+                  f"({before} B before, {after} B after) — the engine is serving "
+                  "without recording (wrong dir, full disk, read-only mount)", flush=True)
+            sys.exit(1)
     else:
         print("max rounds exhausted, tap not done", flush=True)
         sys.exit(1)

scripts/convert-qwen38-w5nl64.sh

@@ -71,10 +71,12 @@ PLAN_ONLY=0
 KEEP_RECIPE=0
 while [ "$#" -gt 0 ]; do
     case "$1" in
-      --arm) ARMS+=("$2"); shift 2 ;;
+      --arm) [ "$#" -ge 2 ] || { echo "--arm needs a value (cd3|cd6|ao1)" >&2; exit 2; }
+             ARMS+=("$2"); shift 2 ;;
       --dry-run) DRY=1; shift ;;
       --plan-only) PLAN_ONLY=1; shift ;;
-      --target-count) TARGET_COUNT="$2"; shift 2 ;;
+      --target-count) [ "$#" -ge 2 ] || { echo "--target-count needs a value" >&2; exit 2; }
+             TARGET_COUNT="$2"; shift 2 ;;
       --keep-recipe) KEEP_RECIPE=1; shift ;;
       *) echo "unknown argument: $1" >&2; exit 2 ;;
     esac
@@ -124,6 +126,13 @@ fi
 for a in "${ARMS[@]}"; do
     [ -e "$OUTDIR/qwen38-w5nl64-$a.rad" ] && {
         echo "$OUTDIR/qwen38-w5nl64-$a.rad already exists; move it to trash first" >&2; exit 2; }
+    # A killed run leaves its container behind (--rm only fires when podman itself
+    # exits), and the next arm then dies on "name already in use" AFTER the plan
+    # phase; refuse here, where the fix is one `podman rm -f`.
+    podman container exists "qwen38-convert-$a" && {
+        echo "container qwen38-convert-$a still exists (a killed run?);" \
+             "'podman rm -f qwen38-convert-$a' and re-run" >&2
+        exit 2; }
 done
 
 # --- the protected plane, spliced into a per-run copy of the template --------------------------
@@ -203,6 +212,7 @@ for arm in "${ARMS[@]}"; do
     rc=0; run_convert "$arm" || rc=$?
     [ "$rc" -eq 0 ] || { echo "arm $arm: rad-convert rc=$rc; log: $STATE/convert-$arm-$TS.log" >&2; overall=1; continue; }
     out="$OUTDIR/qwen38-w5nl64-$arm.rad"
+    [ -f "$out" ] || { echo "arm $arm: rad-convert exited 0 but wrote no $out; log: $STATE/convert-$arm-$TS.log" >&2; overall=1; continue; }
     got_mib=$(( $(stat -c %s "$out") / 1048576 ))
     # The receipt: the artifact is what the plan predicted, within the metadata the plan does not
     # count. An arm that came out materially smaller lost a plane somewhere.

scripts/make-calib-corpus.py

@@ -34,14 +34,25 @@ import json
 import os
 import random
 import re
+import sys
 
 from jinja2 import Environment, FileSystemLoader
-from tokenizers import Tokenizer
+
+_HERE = os.path.dirname(os.path.abspath(__file__)) if "__file__" in globals() else os.getcwd()
+if _HERE not in sys.path:
+    sys.path.insert(0, _HERE)
+
+import qwen38_common as qc  # noqa: E402 - sibling module, same directory
 
 CODE_EXT = (".sh", ".py", ".jinja", ".recipe", ".env", ".example",
             ".yaml", ".yml", ".bats", ".service", ".container", ".toml")
 JSONL_EXTS = (".jsonl", ".json")
 DOC_DIR = "serverAppConfigs-docs"
+FRAGMENT_TOKENS = 2400   # what one fragment may contribute
+CLASS_MIX = {"code": .45, "prose": .30, "tools": .25}
+HF_CAP = {"code": 40000, "prose": 60000, "tools": 0}
+POOL_CYCLES = 40         # how many times a class pool may repeat before giving up
+KLD_GUARD = 5000         # rendered conversations tried before declaring the pool short
 
 FLEET_CMDS = [
     "systemctl status frank-archon --no-pager",
@@ -142,7 +153,7 @@ def load_parquet_texts(path: str, kind: str, rng: random.Random,
     return out[:cap]
 
 
-def main() -> None:
+def parse_args() -> argparse.Namespace:
     ap = argparse.ArgumentParser(description=__doc__)
     ap.add_argument("--repo", required=True)
     ap.add_argument("--tokenizer-json", required=True)
@@ -156,22 +167,26 @@ def main() -> None:
     ap.add_argument("--hf-code-parquet", default="")
     ap.add_argument("--hf-prose-parquet", default="")
     ap.add_argument("--seed", type=int, default=20261006)
-    a = ap.parse_args()
+    return ap.parse_args()
 
-    rng = random.Random(a.seed)
-    tokenizer = Tokenizer.from_file(a.tokenizer_json)
-    env = Environment(loader=FileSystemLoader(os.path.dirname(a.template)),
+
+def load_template(template: str) -> tuple[Environment, str]:
+    """The chat template as the engine will render it, plus its short name."""
+    env = Environment(loader=FileSystemLoader(os.path.dirname(template)),
                       trim_blocks=True, lstrip_blocks=True)
     env.filters["tojson"] = lambda v: json.dumps(v, ensure_ascii=False)
     env.filters["raise_exception"] = lambda m: (_ for _ in ()).throw(
         ValueError(m))
-    tname = os.path.basename(a.template)
+    return env, os.path.basename(template)
 
+
+def classify(repo: str) -> dict[str, list[str]]:
+    """Walk the checkout into the three source classes the mix draws from."""
     code, prose, jsonl = [], [], []
-    for root, dirs, files in os.walk(a.repo):
+    for root, dirs, files in os.walk(repo):
         dirs[:] = [d for d in dirs if not d.startswith(".")]
         for f in sorted(files):
-            p = os.path.relpath(os.path.join(root, f), a.repo)
+            p = os.path.relpath(os.path.join(root, f), repo)
             if DOC_DIR in p and f.endswith(".md"):
                 prose.append(p)
             elif p.startswith(DOC_DIR):
@@ -180,137 +195,185 @@ def main() -> None:
                 code.append(p)
             elif f.endswith(JSONL_EXTS) and f != "package-lock.json":
                 jsonl.append(p)
-    # file-level split: bucket 0 -> KLD pool, 1..4 -> calibration pool
-    pools: dict[str, tuple[list, list]] = {}
-    for cls, lst in (("code", code), ("prose", prose), ("tools", jsonl)):
-        pools[cls] = ([p for p in lst if fnv(p) % 5 == 0],
-                      [p for p in lst if fnv(p) % 5 != 0])
+    return {"code": code, "prose": prose, "tools": jsonl}
 
-    def text_of(p: str) -> str:
-        path = os.path.join(a.repo, p)
-        raw = open(path, encoding="utf-8", errors="replace").read()
-        if p.endswith(JSONL_EXTS):
-            lines = [json.loads(l)["content"] for l in raw.splitlines()
-                     if l.strip() and "content" in l]
-            raw = "\n".join(lines)
-        toks = tokenizer.encode(raw, add_special_tokens=False).ids
-        return tokenizer.decode(toks[:2400], skip_special_tokens=True)
 
+def split_pools(classes: dict[str, list[str]]) -> dict[str, tuple[list, list]]:
+    """File-level split: bucket 0 -> KLD pool, everything else -> calibration."""
+    return {cls: ([p for p in lst if fnv(p) % 5 == 0],
+                  [p for p in lst if fnv(p) % 5 != 0])
+            for cls, lst in classes.items()}
+
+
+def jsonl_content(raw: str) -> str:
+    """The content fields of a content-keyed JSONL, or "" for anything else.
+
+    A .json in this repo is not necessarily a content-keyed JSONL: the substring
+    test passes on any line that merely mentions "content", and a list-shaped
+    document raises TypeError — both used to crash the whole build. Skip what
+    does not fit instead.
+    """
+    lines = []
+    for line in raw.splitlines():
+        if not line.strip() or "content" not in line:
+            continue
+        try:
+            rec = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(rec, dict) and isinstance(rec.get("content"), str):
+            lines.append(rec["content"])
+    return "\n".join(lines)
+
+
+def fragment_text(repo: str, rel: str, tok) -> str:
+    """One file as a fragment: its first FRAGMENT_TOKENS tokens, decoded."""
+    raw = open(os.path.join(repo, rel), encoding="utf-8",
+               errors="replace").read()
+    if rel.endswith(JSONL_EXTS):
+        raw = jsonl_content(raw)
+    ids = tok.encode(raw, add_special_tokens=False).ids
+    return tok.decode(ids[:FRAGMENT_TOKENS], skip_special_tokens=True)
+
+
+def build_frags(a: argparse.Namespace, tok,
+                rng: random.Random) -> dict[str, dict[str, list[str]]]:
+    """Both fragment pools per class, seeded so the same seed rebuilds the corpus."""
+    pools = split_pools(classify(a.repo))
     cmds = harvest_cmds(a.repo, rng)
-    # fragment pools per class, corpus-split first
     frags: dict[str, dict[str, list[str]]] = {}
-    for cls in pools:
-        kld_files, cal_files = pools[cls]
-        fr = {}
-        src = {"kld": kld_files, "cal": cal_files}
+    for cls, (kld_files, cal_files) in pools.items():
         if cls == "tools":
-            fr["kld"] = [tool_call(rng, cmds) for _ in range(600)]
-            fr["cal"] = [tool_call(rng, cmds) for _ in range(40000)]
+            frags[cls] = {"kld": [tool_call(rng, cmds) for _ in range(600)],
+                          "cal": [tool_call(rng, cmds) for _ in range(40000)]}
         else:
-            fr["kld"] = [text_of(p) for p in src["kld"]]
-            fr["cal"] = [text_of(p) for p in src["cal"]]
-        frags[cls] = fr
+            frags[cls] = {"kld": [fragment_text(a.repo, p, tok) for p in kld_files],
+                          "cal": [fragment_text(a.repo, p, tok) for p in cal_files]}
     # per-class fragment caps: prose rows are short (~66 tok) so need many;
     # codeparrot files are long (~2.6k tok) so a small cap suffices
-    hfcap = {"code": 40000, "prose": 60000, "tools": 0}
     for cls, spec in (("code", a.hf_code_parquet),
                       ("prose", a.hf_prose_parquet)):
         for path in [p for p in spec.split(",") if p]:
             frags[cls]["cal"].extend(
-                load_parquet_texts(path, cls, rng, cap=hfcap[cls]))
+                load_parquet_texts(path, cls, rng, cap=HF_CAP[cls]))
     for cls in frags:
         rng.shuffle(frags[cls]["cal"])
+    return frags
 
-    def toks(s: str) -> int:
-        return len(tokenizer.encode(s, add_special_tokens=False).ids)
 
-    def emit_calib() -> tuple[int, list[dict]]:
-        targets = {"code": .45, "prose": .30, "tools": .25}
-        rows: list[dict] = []
-        for cls, frac in targets.items():
-            want = int(a.calib_tokens * frac)
-            pool = frags[cls]["cal"]
+def pack_class(cls: str, pool: list[str], want: int, tok,
+               chunk_tokens: int) -> list[dict]:
+    """Greedily pack `pool` into chunk_tokens-sized calibration prompts."""
+    if not pool:
+        return []
+    rows: list[dict] = []
+    # Count every fragment once: the loop cycles the pool up to POOL_CYCLES
+    # times, and re-tokenizing a 2.4k-token fragment on each pass dominated the
+    # build's wall time.
+    counts = [qc.n_tokens(tok, f) for f in pool]
+    got = 0
+    i = 0
+    buf: list[str] = []
+    nbuf = 0
+    while got < want:
+        j = i % len(pool)
+        frag = pool[j]
+        i += 1
+        n = counts[j]
+        if nbuf + n > chunk_tokens and buf:
+            rows.append({"prompt": "\n".join(buf), "cls": cls})
+            got += nbuf
+            buf, nbuf = [], 0
+            if got >= want:
+                break
+        buf.append(frag)
+        nbuf += n
+        if i > len(pool) * POOL_CYCLES:
+            break  # pool exhausted with repeats; stop
+    if buf and got < want:
+        rows.append({"prompt": "\n".join(buf), "cls": cls})
+        got += nbuf
+    print(f"calib {cls}: {got} tokens, {i} fragments")
+    return rows
+
+
+def emit_calib(a: argparse.Namespace, frags, tok,
+               rng: random.Random) -> tuple[int, list[dict]]:
+    """calib.jsonl rows: the class mix, shuffled, prompts only."""
+    rows: list[dict] = []
+    for cls, frac in CLASS_MIX.items():
+        rows.extend(pack_class(cls, frags[cls]["cal"], int(a.calib_tokens * frac),
+                               tok, a.chunk_tokens))
+    rng.shuffle(rows)
+    return len(rows), [{"prompt": r["prompt"]} for r in rows]
+
+
+def emit_kld(a: argparse.Namespace, frags, tok, rng: random.Random,
+             env: Environment, tname: str) -> tuple[int, list[dict], int]:
+    """kld.jsonl rows: multi-turn conversations, each with a scored tail."""
+    template = env.get_template(tname)
+    rows: list[dict] = []
+    total = scored_total = 0
+    classes = ["code", "prose", "tools"]
+    cursor = {c: 0 for c in classes}
+    guard = 0
+    while total < a.kld_tokens and guard < KLD_GUARD:
+        guard += 1
+        msgs, srcs = [], []
+        for t in range(rng.choice([2, 3, 4])):
+            cls = classes[(len(rows) + t) % 3]
+            pool = frags[cls]["kld"]
             if not pool:
                 continue
-            got, i, buf, nbuf = 0, 0, [], 0
-            while got < want:
-                frag = pool[i % len(pool)]
-                i += 1
-                n = toks(frag)
-                if nbuf + n > a.chunk_tokens and buf:
-                    rows.append({"prompt": "\n".join(buf), "cls": cls})
-                    got += nbuf
-                    buf, nbuf = [], 0
-                    if got >= want:
-                        break
-                buf.append(frag)
-                nbuf += n
-                if i > len(pool) * 40:
-                    break  # pool exhausted with repeats; stop
-            if buf and got < want:
-                rows.append({"prompt": "\n".join(buf), "cls": cls})
-                got += nbuf
-            print(f"calib {cls}: {got} tokens, {i} fragments")
-        rng.shuffle(rows)
-        return len(rows), [{k: v for k, v in r.items() if k == "prompt"}
-                           for r in rows]
+            frag = pool[cursor[cls] % len(pool)]
+            cursor[cls] += 1
+            msgs.append({"role": "user",
+                         "content": "Summarize and explain this "
+                                    f"engineering material:\n{frag}"})
+            msgs.append({"role": "assistant",
+                         "content": frag[:4000]})
+            srcs.append(cls)
+        rendered = template.render(messages=msgs, add_generation_prompt=False)
+        n = qc.n_tokens(tok, rendered)
+        if n < a.kld_min_prompt:
+            continue
+        rows.append({"prompt": rendered,
+                     "score_from": n - a.score_tail,
+                     "source": "+".join(sorted(set(srcs)))})
+        total += n
+        scored_total += a.score_tail
+    return len(rows), rows, scored_total
 
-    def emit_kld() -> tuple[int, list[dict]]:
-        rows, total = [], 0
-        classes = ["code", "prose", "tools"]
-        i = {c: 0 for c in classes}
-        guard = 0
-        while total < a.kld_tokens and guard < 5000:
-            guard += 1
-            msgs, srcs = [], []
-            for t in range(rng.choice([2, 3, 4])):
-                cls = classes[(len(rows) + t) % 3]
-                pool = frags[cls]["kld"]
-                if not pool:
-                    continue
-                frag = pool[i[cls] % len(pool)]
-                i[cls] += 2
-                msgs.append({"role": "user",
-                             "content": "Summarize and explain this "
-                                        f"engineering material:\n{frag}"})
-                msgs.append({"role": "assistant",
-                             "content": frag[:4000]})
-                srcs.append(cls)
-            rendered = env.get_template(tname).render(
-                messages=msgs, add_generation_prompt=False)
-            n = toks(rendered)
-            if n < a.kld_min_prompt:
-                continue
-            rows.append({"prompt": rendered,
-                         "score_from": n - a.score_tail,
-                         "source": "+".join(sorted(set(srcs)))})
-            total += n
-        return len(rows), rows
 
-    os.makedirs(a.out_dir, exist_ok=True)
-    nc, rows = emit_calib()
-    with open(os.path.join(a.out_dir, "calib.jsonl"), "w") as f:
-        for r in rows:
-            f.write(json.dumps(r) + "\n")
-    nk, krows = emit_kld()
-    with open(os.path.join(a.out_dir, "kld.jsonl"), "w") as f:
-        for r in krows:
-            f.write(json.dumps(r) + "\n")
-    scored = sum(len(tokenizer.encode(r["prompt"], add_special_tokens=False).ids)
-                 - r["score_from"] for r in krows)
-    manifest = {
+def manifest(a: argparse.Namespace, tname: str, nc: int, nk: int,
+             scored: int) -> dict:
+    return {
         "seed": a.seed, "template": tname,
         "tokenizer": os.path.basename(a.tokenizer_json),
         "calib_prompts": nc, "kld_prompts": nk,
         "kld_scored_positions": scored,
         "kld_score_tail": a.score_tail,
         "disjoint_file_pools": "local KLD material only; HF code feeds calib",
-        "calib_mix": {"code": .45, "prose": .30, "tools": .25},
+        "calib_mix": dict(CLASS_MIX),
         "hf_code": a.hf_code_parquet, "hf_prose": a.hf_prose_parquet,
     }
-    with open(os.path.join(a.out_dir, "manifest.json"), "w") as f:
-        json.dump(manifest, f, indent=1)
-    print(json.dumps(manifest, indent=1))
+
+
+def main() -> None:
+    a = parse_args()
+    rng = random.Random(a.seed)
+    tok = qc.load_tokenizer(a.tokenizer_json)
+    env, tname = load_template(a.template)
+
+    frags = build_frags(a, tok, rng)
+    os.makedirs(a.out_dir, exist_ok=True)
+    nc, calib_rows = emit_calib(a, frags, tok, rng)
+    qc.write_jsonl(os.path.join(a.out_dir, "calib.jsonl"), calib_rows)
+    nk, kld_rows, scored = emit_kld(a, frags, tok, rng, env, tname)
+    qc.write_jsonl(os.path.join(a.out_dir, "kld.jsonl"), kld_rows)
+    man = manifest(a, tname, nc, nk, scored)
+    qc.write_json(os.path.join(a.out_dir, "manifest.json"), man)
+    print(json.dumps(man, indent=1))
 
 
 if __name__ == "__main__":

scripts/qwen38_common.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""Shared plumbing for the qwen38 campaign corpus tools.
+
+The three tools move the same two shapes — a JSONL of prompt records and a
+one-artifact JSON manifest — and they all tokenize with the SAME tokenizer. That
+last part is not a detail: a KLD reference is only meaningful against the exact
+token sequence the corpus recorded, so `n_tokens` exists once, here, and every
+tool counts the same way. A window that decode/encode does not round-trip is a
+refusal, never a warning.
+
+Loaded as a sibling module. Each tool puts its own directory on `sys.path`
+first, so `python3 scripts/<tool>.py` and the driver's container mount
+(`runpy.run_path('/work/<tool>.py')`, `-v scripts:/work:ro`) both resolve it —
+run the tools by path, not piped through stdin.
+"""
+import contextlib
+import json
+import os
+import sys
+import tempfile
+
+
+def refuse(msg: str) -> None:
+    """Exit 2 with a REFUSED line — the code argparse uses for unusable input."""
+    print(f"REFUSED: {msg}", file=sys.stderr)
+    sys.exit(2)
+
+
+def load_tokenizer(path: str):
+    try:
+        from tokenizers import Tokenizer
+    except ImportError as exc:  # the corpus tools carry their own venv
+        refuse(f"tokenizers is not importable by {sys.executable} — run the "
+               f"corpus venv's python (on dankserver: "
+               f"/home/dank/venvs/radcorpus/bin/python) or set "
+               f"FRANK_QWEN38_TOKENIZER: {exc}")
+    if not os.path.isfile(path):
+        refuse(f"no tokenizer at {path}")
+    return Tokenizer.from_file(path)
+
+
+def n_tokens(tok, text: str) -> int:
+    """The corpus's own count: no special tokens added, prompt side only."""
+    return len(tok.encode(text, add_special_tokens=False).ids)
+
+
+def iter_jsonl(path: str):
+    """Yield `(lineno, record)` for every non-blank line; refuse on a bad line.
+
+    Field validation stays with the caller: the three corpora carry different
+    fields (`prompt` alone, `prompt`+`cls`, `prompt`+`score_from`+`source`), and
+    a record missing the one the tool needs is that tool's refusal to make.
+    """
+    with open(path, encoding="utf-8", errors="replace") as fh:
+        for lineno, line in enumerate(fh, 1):
+            if not line.strip():
+                continue
+            try:
+                rec = json.loads(line)
+            except json.JSONDecodeError as exc:
+                refuse(f"{path}:{lineno} is not JSON: {exc}")
+            yield lineno, rec
+
+
+def write_jsonl(path: str, rows) -> None:
+    with open(path, "w", encoding="utf-8") as fh:
+        for row in rows:
+            fh.write(json.dumps(row, ensure_ascii=False) + "\n")
+
+
+@contextlib.contextmanager
+def atomic_writer(path: str):
+    """Write `path` through a temporary sibling and rename on success.
+
+    An interrupted run then leaves no half corpus behind, and the temporary name
+    keeps the one-artifact-one-name law the driver applies to a score report.
+    """
+    d = os.path.dirname(os.path.abspath(path)) or "."
+    fd, tmp = tempfile.mkstemp(dir=d, prefix="." + os.path.basename(path) + "-")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as fh:
+            yield fh
+        os.replace(tmp, path)
+    except BaseException:
+        if os.path.exists(tmp):
+            os.unlink(tmp)
+        raise
+    os.chmod(path, 0o644)
+
+
+def write_json(path: str, obj) -> None:
+    with atomic_writer(path) as fh:
+        json.dump(obj, fh, indent=1)
+        fh.write("\n")

scripts/rechunk-kld-corpus.py

@@ -38,12 +38,19 @@ one-artifact-one-name law run-qwen38-bf16.sh applies to a score report. The file
 is written through a temporary name and renamed, so an interrupted run leaves no
 half corpus behind. A JSON summary of what was written goes to stdout.
 """
-
 import argparse
+import dataclasses
 import json
 import os
 import sys
-import tempfile
+
+_HERE = os.path.dirname(os.path.abspath(__file__)) if "__file__" in globals() else os.getcwd()
+if _HERE not in sys.path:
+    sys.path.insert(0, _HERE)
+
+import qwen38_common as qc
+
+MISSING = object()  # "the parent record carried no source field"
 
 
 def parse_args() -> argparse.Namespace:
@@ -69,95 +76,96 @@ def parse_args() -> argparse.Namespace:
     return ap.parse_args()
 
 
-def load_tokenizer(path: str):
-    try:
-        from tokenizers import Tokenizer
-    except ImportError as exc:  # the corpus scripts carry their own venv
-        raise SystemExit(
-            f"REFUSED: tokenizers is not importable by {sys.executable} — run the "
-            f"corpus venv's python (on dankserver: /home/dank/venvs/radcorpus/bin/python) "
-            f"or set FRANK_QWEN38_TOKENIZER: {exc}") from None
-    if not os.path.isfile(path):
-        raise SystemExit(f"REFUSED: no tokenizer at {path}")
-    return Tokenizer.from_file(path)
+@dataclasses.dataclass(frozen=True)
+class Window:
+    """One scoring document cut out of a parent record."""
+    prompt: str
+    score_from: int
+    n_tokens: int
+    clamped: bool
+    source: object = MISSING
+
+    def record(self) -> dict:
+        rec = {"prompt": self.prompt, "score_from": self.score_from}
+        if self.source is not MISSING:
+            rec["source"] = self.source
+        return rec
 
 
-def refuse(msg: str) -> None:
-    print(f"REFUSED: {msg}", file=sys.stderr)
-    sys.exit(2)  # 2 is the code argparse uses for unusable arguments
-
-
-def main() -> None:
-    a = parse_args()
+def validate(a: argparse.Namespace) -> None:
+    """Everything that makes this run unusable, named before a token is cut."""
     if a.max_doc_tokens <= 0:
-        refuse(f"--max-doc-tokens must be positive (got {a.max_doc_tokens})")
+        qc.refuse(f"--max-doc-tokens must be positive (got {a.max_doc_tokens})")
     if a.score_tail <= 0:
-        refuse(f"--score-tail must be positive (got {a.score_tail})")
+        qc.refuse(f"--score-tail must be positive (got {a.score_tail})")
     if a.max_doc_tokens <= a.score_tail:
-        refuse(f"--max-doc-tokens {a.max_doc_tokens} cannot hold a {a.score_tail}-position "
-               "score tail: every window would be all tail and no context")
+        qc.refuse(f"--max-doc-tokens {a.max_doc_tokens} cannot hold a "
+                  f"{a.score_tail}-position score tail: every window would be all "
+                  "tail and no context")
     if not os.path.isfile(a.corpus):
-        refuse(f"no corpus at {a.corpus}")
+        qc.refuse(f"no corpus at {a.corpus}")
     if os.path.realpath(a.out) == os.path.realpath(a.corpus):
-        refuse(f"--out {a.out} is the corpus itself; move it to trash first — the trash law "
-               "means the old segmentation stays recoverable")
+        qc.refuse(f"--out {a.out} is the corpus itself; move it to trash first — the "
+                  "trash law means the old segmentation stays recoverable")
     if os.path.exists(a.out) and os.path.getsize(a.out) > 0 and not a.force:
-        refuse(f"{a.out} exists and is not empty; pass --force after moving it to trash")
+        qc.refuse(f"{a.out} exists and is not empty; pass --force after moving it to trash")
 
-    tok = load_tokenizer(a.tokenizer_json)
 
+def cut(rec: dict, tok, a: argparse.Namespace,
+        where: str) -> tuple[list[Window], int]:
+    """Cut one record's token ids into windows, proving each one round-trips.
+
+    The round-trip encode is the whole reason this tool may claim the reference
+    scores the corpus's own token sequence: decoding a window to text and
+    encoding it back must return the identical ids, or the score is of a
+    different document than the one on disk.
+    """
+    ids = tok.encode(rec["prompt"], add_special_tokens=False).ids
+    windows: list[Window] = []
+    for start in range(0, max(len(ids), 1), a.max_doc_tokens):
+        window = ids[start:start + a.max_doc_tokens]
+        if not window:
+            continue
+        text = tok.decode(window)
+        if tok.encode(text, add_special_tokens=False).ids != window:
+            qc.refuse(f"{where} window at token {start} does not round-trip through "
+                      "text — the reference would score a different token sequence "
+                      "than the corpus holds")
+        windows.append(Window(prompt=text,
+                              score_from=max(len(window) - a.score_tail, 0),
+                              n_tokens=len(window),
+                              clamped=len(window) < a.score_tail,
+                              source=rec["source"] if "source" in rec else MISSING))
+    return windows, len(ids)
+
+
+def rechunk(a: argparse.Namespace, tok) -> dict:
+    """Write the re-chunked corpus and return the summary it printed."""
     docs_in = docs_out = tokens_in = tokens_out = 0
-    longest_child = 0
-    clamped = scored = 0
+    longest_child = clamped = scored = 0
     sources: dict[str, int] = {}
-    fd, tmp = tempfile.mkstemp(dir=os.path.dirname(os.path.abspath(a.out)) or ".",
-                              prefix=".rechunk-")
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as out:
-            with open(a.corpus, encoding="utf-8") as src:
-                for lineno, line in enumerate(src, 1):
-                    if not line.strip():
-                        continue
-                    try:
-                        rec = json.loads(line)
-                    except json.JSONDecodeError as exc:
-                        refuse(f"{a.corpus}:{lineno} is not JSON: {exc}")
-                    prompt = rec.get("prompt")
-                    if not isinstance(prompt, str):
-                        refuse(f"{a.corpus}:{lineno} has no string 'prompt' field")
-                    ids = tok.encode(prompt, add_special_tokens=False).ids
-                    docs_in += 1
-                    tokens_in += len(ids)
-                    for start in range(0, max(len(ids), 1), a.max_doc_tokens):
-                        window = ids[start:start + a.max_doc_tokens]
-                        if not window:
-                            continue
-                        score_from = max(len(window) - a.score_tail, 0)
-                        if len(window) < a.score_tail:
-                            clamped += 1
-                        scored += len(window) - score_from
-                        child = {"prompt": tok.decode(window),
-                                 "score_from": score_from}
-                        if "source" in rec:
-                            child["source"] = rec["source"]
-                        out.write(json.dumps(child, ensure_ascii=False) + "\n")
-                        docs_out += 1
-                        tokens_out += len(window)
-                        longest_child = max(longest_child, len(window))
-                        if "source" in rec:
-                            for cls in str(rec["source"]).split("+"):
-                                sources[cls] = sources.get(cls, 0) + 1
-        if tokens_out != tokens_in:
-            raise SystemExit(f"REFUSED: the windows hold {tokens_out} tokens and the corpus "
-                             f"has {tokens_in} — re-chunking lost or invented tokens")
-        os.replace(tmp, a.out)
-    except BaseException:
-        if os.path.exists(tmp):
-            os.unlink(tmp)
-        raise
-    os.chmod(a.out, 0o644)
-
-    print(json.dumps({
+    with qc.atomic_writer(a.out) as out:
+        for lineno, rec in qc.iter_jsonl(a.corpus):
+            where = f"{a.corpus}:{lineno}"
+            if not isinstance(rec.get("prompt"), str):
+                qc.refuse(f"{where} has no string 'prompt' field")
+            windows, n_parent = cut(rec, tok, a, where)
+            docs_in += 1
+            tokens_in += n_parent
+            for w in windows:
+                out.write(json.dumps(w.record(), ensure_ascii=False) + "\n")
+                docs_out += 1
+                tokens_out += w.n_tokens
+                longest_child = max(longest_child, w.n_tokens)
+                clamped += 1 if w.clamped else 0
+                scored += w.n_tokens - w.score_from
+                if "source" in rec:
+                    for cls in str(rec["source"]).split("+"):
+                        sources[cls] = sources.get(cls, 0) + 1
+    if tokens_out != tokens_in:
+        raise SystemExit(f"REFUSED: the windows hold {tokens_out} tokens and the corpus "
+                         f"has {tokens_in} — re-chunking lost or invented tokens")
+    return {
         "corpus": a.corpus,
         "out": a.out,
         "tokenizer": os.path.basename(a.tokenizer_json),
@@ -171,7 +179,13 @@ def main() -> None:
         "scored_positions": scored,
         "clamped_score_from": clamped,
         "source_documents": sources,
-    }, indent=1))
+    }
+
+
+def main() -> None:
+    a = parse_args()
+    validate(a)
+    print(json.dumps(rechunk(a, qc.load_tokenizer(a.tokenizer_json)), indent=1))
 
 
 if __name__ == "__main__":

scripts/run-qwen38-bf16.sh

@@ -120,7 +120,8 @@ while [ $# -gt 0 ]; do
     case "$1" in
         kldref|calib|score) MODE="$1" ;;
         --dry-run)          DRY=1 ;;
-        --image)            IMAGE="$2"; shift ;;
+        --image)            [ "$#" -ge 2 ] || { echo "--image needs a reference" >&2; exit 2; }
+                            IMAGE="$2"; shift ;;
         *)
             # score's two positional arguments, and nothing else may be bare.
             if [ "$MODE" = score ] && [ "${#POS[@]}" -lt 2 ]; then
@@ -250,7 +251,8 @@ case "$MODE" in
     # time the engine reads a recording and writes a report instead of the
     # reverse. The reference dir mounts READ-ONLY — a scoring pass must not be
     # able to touch the zero point — and the report gets its own writable bind.
-    if [ "$OUT_DIR" = "$KLD_REF_DIR" ] || same_dir "$OUT_DIR" "$KLD_REF_DIR"; then
+    if [ "$OUT_DIR" = "$KLD_REF_DIR" ] || same_dir "$OUT_DIR" "$KLD_REF_DIR" \
+       || [ "$(realpath -m "$OUT_DIR")" = "$(realpath -m "$KLD_REF_DIR")" ]; then
         echo "REFUSED: score output $SCORE_OUT is inside the reference dir $KLD_REF_DIR, which this mode mounts read-only" >&2
         exit 2
     fi
@@ -276,9 +278,17 @@ case "$MODE" in
     # through MAX_SEQS, exactly as kldref and score do.
     ARGS+=(--model "$MODEL_IN" --port 7770
            --max-num-seqs "$MAX_SEQS" --max-num-batched-tokens "$MNBT")
+    # The tap is only real if this directory exists and the engine can write it:
+    # without the check a typo'd CALIB_DIR auto-creates a root-owned empty dir in
+    # whatever path was given, and the run burns hours serving prompts into a tap
+    # nobody reads. RADIANCE_CALIB_TOKENS is the whole corpus' floor, not a round's.
+    [ "$DRY" = 1 ] || [ -w "$CALIB_DIR" ] || {
+        echo "REFUSED: calib tap dir $CALIB_DIR is not writable (mkdir it, chown it, then re-run)" >&2
+        exit 2
+    }
     MOUNTS+=(-v "$CALIB_DIR:/data/calib")
     TAPENV=(-e RADIANCE_CALIB_DIR=/data/calib
-            -e RADIANCE_CALIB_TOKENS=12000000)
+            -e "RADIANCE_CALIB_TOKENS=${FRANK_QWEN38_CALIB_TOKENS:-12000000}")
     ;;
 esac
 
@@ -352,7 +362,14 @@ cp -f "$TEMPLATE" "$TPL" || {
 [ "$MODE" = kldref ] && mkdir -p "$KLD_REF_DIR"
 [ "$MODE" = calib ]  && mkdir -p "$CALIB_DIR"
 
-restore() { sudo systemctl start "$SERVICE" >/dev/null 2>&1 || true; }
+# Stop the campaign container BEFORE handing the cards back: a killed run leaves
+# its conmon alive (--rm only fires when podman itself exits), and production
+# then boots beside a container still holding the GPUs — measured 2026-10-29,
+# where the service reported `active` while its engine starved on the rig's VRAM.
+restore() {
+    podman rm -f "$C" >/dev/null 2>&1 || true
+    sudo systemctl start "$SERVICE" >/dev/null 2>&1 || true
+}
 trap restore EXIT
 echo "run-qwen38-bf16 $MODE: stopping $SERVICE ($(date -Is)); log: $LOG"
 sudo systemctl stop "$SERVICE" >/dev/null 2>&1 || true

scripts/stack-qwen38-official-fp8.py

@@ -337,8 +337,12 @@ def main():
     if any(meta[shard_ids[i]][4][1] != hd for i in range(n_shards)):
         log("ERROR: PLE shards disagree on HD")
         return 2
-    if hd % 128 != 0 and hd != 160:
-        log(f"WARNING: PLE HD={hd} is neither 160 nor a multiple of 128")
+    if hd != 160 and hd % 128 != 0:
+        # The pack below writes lanes of 128 codes; any other HD would produce a
+        # table that decodes to the wrong rows. The old warning let it continue.
+        log(f"ERROR: PLE HD={hd} is neither 160 nor a multiple of 128, and the "
+            "128-lane pack cannot express it")
+        return 2
 
     # ---- idempotence gate -------------------------------------------------
     stacked_names = [f"{p}.mlp.experts.gate_up_proj" for p in prefixes]
@@ -521,12 +525,16 @@ def verify(stage, prefixes, n_e, shapes, meta, shard_ids, rows, hd,
         }
         for name, (dt, shp) in want.items():
             e = hdr[name]
-            assert e["dtype"] == dt and list(e["shape"]) == shp, name
-            assert e["data_offsets"][1] - e["data_offsets"][0] == \
-                int(np.prod(shp)) * (1 if dt == F8 else 2), name
+            if e["dtype"] != dt or list(e["shape"]) != shp:
+                raise RuntimeError(f"{name}: header is {e['dtype']}{e['shape']}, expected {dt}{shp}")
+            if e["data_offsets"][1] - e["data_offsets"][0] != int(np.prod(shp)) * (1 if dt == F8 else 2):
+                raise RuntimeError(f"{name}: data span is {e['data_offsets']}, shape {shp} needs "
+                                   f"{int(np.prod(shp)) * (1 if dt == F8 else 2)} B")
     _, phdr = parse_header(out_path)
     e = phdr[next(iter(phdr))]
-    assert e["dtype"] == BF16 and list(e["shape"]) == [sum(rows), hd]
+    if e["dtype"] != BF16 or list(e["shape"]) != [sum(rows), hd]:
+        raise RuntimeError(f"PLE table header is {e['dtype']}{e['shape']}, "
+                           f"expected {BF16}[{sum(rows)}, {hd}]")
 
     # sampled PLE rows: recompute and compare bytes exactly
     lut = e4m3_lut()
@@ -546,7 +554,8 @@ def verify(stage, prefixes, n_e, shapes, meta, shard_ids, rows, hd,
                 with open(out_path, "rb") as ofh:
                     ofh.seek(out_off + got_row * hd * 2)
                     got = ofh.read(hd * 2)
-                assert got == want, f"PLE row {got_row} mismatch"
+                if got != want:
+                    raise RuntimeError(f"PLE row {got_row} does not match the decoded source")
             mm.close()
     log("verify: new-file headers, sizes and sampled PLE rows all exact; "
         "expert slabs were byte-checked in-stack")

serverAppConfigs-docs/LLM/frank-archon.md

@@ -641,7 +641,10 @@ vllm-radiance era (this table opens on 09-26; the era page claims 10-10).
 - Convert campaign (dormant, kept for a future quant): `scripts/convert-swift15-w4dn8.sh`
   `scripts/recipes/qwen4exp-swift15-w4dn8-i8-hc8m.recipe`, `scripts/run-swift15-bf16.sh`
   `scripts/kld-gate-swift15.sh`, `scripts/kld-then-calib.sh`, `scripts/calib-push.py`
-  `scripts/make-calib-corpus.py`, `scripts/derive-protected-experts.py`
+  `scripts/make-calib-corpus.py`, `scripts/rechunk-kld-corpus.py`,
+  `scripts/qwen38_common.py` (the tokenizer, JSONL and refusal plumbing those three
+  corpus tools share — one `n_tokens`, so every tool counts the same way),
+  `scripts/derive-protected-experts.py`
   `scripts/apply-protected-cast.sh`, `scripts/verify-swift15-ple.py`.
 - Probes: `scripts/probe-frank-archon-kv.py` (reads slots/window/dtype from the deployed env so
   it cannot drift; leg D = decode), `scripts/frank-archon-expert-activity.py`