From 1e4fa14772ef0307fac20c1ce14d06b809753fa7 Mon Sep 17 00:00:00 2001 From: Frank Date: Thu, 29 Oct 2026 15:14:43 -0700 Subject: [PATCH] fix(quant): prove window token identity, gate the tap on progress, kill the runner's orphan race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review of the campaign pipeline. Correctness: rechunk-kld-corpus.py wrote each window as text and asserted only lengths, so a KLD reference could score a different token sequence than the corpus holds — every window is now re-encoded and the tool refuses on mismatch; make-calib-corpus.py no longer crashes on a .json that is not a content-keyed JSONL, and its KLD cursor advanced by 2 per fragment (half the pool unused). Silent waste: calib-push.py now aborts when the tap sealed nothing or no prompt was accepted, instead of re-pushing 10M tokens per round; div-by-zero gone, served model id parameterised. Races: run-qwen38-bf16.sh restore() removes the rig container before starting production (today's orphan held the GPUs while the service reported active), the calib tap dir is proven writable, the score ro-guard resolves paths; convert-qwen38-w5nl64.sh refuses a missing --arm value, an existing qwen38-convert- container, and a zero-exit convert that wrote no artifact. stack-qwen38-official-fp8.py: bare asserts in verify() (skipped under -O) and a non-128-lane PLE HD that only warned are hard failures. Speed: fragment token counts computed once, KLD scored positions accumulated instead of a second full re-encode. Tests: 576 ok / 0 fail. --- scripts/calib-push.py | 52 ++++++++++++++++++++++++---- scripts/convert-qwen38-w5nl64.sh | 14 ++++++-- scripts/make-calib-corpus.py | 38 ++++++++++++++------ scripts/rechunk-kld-corpus.py | 7 +++- scripts/run-qwen38-bf16.sh | 25 ++++++++++--- scripts/stack-qwen38-official-fp8.py | 23 ++++++++---- 6 files changed, 127 insertions(+), 32 deletions(-) diff --git a/scripts/calib-push.py b/scripts/calib-push.py index 562ffca..9a4114d 100755 --- a/scripts/calib-push.py +++ b/scripts/calib-push.py @@ -29,6 +29,7 @@ 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 @@ -49,9 +50,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,11 +69,33 @@ def push_one(url, prompt): return False +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 main(): 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() @@ -84,22 +107,37 @@ def main(): line = line.strip() if line: prompts.append(json.loads(line)["prompt"]) - print(f"corpus: {len(prompts)} prompts", flush=True) + chars = sum(len(p) for p in prompts) + print(f"corpus: {len(prompts)} prompts, ~{chars // 4} tokens (chars/4)", + flush=True) + if not prompts: + print("REFUSED: the corpus holds no prompts", flush=True) + sys.exit(2) 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): + for good in ex.map(lambda p: push_one(url, p, a.model), prompts): ok += 1 if good else 0 - dt = time.time() - t0 + 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) diff --git a/scripts/convert-qwen38-w5nl64.sh b/scripts/convert-qwen38-w5nl64.sh index e95813e..58cb384 100755 --- a/scripts/convert-qwen38-w5nl64.sh +++ b/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. diff --git a/scripts/make-calib-corpus.py b/scripts/make-calib-corpus.py index ddad598..f5bacfd 100755 --- a/scripts/make-calib-corpus.py +++ b/scripts/make-calib-corpus.py @@ -190,8 +190,20 @@ def main() -> None: 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] + # 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 crashed 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"]) raw = "\n".join(lines) toks = tokenizer.encode(raw, add_special_tokens=False).ids return tokenizer.decode(toks[:2400], skip_special_tokens=True) @@ -233,10 +245,15 @@ def main() -> None: if not pool: continue got, i, buf, nbuf = 0, 0, [], 0 + # Count every fragment once: the loop cycles the pool up to 40 times, + # and re-tokenizing a 2.4k-token fragment on each pass dominated the + # build's wall time. + counts = [toks(f) for f in pool] while got < want: - frag = pool[i % len(pool)] + j = i % len(pool) + frag = pool[j] i += 1 - n = toks(frag) + n = counts[j] if nbuf + n > a.chunk_tokens and buf: rows.append({"prompt": "\n".join(buf), "cls": cls}) got += nbuf @@ -255,8 +272,8 @@ def main() -> None: return len(rows), [{k: v for k, v in r.items() if k == "prompt"} for r in rows] - def emit_kld() -> tuple[int, list[dict]]: - rows, total = [], 0 + def emit_kld() -> tuple[int, list[dict], int]: + rows, total, scored_total = [], 0, 0 classes = ["code", "prose", "tools"] i = {c: 0 for c in classes} guard = 0 @@ -269,7 +286,7 @@ def main() -> None: if not pool: continue frag = pool[i[cls] % len(pool)] - i[cls] += 2 + i[cls] += 1 msgs.append({"role": "user", "content": "Summarize and explain this " f"engineering material:\n{frag}"}) @@ -285,19 +302,18 @@ def main() -> None: "score_from": n - a.score_tail, "source": "+".join(sorted(set(srcs)))}) total += n - return len(rows), rows + scored_total += a.score_tail + return len(rows), rows, scored_total 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() + nk, krows, scored = 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 = { "seed": a.seed, "template": tname, "tokenizer": os.path.basename(a.tokenizer_json), diff --git a/scripts/rechunk-kld-corpus.py b/scripts/rechunk-kld-corpus.py index bfefad0..e8b69d1 100755 --- a/scripts/rechunk-kld-corpus.py +++ b/scripts/rechunk-kld-corpus.py @@ -136,7 +136,12 @@ def main() -> None: if len(window) < a.score_tail: clamped += 1 scored += len(window) - score_from - child = {"prompt": tok.decode(window), + text = tok.decode(window) + if tok.encode(text, add_special_tokens=False).ids != window: + refuse(f"{a.corpus}:{lineno} window at token {start} does not " + "round-trip through text — the reference would score a " + "different token sequence than the corpus holds") + child = {"prompt": text, "score_from": score_from} if "source" in rec: child["source"] = rec["source"] diff --git a/scripts/run-qwen38-bf16.sh b/scripts/run-qwen38-bf16.sh index 1d10a20..fe358b6 100755 --- a/scripts/run-qwen38-bf16.sh +++ b/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 diff --git a/scripts/stack-qwen38-official-fp8.py b/scripts/stack-qwen38-official-fp8.py index 106e983..968a0a5 100755 --- a/scripts/stack-qwen38-official-fp8.py +++ b/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")