fix(quant): harden the campaign pipeline, then refactor its corpus tools onto one module #83

Merged
dank merged 2 commits from fix/quant-pipeline-hardening into main 2026-10-29 15:29:34 -07:00
6 changed files with 127 additions and 32 deletions
Showing only changes of commit 1e4fa14772 - Show all commits

No files matched your search

+45 -7
View File
@@ -29,6 +29,7 @@ Usage: calib-push.py [--port 7770] [--parallel 64] [--container arbench]
import argparse import argparse
import concurrent.futures import concurrent.futures
import json import json
import os
import subprocess import subprocess
import sys import sys
import time import time
@@ -49,9 +50,9 @@ def tap_done(container):
return DONE_MARK in (out.stdout + out.stderr) return DONE_MARK in (out.stdout + out.stderr)
def push_one(url, prompt): def push_one(url, prompt, model):
body = json.dumps({ body = json.dumps({
"model": "frank-archon", "model": model,
"prompt": prompt, "prompt": prompt,
"max_tokens": 1, "max_tokens": 1,
"temperature": 0, "temperature": 0,
@@ -68,11 +69,33 @@ def push_one(url, prompt):
return False 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(): def main():
ap = argparse.ArgumentParser() ap = argparse.ArgumentParser()
ap.add_argument("--port", type=int, default=7770) ap.add_argument("--port", type=int, default=7770)
ap.add_argument("--parallel", type=int, default=64) ap.add_argument("--parallel", type=int, default=64)
ap.add_argument("--container", default="arbench") 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("--corpus", default="/home/dank/corpora/out/calib.jsonl")
ap.add_argument("--max-rounds", type=int, default=4) ap.add_argument("--max-rounds", type=int, default=4)
a = ap.parse_args() a = ap.parse_args()
@@ -84,22 +107,37 @@ def main():
line = line.strip() line = line.strip()
if line: if line:
prompts.append(json.loads(line)["prompt"]) 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): for rnd in range(1, a.max_rounds + 1):
if tap_done(a.container): if tap_done(a.container):
break break
before = tap_bytes(a.tap_dir)
t0 = time.time() t0 = time.time()
ok = 0 ok = 0
with concurrent.futures.ThreadPoolExecutor(a.parallel) as ex: 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 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 " 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)", f"({chars / 4 / dt:.0f} tokens/s est)", flush=True)
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): if tap_done(a.container):
break 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: else:
print("max rounds exhausted, tap not done", flush=True) print("max rounds exhausted, tap not done", flush=True)
sys.exit(1) sys.exit(1)
+12 -2
View File
@@ -71,10 +71,12 @@ PLAN_ONLY=0
KEEP_RECIPE=0 KEEP_RECIPE=0
while [ "$#" -gt 0 ]; do while [ "$#" -gt 0 ]; do
case "$1" in 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 ;; --dry-run) DRY=1; shift ;;
--plan-only) PLAN_ONLY=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 ;; --keep-recipe) KEEP_RECIPE=1; shift ;;
*) echo "unknown argument: $1" >&2; exit 2 ;; *) echo "unknown argument: $1" >&2; exit 2 ;;
esac esac
@@ -124,6 +126,13 @@ fi
for a in "${ARMS[@]}"; do for a in "${ARMS[@]}"; do
[ -e "$OUTDIR/qwen38-w5nl64-$a.rad" ] && { [ -e "$OUTDIR/qwen38-w5nl64-$a.rad" ] && {
echo "$OUTDIR/qwen38-w5nl64-$a.rad already exists; move it to trash first" >&2; exit 2; } 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 done
# --- the protected plane, spliced into a per-run copy of the template -------------------------- # --- 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=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; } [ "$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" 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 )) got_mib=$(( $(stat -c %s "$out") / 1048576 ))
# The receipt: the artifact is what the plan predicted, within the metadata the plan does not # 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. # count. An arm that came out materially smaller lost a plane somewhere.
+27 -11
View File
@@ -190,8 +190,20 @@ def main() -> None:
path = os.path.join(a.repo, p) path = os.path.join(a.repo, p)
raw = open(path, encoding="utf-8", errors="replace").read() raw = open(path, encoding="utf-8", errors="replace").read()
if p.endswith(JSONL_EXTS): if p.endswith(JSONL_EXTS):
lines = [json.loads(l)["content"] for l in raw.splitlines() # A .json in this repo is not necessarily a content-keyed JSONL: the
if l.strip() and "content" in l] # 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) raw = "\n".join(lines)
toks = tokenizer.encode(raw, add_special_tokens=False).ids toks = tokenizer.encode(raw, add_special_tokens=False).ids
return tokenizer.decode(toks[:2400], skip_special_tokens=True) return tokenizer.decode(toks[:2400], skip_special_tokens=True)
@@ -233,10 +245,15 @@ def main() -> None:
if not pool: if not pool:
continue continue
got, i, buf, nbuf = 0, 0, [], 0 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: while got < want:
frag = pool[i % len(pool)] j = i % len(pool)
frag = pool[j]
i += 1 i += 1
n = toks(frag) n = counts[j]
if nbuf + n > a.chunk_tokens and buf: if nbuf + n > a.chunk_tokens and buf:
rows.append({"prompt": "\n".join(buf), "cls": cls}) rows.append({"prompt": "\n".join(buf), "cls": cls})
got += nbuf got += nbuf
@@ -255,8 +272,8 @@ def main() -> None:
return len(rows), [{k: v for k, v in r.items() if k == "prompt"} return len(rows), [{k: v for k, v in r.items() if k == "prompt"}
for r in rows] for r in rows]
def emit_kld() -> tuple[int, list[dict]]: def emit_kld() -> tuple[int, list[dict], int]:
rows, total = [], 0 rows, total, scored_total = [], 0, 0
classes = ["code", "prose", "tools"] classes = ["code", "prose", "tools"]
i = {c: 0 for c in classes} i = {c: 0 for c in classes}
guard = 0 guard = 0
@@ -269,7 +286,7 @@ def main() -> None:
if not pool: if not pool:
continue continue
frag = pool[i[cls] % len(pool)] frag = pool[i[cls] % len(pool)]
i[cls] += 2 i[cls] += 1
msgs.append({"role": "user", msgs.append({"role": "user",
"content": "Summarize and explain this " "content": "Summarize and explain this "
f"engineering material:\n{frag}"}) f"engineering material:\n{frag}"})
@@ -285,19 +302,18 @@ def main() -> None:
"score_from": n - a.score_tail, "score_from": n - a.score_tail,
"source": "+".join(sorted(set(srcs)))}) "source": "+".join(sorted(set(srcs)))})
total += n 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) os.makedirs(a.out_dir, exist_ok=True)
nc, rows = emit_calib() nc, rows = emit_calib()
with open(os.path.join(a.out_dir, "calib.jsonl"), "w") as f: with open(os.path.join(a.out_dir, "calib.jsonl"), "w") as f:
for r in rows: for r in rows:
f.write(json.dumps(r) + "\n") 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: with open(os.path.join(a.out_dir, "kld.jsonl"), "w") as f:
for r in krows: for r in krows:
f.write(json.dumps(r) + "\n") 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 = { manifest = {
"seed": a.seed, "template": tname, "seed": a.seed, "template": tname,
"tokenizer": os.path.basename(a.tokenizer_json), "tokenizer": os.path.basename(a.tokenizer_json),
+6 -1
View File
@@ -136,7 +136,12 @@ def main() -> None:
if len(window) < a.score_tail: if len(window) < a.score_tail:
clamped += 1 clamped += 1
scored += len(window) - score_from 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} "score_from": score_from}
if "source" in rec: if "source" in rec:
child["source"] = rec["source"] child["source"] = rec["source"]
+21 -4
View File
@@ -120,7 +120,8 @@ while [ $# -gt 0 ]; do
case "$1" in case "$1" in
kldref|calib|score) MODE="$1" ;; kldref|calib|score) MODE="$1" ;;
--dry-run) DRY=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. # score's two positional arguments, and nothing else may be bare.
if [ "$MODE" = score ] && [ "${#POS[@]}" -lt 2 ]; then 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 # 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 # 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. # 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 echo "REFUSED: score output $SCORE_OUT is inside the reference dir $KLD_REF_DIR, which this mode mounts read-only" >&2
exit 2 exit 2
fi fi
@@ -276,9 +278,17 @@ case "$MODE" in
# through MAX_SEQS, exactly as kldref and score do. # through MAX_SEQS, exactly as kldref and score do.
ARGS+=(--model "$MODEL_IN" --port 7770 ARGS+=(--model "$MODEL_IN" --port 7770
--max-num-seqs "$MAX_SEQS" --max-num-batched-tokens "$MNBT") --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") MOUNTS+=(-v "$CALIB_DIR:/data/calib")
TAPENV=(-e RADIANCE_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 esac
@@ -352,7 +362,14 @@ cp -f "$TEMPLATE" "$TPL" || {
[ "$MODE" = kldref ] && mkdir -p "$KLD_REF_DIR" [ "$MODE" = kldref ] && mkdir -p "$KLD_REF_DIR"
[ "$MODE" = calib ] && mkdir -p "$CALIB_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 trap restore EXIT
echo "run-qwen38-bf16 $MODE: stopping $SERVICE ($(date -Is)); log: $LOG" echo "run-qwen38-bf16 $MODE: stopping $SERVICE ($(date -Is)); log: $LOG"
sudo systemctl stop "$SERVICE" >/dev/null 2>&1 || true sudo systemctl stop "$SERVICE" >/dev/null 2>&1 || true
+16 -7
View File
@@ -337,8 +337,12 @@ def main():
if any(meta[shard_ids[i]][4][1] != hd for i in range(n_shards)): if any(meta[shard_ids[i]][4][1] != hd for i in range(n_shards)):
log("ERROR: PLE shards disagree on HD") log("ERROR: PLE shards disagree on HD")
return 2 return 2
if hd % 128 != 0 and hd != 160: if hd != 160 and hd % 128 != 0:
log(f"WARNING: PLE HD={hd} is neither 160 nor a multiple of 128") # 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 ------------------------------------------------- # ---- idempotence gate -------------------------------------------------
stacked_names = [f"{p}.mlp.experts.gate_up_proj" for p in prefixes] 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(): for name, (dt, shp) in want.items():
e = hdr[name] e = hdr[name]
assert e["dtype"] == dt and list(e["shape"]) == shp, name if e["dtype"] != dt or list(e["shape"]) != shp:
assert e["data_offsets"][1] - e["data_offsets"][0] == \ raise RuntimeError(f"{name}: header is {e['dtype']}{e['shape']}, expected {dt}{shp}")
int(np.prod(shp)) * (1 if dt == F8 else 2), name 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) _, phdr = parse_header(out_path)
e = phdr[next(iter(phdr))] 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 # sampled PLE rows: recompute and compare bytes exactly
lut = e4m3_lut() 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: with open(out_path, "rb") as ofh:
ofh.seek(out_off + got_row * hd * 2) ofh.seek(out_off + got_row * hd * 2)
got = ofh.read(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() mm.close()
log("verify: new-file headers, sizes and sampled PLE rows all exact; " log("verify: new-file headers, sizes and sampled PLE rows all exact; "
"expert slabs were byte-checked in-stack") "expert slabs were byte-checked in-stack")