Compare commits

..
Author SHA1 Message Date
Frank c238e32d6f fix(quant): a tap is only a tap when its rig is running
Four gates, all bought with the 2026-10-29 incident: the rig's calibration boot died of an OOM,
the chain's health loop expired without failing, and the pusher fed four rounds of 10M tokens
(86 minutes of prefill) into RESTORED PRODUCTION, recording nothing.

- run-qwen38-bf16.sh: --ngram-placement becomes a knob whose calib default is `disk` (the 95.37
  GiB table pin is what starved that boot: 230.22 GiB pinned of 246, 8.06 GiB available) while
  kldref and score keep `ram`. calib writes .rig-boot into the tap dir and retires it on a
  non-zero exit, so a dead boot cannot leave the tap looking rig-authorized.
- calib-push.py: refuses before the first prompt unless the marker exists, names this container,
  and that container is running. With the driver's existing tap-progress gate, a push can no
  longer spend hours feeding an endpoint that is not the rig.
- qwen38-calib-window.sh: the calibration window as a gated chain — stop on a failed phase
  instead of falling through the health wait, bank the 336 GiB reference container into trash
  only when both phases exit 0 and the tap holds grams, and print the whole plan on --dry-run.
- kld-gate-qwen38.sh: two-report mode (candidate vs the served payload) for the disk where the
  172.8 GiB FP8 anchor cannot live, plus the laws the report always carried and the gate never
  read: a p99.9 tail ceiling and every by_source slice passing mean, tail and top-1.
- frank-archon.md: the pipeline laws (marker, n-gram tier, three-metric acceptance) and a Files
  section that names the live campaign instead of the retired swift15 era.

Tests: 587/0 — 13 new behavioral cases (dry-run argv, marker refusals, two-report gate math).
2026-10-29 16:14:16 -07:00
Frank 3e5a37a21f 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.
2026-10-29 15:26:08 -07:00
Frank 1e4fa14772 fix(quant): prove window token identity, gate the tap on progress, kill the runner's orphan race
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-<arm> 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.
2026-10-29 15:14:43 -07:00
11 changed files with 962 additions and 224 deletions

No files matched your search

+121 -18
View File
@@ -22,6 +22,14 @@ Mechanics notes:
prompts do not bias the Hessians (the corpus is drawn round-robin over 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; 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 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.
- before any prompt, the tap must carry the rig's own marker
(`--tap-marker`, written by run-qwen38-bf16.sh calib) naming THIS
container, and that container must be running. Measured 2026-10-29:
with neither check, four rounds of 10M tokens went into restored
production after the rig's boot had died, and recorded nothing.
Usage: calib-push.py [--port 7770] [--parallel 64] [--container arbench] Usage: calib-push.py [--port 7770] [--parallel 64] [--container arbench]
[--corpus /home/dank/corpora/out/calib.jsonl] [--max-rounds 4] [--corpus /home/dank/corpora/out/calib.jsonl] [--max-rounds 4]
@@ -29,11 +37,18 @@ 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
import urllib.request 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" DONE_MARK = "calibration wrote"
@@ -49,9 +64,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,38 +83,126 @@ def push_one(url, prompt):
return False 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 require_rig(a):
"""Refuse unless the rig that opened this tap is up and serving right now.
Three facts, in order: the runner wrote its marker into the tap dir, the
marker names the container this pusher targets, and that container is
running. A marker without a live container is a boot that died; a live
container without a marker is some other engine on the port — on 2026-10-29
it was restored production, which took four rounds of 10M tokens before the
old driver admitted it had recorded nothing.
"""
marker = os.path.join(a.tap_dir, a.tap_marker)
if not os.path.isfile(marker):
qc.refuse(f"no rig marker at {marker}: this tap was not opened by "
"run-qwen38-bf16.sh calib, so a push here would feed whatever "
"answers the port and record nothing")
try:
lines = open(marker, encoding="utf-8", errors="replace").read().splitlines()
except OSError as exc:
qc.refuse(f"cannot read the rig marker {marker}: {exc}")
meta = dict(line.split("=", 1) for line in lines if "=" in line)
who = meta.get("container", "")
if who != a.container:
qc.refuse(f"the rig marker names container {who!r} and this pusher targets "
f"{a.container!r} — wrong tap, or a stale marker")
try:
p = subprocess.run(["podman", "inspect", "-f", "{{.State.Running}}", a.container],
capture_output=True, text=True, timeout=30)
except Exception as exc:
qc.refuse(f"cannot inspect the rig container {a.container}: {exc}")
if p.returncode != 0 or p.stdout.strip() != "true":
qc.refuse(f"the rig container {a.container} is not running — whatever is "
"answering the port, it is not the calibration rig")
def parse_args():
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("--tap-marker", default=".rig-boot",
help="the file run-qwen38-bf16.sh calib writes into the tap dir; without it the pusher refuses, because an unmarked tap is not a rig")
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() return ap.parse_args()
url = f"http://127.0.0.1:{a.port}/v1/completions"
def load_prompts(a) -> list[str]:
if not os.path.isfile(a.corpus):
qc.refuse(f"no corpus at {a.corpus}")
prompts = [] prompts = []
with open(a.corpus) as f: for lineno, rec in qc.iter_jsonl(a.corpus):
for line in f: if not isinstance(rec.get("prompt"), str):
line = line.strip() qc.refuse(f"{a.corpus}:{lineno} has no string 'prompt' field")
if line: prompts.append(rec["prompt"])
prompts.append(json.loads(line)["prompt"]) if not prompts:
print(f"corpus: {len(prompts)} prompts", flush=True) 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)
require_rig(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): 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 = push_round(url, prompts, a.model, a.parallel)
with concurrent.futures.ThreadPoolExecutor(a.parallel) as ex: dt = max(time.time() - t0, 1e-9)
for good in ex.map(lambda p: push_one(url, p), prompts):
ok += 1 if good else 0
dt = time.time() - t0
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.
+103 -39
View File
@@ -1,44 +1,53 @@
#!/usr/bin/env bash #!/usr/bin/env bash
# kld-gate-qwen38.sh — RUNS ON DANKSERVER (or anywhere the reports are). The campaign's # kld-gate-qwen38.sh — RUNS ON DANKSERVER (or anywhere the reports are). The campaign's
# acceptance gate: three KLD reports against ONE reference recording, one verdict. # acceptance gate: one verdict from the KLD reports of the containers that ride one reference
# recording.
# #
# The reference is the BF16 container built from the vendor checkpoint (run-qwen38-bf16.sh # The reference is the BF16 container built from the vendor checkpoint (run-qwen38-bf16.sh
# kldref, in the same drained window that records the calibration grams). Every scored # kldref, in the same drained window that records the calibration grams). Every scored
# container is measured against that one recording, so the KL scale runs from 0 = the # container is measured against that one recording, so the KL scale runs from 0 = the
# checkpoint itself downward in faithfulness. Three containers ride it: # checkpoint itself downward in faithfulness.
# #
# fp8.json the published FP8-native release — the achievable end of the interval # TWO MODES, because the disk decides how many anchors can exist at once:
# iq4r.json the payload frank-archon serves — the bar to beat
# cand.json the arm under test — the subject
# #
# The FP8 number is what makes the interval absolute instead of relative: "half the gap" # gate IQ4R.json CAND.json candidate vs the served payload
# is half of the distance between the 4-bit payload and the near-lossless release, not half # gate FP8.json IQ4R.json CAND.json the same, inside the absolute interval
# of an invented distance to the checkpoint. That is also why the FP8 payload is FETCHED and
# scored, never converted: it is a second opinion about what this checkpoint tolerates, not a
# step of the campaign. It lands LAST (disk sequencing) — by then the 335 GiB bf16 container
# is gone, so its 172.8 GiB fits.
# #
# The law, in order of severity, on the mean KL of one slice (--slice all | a corpus source): # iq4r.json is the payload frank-archon serves — the bar to beat. When the FP8-native release
# 1. KL(fp8) < KL(iq4r) the anchors are ordered. If the near-lossless # (172.8 GiB, fetched last because of the peak-disk law) is on disk, pass its report first:
# release does NOT sit closer to the checkpoint than # "half the gap" is then half of the distance between the 4-bit payload and the near-lossless
# the 4-bit payload, the corpus, the tokenizer or the # release, not half of an invented distance to the checkpoint. Without it the gate still
# boot shape is wrong and no arm number means anything. # answers the campaign's question — does the candidate beat what is served — and says so.
# 2. KL(fp8) < KL(cand) < KL(iq4r) the candidate lies strictly inside the interval — #
# outside it, something is mis-measured, not better # The law, on the summary of one slice (--slice all | a corpus source):
# 3. KL(cand) <= KL(iq4r) - R x gap it closes at least R (--recover, default 0.5) of the #
# gap = KL(iq4r) - KL(fp8): at least half the way from # candidate vs served (2 reports)
# what is served toward the release (agreed 2026-10-27) # 1. mean KL(cand) < mean KL(iq4r) the served payload is beaten on the corpus
# 4. top-1(cand) >= top-1(iq4r) the agreement the harness keys on does not fall # 2. p99.9(cand) <= p99.9(iq4r) and the tail is no worse — a mean can hide a few
# positions taking the whole hit
# 3. top-1(cand) >= top-1(iq4r) the agreement the harness keys on does not fall
# 4. every source slice passes 1-3 a win bought by regressing one distribution
# (code, prose, tools) is not a win
#
# inside the interval (3 reports), laws 1-4 above (1-3 renumbered to the interval and the
# recovery fraction, as agreed 2026-10-27) plus:
# 5. p99.9(cand) <= p99.9(iq4r) the tail law
# 6. every source slice passes the slice law
# #
# Booting the containers is NOT this script's job: `run-qwen38-bf16.sh score RAD OUT.json` # Booting the containers is NOT this script's job: `run-qwen38-bf16.sh score RAD OUT.json`
# writes the reports and owns the single engine argv (same pin, --tp 2, --kv-cache-dtype fp8, # writes the reports and owns the single engine argv (same pin, --tp 2, --kv-cache-dtype fp8,
# same step size, --deterministic, no speculation) that makes the three numbers comparable. # same step size, --deterministic, no speculation) that makes the numbers comparable.
# This script is arithmetic, so it re-runs against old reports and is testable without a GPU. # This script is arithmetic, so it re-runs against old reports and is testable without a GPU.
# #
# Usage: kld-gate-qwen38.sh FP8.json IQ4R.json CAND.json [--slice NAME] [--recover F] # Usage: kld-gate-qwen38.sh IQ4R.json CAND.json [--slice NAME]
# kld-gate-qwen38.sh FP8.json IQ4R.json CAND.json [--slice NAME] [--recover F]
set -uo pipefail set -uo pipefail
[ "$#" -ge 3 ] || { echo "usage: kld-gate-qwen38.sh FP8.json IQ4R.json CAND.json [--slice NAME] [--recover F]" >&2; exit 2; } if [ "$#" -lt 2 ]; then
echo "usage: kld-gate-qwen38.sh IQ4R.json CAND.json [--slice NAME]" >&2
echo " kld-gate-qwen38.sh FP8.json IQ4R.json CAND.json [--slice NAME] [--recover F]" >&2
exit 2
fi
python3 - "$@" <<'PY' python3 - "$@" <<'PY'
import json, sys import json, sys
@@ -52,13 +61,15 @@ while i < len(a):
else: else:
files.append(a[i]) files.append(a[i])
i += 1 i += 1
if len(files) != 3: if len(files) not in (2, 3):
sys.exit("usage: kld-gate-qwen38.sh FP8.json IQ4R.json CAND.json [--slice NAME] [--recover F]") sys.exit("usage: kld-gate-qwen38.sh IQ4R.json CAND.json [--slice NAME]\n"
" kld-gate-qwen38.sh FP8.json IQ4R.json CAND.json [--slice NAME] [--recover F]")
o = dict(zip(opts[0::2], opts[1::2])) o = dict(zip(opts[0::2], opts[1::2]))
slice_ = o.get("--slice", "all") slice_ = o.get("--slice", "all")
recover = float(o.get("--recover", 0.5)) recover = float(o.get("--recover", 0.5))
if not 0.0 < recover <= 1.0: if not 0.0 < recover <= 1.0:
sys.exit(f"--recover must be in (0,1], got {recover}") sys.exit(f"--recover must be in (0,1], got {recover}")
interval = len(files) == 3
def die(msg): def die(msg):
@@ -66,11 +77,15 @@ def die(msg):
raise SystemExit(2) raise SystemExit(2)
def read(path): def load(path):
try: try:
d = json.load(open(path)) return json.load(open(path))
except Exception as e: except Exception as e:
die(f"cannot read the KLD report {path}: {e}") die(f"cannot read the KLD report {path}: {e}")
def summary(path, d):
"""The slice this gate grades: 'all', or one corpus source."""
if slice_ == "all": if slice_ == "all":
g = d.get("all") g = d.get("all")
else: else:
@@ -80,23 +95,25 @@ def read(path):
f"{', '.join(sorted(d.get('by_source') or {})) or '(none)'}") f"{', '.join(sorted(d.get('by_source') or {})) or '(none)'}")
if g is None or "kld" not in g or "top1_agreement" not in g: if g is None or "kld" not in g or "top1_agreement" not in g:
die(f"{path} has no {slice_!r} summary — wrong report version?") die(f"{path} has no {slice_!r} summary — wrong report version?")
return d, g return g
docs = {label: load(path) for label, path in
zip(("fp8", "iq4r", "cand") if interval else ("iq4r", "cand"), files)}
rows = [] rows = []
for label, path in zip(("fp8", "iq4r", "cand"), files): for label, path in zip(("fp8", "iq4r", "cand") if interval else ("iq4r", "cand"), files):
d, g = read(path) g = summary(path, docs[label])
rows.append((label, path, d.get("model", "?"), g["kld"]["mean"], g["kld"]["p99_9"], rows.append((label, path, docs[label].get("model", "?"), g["kld"]["mean"],
g["top1_agreement"], g["positions"])) g["kld"]["p99_9"], g["top1_agreement"], g["positions"]))
print(f"reference: BF16 container; slice: {slice_}") print(f"reference: BF16 container; slice: {slice_}; "
+ ("inside the FP8 interval" if interval else "candidate vs served (no FP8 anchor)"))
print(f"{'':6} {'model':38} {'KL mean':>10} {'KL p99.9':>10} {'top-1':>8} positions") print(f"{'':6} {'model':38} {'KL mean':>10} {'KL p99.9':>10} {'top-1':>8} positions")
for label, path, model, kl, p999, top1, n in rows: for label, path, model, kl, p999, top1, n in rows:
print(f"{label:6} {str(model)[:38]:38} {kl:10.6f} {p999:10.4f} {100*top1:7.2f}% {n}") print(f"{label:6} {str(model)[:38]:38} {kl:10.6f} {p999:10.4f} {100*top1:7.2f}% {n}")
r = {label: (kl, top1) for label, _p, _m, kl, _q, top1, _n in rows} r = {label: (kl, p999, top1) for label, _p, _m, kl, p999, top1, _n in rows}
kl_fp8, kl_iq4r, kl_cand = r["fp8"][0], r["iq4r"][0], r["cand"][0] kl_cand, p_cand, t_cand = r["cand"]
t_iq4r, t_cand = r["iq4r"][1], r["cand"][1] kl_iq4r, p_iq4r, t_iq4r = r["iq4r"]
gap = kl_iq4r - kl_fp8
ok = True ok = True
@@ -106,6 +123,43 @@ def law(n, text, passed):
print(f" [{'PASS' if passed else 'FAIL'}] {n}. {text}") print(f" [{'PASS' if passed else 'FAIL'}] {n}. {text}")
def tail_law(n):
law(n, f"candidate KL p99.9 {p_cand:.4f} <= served iq4r {p_iq4r:.4f}",
p_cand <= p_iq4r)
# --- the slice law: every corpus source must pass the metric laws -------------
def slice_law(n):
"""A win that regresses one source class is not a win: check them all."""
mine = docs["cand"].get("by_source") or {}
theirs = docs["iq4r"].get("by_source") or {}
if not mine or not theirs:
law(n, "no per-source summaries to check", True)
return
missing = sorted(set(theirs) - set(mine))
extra = sorted(set(mine) - set(theirs))
if missing or extra:
law(n, "same source slices in both reports"
+ (f" (missing from cand: {', '.join(missing)})" if missing else "")
+ (f" (only in cand: {', '.join(extra)})" if extra else ""), False)
return
bad = []
for src in sorted(theirs):
c, q = mine[src], theirs[src]
worse_mean = c["kld"]["mean"] >= q["kld"]["mean"]
worse_tail = c["kld"]["p99_9"] > q["kld"]["p99_9"]
worse_top1 = c["top1_agreement"] < q["top1_agreement"]
if worse_mean or worse_tail or worse_top1:
why = ",".join(m for m, w in (("mean", worse_mean), ("p99.9", worse_tail),
("top-1", worse_top1)) if w)
bad.append(f"{src} ({why})")
law(n, "every source slice passes (mean, p99.9, top-1)"
+ (f" — regressed: {'; '.join(bad)}" if bad else ""), not bad)
if interval:
kl_fp8 = r["fp8"][0]
gap = kl_iq4r - kl_fp8
law(1, f"anchors ordered: fp8 {kl_fp8:.6f} < iq4r {kl_iq4r:.6f} (gap {gap:.6f})", law(1, f"anchors ordered: fp8 {kl_fp8:.6f} < iq4r {kl_iq4r:.6f} (gap {gap:.6f})",
gap > 0) gap > 0)
law(2, f"candidate inside the interval: {kl_fp8:.6f} < {kl_cand:.6f} < {kl_iq4r:.6f}", law(2, f"candidate inside the interval: {kl_fp8:.6f} < {kl_cand:.6f} < {kl_iq4r:.6f}",
@@ -117,6 +171,16 @@ law(3, f"candidate closes {100*max((kl_iq4r - kl_cand)/gap, 0):.1f}% of the "
gap > 0 and kl_cand <= need) gap > 0 and kl_cand <= need)
law(4, f"candidate top-1 {100*t_cand:.2f}% >= served iq4r top-1 {100*t_iq4r:.2f}%", law(4, f"candidate top-1 {100*t_cand:.2f}% >= served iq4r top-1 {100*t_iq4r:.2f}%",
t_cand >= t_iq4r) t_cand >= t_iq4r)
tail_law(5)
slice_law(6)
else:
law(1, f"candidate beats the served payload: KL {kl_cand:.6f} < iq4r {kl_iq4r:.6f}",
kl_cand < kl_iq4r)
tail_law(2)
law(3, f"candidate top-1 {100*t_cand:.2f}% >= served iq4r top-1 {100*t_iq4r:.2f}%",
t_cand >= t_iq4r)
slice_law(4)
print("KLD GATE " + ("PASS" if ok else "FAIL")) print("KLD GATE " + ("PASS" if ok else "FAIL"))
sys.exit(0 if ok else 1) sys.exit(0 if ok else 1)
PY PY
+140 -77
View File
@@ -34,14 +34,25 @@ import json
import os import os
import random import random
import re import re
import sys
from jinja2 import Environment, FileSystemLoader 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", CODE_EXT = (".sh", ".py", ".jinja", ".recipe", ".env", ".example",
".yaml", ".yml", ".bats", ".service", ".container", ".toml") ".yaml", ".yml", ".bats", ".service", ".container", ".toml")
JSONL_EXTS = (".jsonl", ".json") JSONL_EXTS = (".jsonl", ".json")
DOC_DIR = "serverAppConfigs-docs" 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 = [ FLEET_CMDS = [
"systemctl status frank-archon --no-pager", "systemctl status frank-archon --no-pager",
@@ -142,7 +153,7 @@ def load_parquet_texts(path: str, kind: str, rng: random.Random,
return out[:cap] return out[:cap]
def main() -> None: def parse_args() -> argparse.Namespace:
ap = argparse.ArgumentParser(description=__doc__) ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--repo", required=True) ap.add_argument("--repo", required=True)
ap.add_argument("--tokenizer-json", 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-code-parquet", default="")
ap.add_argument("--hf-prose-parquet", default="") ap.add_argument("--hf-prose-parquet", default="")
ap.add_argument("--seed", type=int, default=20261006) 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) def load_template(template: str) -> tuple[Environment, str]:
env = Environment(loader=FileSystemLoader(os.path.dirname(a.template)), """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) trim_blocks=True, lstrip_blocks=True)
env.filters["tojson"] = lambda v: json.dumps(v, ensure_ascii=False) env.filters["tojson"] = lambda v: json.dumps(v, ensure_ascii=False)
env.filters["raise_exception"] = lambda m: (_ for _ in ()).throw( env.filters["raise_exception"] = lambda m: (_ for _ in ()).throw(
ValueError(m)) 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 = [], [], [] 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(".")] dirs[:] = [d for d in dirs if not d.startswith(".")]
for f in sorted(files): 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"): if DOC_DIR in p and f.endswith(".md"):
prose.append(p) prose.append(p)
elif p.startswith(DOC_DIR): elif p.startswith(DOC_DIR):
@@ -180,64 +195,92 @@ def main() -> None:
code.append(p) code.append(p)
elif f.endswith(JSONL_EXTS) and f != "package-lock.json": elif f.endswith(JSONL_EXTS) and f != "package-lock.json":
jsonl.append(p) jsonl.append(p)
# file-level split: bucket 0 -> KLD pool, 1..4 -> calibration pool return {"code": code, "prose": prose, "tools": jsonl}
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], 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]) [p for p in lst if fnv(p) % 5 != 0])
for cls, lst in classes.items()}
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 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) cmds = harvest_cmds(a.repo, rng)
# fragment pools per class, corpus-split first
frags: dict[str, dict[str, list[str]]] = {} frags: dict[str, dict[str, list[str]]] = {}
for cls in pools: for cls, (kld_files, cal_files) in pools.items():
kld_files, cal_files = pools[cls]
fr = {}
src = {"kld": kld_files, "cal": cal_files}
if cls == "tools": if cls == "tools":
fr["kld"] = [tool_call(rng, cmds) for _ in range(600)] frags[cls] = {"kld": [tool_call(rng, cmds) for _ in range(600)],
fr["cal"] = [tool_call(rng, cmds) for _ in range(40000)] "cal": [tool_call(rng, cmds) for _ in range(40000)]}
else: else:
fr["kld"] = [text_of(p) for p in src["kld"]] frags[cls] = {"kld": [fragment_text(a.repo, p, tok) for p in kld_files],
fr["cal"] = [text_of(p) for p in src["cal"]] "cal": [fragment_text(a.repo, p, tok) for p in cal_files]}
frags[cls] = fr
# per-class fragment caps: prose rows are short (~66 tok) so need many; # 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 # 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), for cls, spec in (("code", a.hf_code_parquet),
("prose", a.hf_prose_parquet)): ("prose", a.hf_prose_parquet)):
for path in [p for p in spec.split(",") if p]: for path in [p for p in spec.split(",") if p]:
frags[cls]["cal"].extend( 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: for cls in frags:
rng.shuffle(frags[cls]["cal"]) 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]]: def pack_class(cls: str, pool: list[str], want: int, tok,
targets = {"code": .45, "prose": .30, "tools": .25} chunk_tokens: int) -> list[dict]:
rows: list[dict] = [] """Greedily pack `pool` into chunk_tokens-sized calibration prompts."""
for cls, frac in targets.items():
want = int(a.calib_tokens * frac)
pool = frags[cls]["cal"]
if not pool: if not pool:
continue return []
got, i, buf, nbuf = 0, 0, [], 0 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: 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 > chunk_tokens and buf:
rows.append({"prompt": "\n".join(buf), "cls": cls}) rows.append({"prompt": "\n".join(buf), "cls": cls})
got += nbuf got += nbuf
buf, nbuf = [], 0 buf, nbuf = [], 0
@@ -245,22 +288,36 @@ def main() -> None:
break break
buf.append(frag) buf.append(frag)
nbuf += n nbuf += n
if i > len(pool) * 40: if i > len(pool) * POOL_CYCLES:
break # pool exhausted with repeats; stop break # pool exhausted with repeats; stop
if buf and got < want: if buf and got < want:
rows.append({"prompt": "\n".join(buf), "cls": cls}) rows.append({"prompt": "\n".join(buf), "cls": cls})
got += nbuf got += nbuf
print(f"calib {cls}: {got} tokens, {i} fragments") print(f"calib {cls}: {got} tokens, {i} fragments")
rng.shuffle(rows) return rows
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_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"] classes = ["code", "prose", "tools"]
i = {c: 0 for c in classes} cursor = {c: 0 for c in classes}
guard = 0 guard = 0
while total < a.kld_tokens and guard < 5000: while total < a.kld_tokens and guard < KLD_GUARD:
guard += 1 guard += 1
msgs, srcs = [], [] msgs, srcs = [], []
for t in range(rng.choice([2, 3, 4])): for t in range(rng.choice([2, 3, 4])):
@@ -268,49 +325,55 @@ def main() -> None:
pool = frags[cls]["kld"] pool = frags[cls]["kld"]
if not pool: if not pool:
continue continue
frag = pool[i[cls] % len(pool)] frag = pool[cursor[cls] % len(pool)]
i[cls] += 2 cursor[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}"})
msgs.append({"role": "assistant", msgs.append({"role": "assistant",
"content": frag[:4000]}) "content": frag[:4000]})
srcs.append(cls) srcs.append(cls)
rendered = env.get_template(tname).render( rendered = template.render(messages=msgs, add_generation_prompt=False)
messages=msgs, add_generation_prompt=False) n = qc.n_tokens(tok, rendered)
n = toks(rendered)
if n < a.kld_min_prompt: if n < a.kld_min_prompt:
continue continue
rows.append({"prompt": rendered, rows.append({"prompt": rendered,
"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)
nc, rows = emit_calib() def manifest(a: argparse.Namespace, tname: str, nc: int, nk: int,
with open(os.path.join(a.out_dir, "calib.jsonl"), "w") as f: scored: int) -> dict:
for r in rows: return {
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 = {
"seed": a.seed, "template": tname, "seed": a.seed, "template": tname,
"tokenizer": os.path.basename(a.tokenizer_json), "tokenizer": os.path.basename(a.tokenizer_json),
"calib_prompts": nc, "kld_prompts": nk, "calib_prompts": nc, "kld_prompts": nk,
"kld_scored_positions": scored, "kld_scored_positions": scored,
"kld_score_tail": a.score_tail, "kld_score_tail": a.score_tail,
"disjoint_file_pools": "local KLD material only; HF code feeds calib", "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, "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__": if __name__ == "__main__":
+122
View File
@@ -0,0 +1,122 @@
#!/usr/bin/env bash
# qwen38-calib-window.sh — RUNS ON DANKSERVER. The calibration window, and nothing else:
# boot the BF16 rig, wait until it is actually serving, push the corpus until the engine says
# the tap is full, stop the rig, restore frank-archon, and bank the reference container.
#
# The reference recording is already taken (kld-ref/kld.json exists), so this window is the
# calibration tap alone — the one job that needs both cards after it. The kldref phase is NOT
# re-run here: the chained launcher that used to do that moved the banked kld-ref into trash
# first, which is how a 16-hour phase gets paid for twice.
#
# Every phase is gated on the previous one, because the last time this ran unguarded
# (2026-10-29) the rig's boot died of an OOM, the health loop quietly fell through, and the
# pusher fed four rounds of 10M tokens into RESTORED PRODUCTION with nothing recorded:
#
# 1. the runner must still be alive and its container must answer /health, or this script
# kills it, restores the service and exits non-zero — no push, ever;
# 2. the pusher refuses on its own unless the rig wrote its marker into the tap dir, the
# marker names this container, and the container is running (scripts/calib-push.py
# --tap-marker), so a second lock survives a mistake in this file;
# 3. the reference container is banked into trash only when BOTH the serve and the push
# exited 0 and the tap dir holds grams. The checkpoint that reproduces it stays.
#
# knobs: Q38_WINDOW_CALIB_CORPUS (calib-7168.jsonl — the re-chunked file the rig's pool can
# hold), Q38_WINDOW_HEALTH_TIMEOUT (seconds to wait for /health), Q38_WINDOW_ROUNDS
# (pusher rounds; one is 10M tokens), Q38_WINDOW_KEEP_RAD=1 to leave the .rad in place.
# Usage: scripts/qwen38-calib-window.sh [--dry-run]
set -uo pipefail
cd "$(dirname "${BASH_SOURCE[0]}")/.." || exit 9
DRY=0
[ "${1:-}" = "--dry-run" ] && DRY=1
say() { printf '%s %s\n' "$(date -Is)" "$*"; }
MODELS="${FRANK_QWEN38_MODELS:-/home/dank/models}"
CAMPAIGN="${FRANK_QWEN38_CAMPAIGN:-/home/dank/campaign/qwen38-w5nl64}"
RAD="${FRANK_QWEN38_RAD:-$MODELS/rad/qwen38-bf16.rad}"
CORPUS="${Q38_WINDOW_CALIB_CORPUS:-/home/dank/corpora/out/calib-7168.jsonl}"
KLDREF="$CAMPAIGN/kld-ref/kld.json"
CALIB="$CAMPAIGN/calib"
HEALTH_TIMEOUT="${Q38_WINDOW_HEALTH_TIMEOUT:-3600}"
ROUNDS="${Q38_WINDOW_ROUNDS:-6}"
RIG=qwen38-bf16
SERVICE=frank-archon
PY="${PY:-python3}"
# ---- preflight: everything this window needs, before a card is touched -------
[ -f "$KLDREF" ] || { echo "REFUSED: no reference recording at $KLDREF — record it first (scripts/run-qwen38-bf16.sh kldref); this window only fills the calibration tap" >&2; exit 2; }
[ -f "$RAD" ] || { echo "REFUSED: no BF16 reference container at $RAD — nothing to serve; scripts/build-qwen38-bf16-rad.sh" >&2; exit 2; }
[ -f "$CORPUS" ] || { echo "REFUSED: no calibration corpus at $CORPUS" >&2; exit 2; }
[ -d "$CALIB" ] || { echo "REFUSED: no tap dir at $CALIB" >&2; exit 2; }
[ -w "$CALIB" ] || { echo "REFUSED: tap dir $CALIB is not writable by $(id -un)" >&2; exit 2; }
! [[ "$HEALTH_TIMEOUT" =~ ^[0-9]+$ ]] && { echo "REFUSED: Q38_WINDOW_HEALTH_TIMEOUT must be seconds (got: $HEALTH_TIMEOUT)" >&2; exit 2; }
! [[ "$ROUNDS" =~ ^[0-9]+$ ]] && { echo "REFUSED: Q38_WINDOW_ROUNDS must be a round count (got: $ROUNDS)" >&2; exit 2; }
if [ "$DRY" = 1 ]; then
echo "dry run: calib window — the service is left running, nothing is mounted or written"
echo "would run: bash scripts/run-qwen38-bf16.sh calib (rig boots, writes $CALIB/.rig-boot)"
echo "would wait: /health up within ${HEALTH_TIMEOUT}s, else kill the rig, restore $SERVICE, exit 1"
echo "would run: $PY scripts/calib-push.py --container $RIG --corpus $CORPUS --max-rounds $ROUNDS"
echo "would bank: $RAD -> ~/.AGENTTRASH/$(date +%Y%m%d)-qwen38-bf16-rad/ (only if both phases exit 0)"
exit 0
fi
say "calib window start pid=$$ corpus=$CORPUS rounds=$ROUNDS"
[ -n "$(ls -A "$CALIB" 2>/dev/null)" ] && say "tap dir $CALIB is not empty; the engine's token floor is pooled, so an earlier partial tap still counts"
# ---- phase 1: the rig -------------------------------------------------------
say "phase=calib serve begin"
bash scripts/run-qwen38-bf16.sh calib & spid=$!
rig_up() {
podman inspect -f '{{.State.Running}}' "$RIG" 2>/dev/null | grep -qx true \
&& curl -sf --max-time 10 -o /dev/null http://127.0.0.1:7770/health
}
waited=0
until rig_up; do
if ! kill -0 "$spid" 2>/dev/null; then
wait "$spid"; crc=$?
say "phase=calib serve end rc=$crc (the rig exited before it ever answered /health)"
echo "FAIL: the rig boot died (rc=$crc) — nothing was pushed; read /home/dank/logs/qwen38_bf16_calib.log" >&2
exit 1
fi
if [ "$waited" -ge "$HEALTH_TIMEOUT" ]; then
say "phase=calib health never came after ${HEALTH_TIMEOUT}s; killing the rig"
podman stop "$RIG" >/dev/null 2>&1 || true
wait "$spid"
echo "FAIL: the rig never answered /health within ${HEALTH_TIMEOUT}s — nothing was pushed" >&2
exit 1
fi
sleep 15
waited=$((waited + 15))
done
say "phase=calib health ok waited=${waited}s"
# ---- phase 2: the tap -------------------------------------------------------
say "phase=calib push begin"
"$PY" scripts/calib-push.py --port 7770 --parallel 8 --container "$RIG" \
--tap-dir "$CALIB" --corpus "$CORPUS" --max-rounds "$ROUNDS"
prc=$?
say "phase=calib push end rc=$prc"
podman stop "$RIG" >/dev/null 2>&1 || true
wait "$spid"; crc=$?
say "phase=calib serve end rc=$crc"
if [ "$prc" -ne 0 ] || [ "$crc" -ne 0 ]; then
echo "FAIL: serve rc=$crc push rc=$prc — the tap is NOT banked and $RAD stays: $(basename "$RAD")" >&2
exit 1
fi
[ -n "$(ls -A "$CALIB" 2>/dev/null)" ] || { echo "FAIL: both phases exited 0 but the tap dir $CALIB is empty" >&2; exit 1; }
say "TAP BANKED: $(ls -1 "$CALIB" | wc -l) file(s) in $CALIB"
# ---- phase 3: bank the reference container ----------------------------------
if [ "${Q38_WINDOW_KEEP_RAD:-0}" = "1" ]; then
say "reference .rad kept (Q38_WINDOW_KEEP_RAD=1)"
else
T="$HOME/.AGENTTRASH/$(date +%Y%m%d)-qwen38-bf16-rad"
mkdir -p "$T" || { echo "FAIL: cannot create trash dir $T" >&2; exit 1; }
mv "$RAD" "$T/" && say "reference .rad banked in $T (the checkpoint that rebuilds it stays)"
fi
say "calib window end"
+94
View File
@@ -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")
+85 -71
View File
@@ -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 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. half corpus behind. A JSON summary of what was written goes to stdout.
""" """
import argparse import argparse
import dataclasses
import json import json
import os import os
import sys 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: def parse_args() -> argparse.Namespace:
@@ -69,95 +76,96 @@ def parse_args() -> argparse.Namespace:
return ap.parse_args() return ap.parse_args()
def load_tokenizer(path: str): @dataclasses.dataclass(frozen=True)
try: class Window:
from tokenizers import Tokenizer """One scoring document cut out of a parent record."""
except ImportError as exc: # the corpus scripts carry their own venv prompt: str
raise SystemExit( score_from: int
f"REFUSED: tokenizers is not importable by {sys.executable} — run the " n_tokens: int
f"corpus venv's python (on dankserver: /home/dank/venvs/radcorpus/bin/python) " clamped: bool
f"or set FRANK_QWEN38_TOKENIZER: {exc}") from None source: object = MISSING
if not os.path.isfile(path):
raise SystemExit(f"REFUSED: no tokenizer at {path}") def record(self) -> dict:
return Tokenizer.from_file(path) 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: def validate(a: argparse.Namespace) -> None:
print(f"REFUSED: {msg}", file=sys.stderr) """Everything that makes this run unusable, named before a token is cut."""
sys.exit(2) # 2 is the code argparse uses for unusable arguments
def main() -> None:
a = parse_args()
if a.max_doc_tokens <= 0: 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: 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: if a.max_doc_tokens <= a.score_tail:
refuse(f"--max-doc-tokens {a.max_doc_tokens} cannot hold a {a.score_tail}-position " qc.refuse(f"--max-doc-tokens {a.max_doc_tokens} cannot hold a "
"score tail: every window would be all tail and no context") f"{a.score_tail}-position score tail: every window would be all "
"tail and no context")
if not os.path.isfile(a.corpus): 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): 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 " qc.refuse(f"--out {a.out} is the corpus itself; move it to trash first — the "
"means the old segmentation stays recoverable") "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: 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)
docs_in = docs_out = tokens_in = tokens_out = 0 def cut(rec: dict, tok, a: argparse.Namespace,
longest_child = 0 where: str) -> tuple[list[Window], int]:
clamped = scored = 0 """Cut one record's token ids into windows, proving each one round-trips.
sources: dict[str, int] = {}
fd, tmp = tempfile.mkstemp(dir=os.path.dirname(os.path.abspath(a.out)) or ".", The round-trip encode is the whole reason this tool may claim the reference
prefix=".rechunk-") scores the corpus's own token sequence: decoding a window to text and
try: encoding it back must return the identical ids, or the score is of a
with os.fdopen(fd, "w", encoding="utf-8") as out: different document than the one on disk.
with open(a.corpus, encoding="utf-8") as src: """
for lineno, line in enumerate(src, 1): ids = tok.encode(rec["prompt"], add_special_tokens=False).ids
if not line.strip(): windows: list[Window] = []
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): for start in range(0, max(len(ids), 1), a.max_doc_tokens):
window = ids[start:start + a.max_doc_tokens] window = ids[start:start + a.max_doc_tokens]
if not window: if not window:
continue continue
score_from = max(len(window) - a.score_tail, 0) text = tok.decode(window)
if len(window) < a.score_tail: if tok.encode(text, add_special_tokens=False).ids != window:
clamped += 1 qc.refuse(f"{where} window at token {start} does not round-trip through "
scored += len(window) - score_from "text — the reference would score a different token sequence "
child = {"prompt": tok.decode(window), "than the corpus holds")
"score_from": score_from} windows.append(Window(prompt=text,
if "source" in rec: score_from=max(len(window) - a.score_tail, 0),
child["source"] = rec["source"] n_tokens=len(window),
out.write(json.dumps(child, ensure_ascii=False) + "\n") 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 = clamped = scored = 0
sources: dict[str, int] = {}
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 docs_out += 1
tokens_out += len(window) tokens_out += w.n_tokens
longest_child = max(longest_child, len(window)) 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: if "source" in rec:
for cls in str(rec["source"]).split("+"): for cls in str(rec["source"]).split("+"):
sources[cls] = sources.get(cls, 0) + 1 sources[cls] = sources.get(cls, 0) + 1
if tokens_out != tokens_in: if tokens_out != tokens_in:
raise SystemExit(f"REFUSED: the windows hold {tokens_out} tokens and the corpus " raise SystemExit(f"REFUSED: the windows hold {tokens_out} tokens and the corpus "
f"has {tokens_in} — re-chunking lost or invented tokens") f"has {tokens_in} — re-chunking lost or invented tokens")
os.replace(tmp, a.out) return {
except BaseException:
if os.path.exists(tmp):
os.unlink(tmp)
raise
os.chmod(a.out, 0o644)
print(json.dumps({
"corpus": a.corpus, "corpus": a.corpus,
"out": a.out, "out": a.out,
"tokenizer": os.path.basename(a.tokenizer_json), "tokenizer": os.path.basename(a.tokenizer_json),
@@ -171,7 +179,13 @@ def main() -> None:
"scored_positions": scored, "scored_positions": scored,
"clamped_score_from": clamped, "clamped_score_from": clamped,
"source_documents": sources, "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__": if __name__ == "__main__":
+59 -5
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
@@ -192,6 +193,25 @@ if ! [[ "$MAX_MODEL_LEN" =~ ^[0-9]+$ ]]; then
exit 2 exit 2
fi fi
# Where the n-gram table lives. On the bf16 payload it is 95.37 GiB and `ram`
# PINS it: the calibration boot died at "startup stopped: out of memory" with
# 230.22 GiB pinned of 246.07 GiB and 8.06 GiB available (measured 2026-10-29),
# while the reference recording and every score boot had already succeeded with
# the same pool and slots. The difference that matters is the job: calib only
# needs the routed weights prefilled, so it reads the table from the container's
# own file tier and the pin never happens; kldref and score keep production's
# placement, because their numbers are only comparable when every other term of
# the shape is identical. FRANK_QWEN38_NGRAM_PLACEMENT overrides either default.
case "$MODE" in
calib) NGRAM_PLACEMENT="${FRANK_QWEN38_NGRAM_PLACEMENT:-disk}" ;;
*) NGRAM_PLACEMENT="${FRANK_QWEN38_NGRAM_PLACEMENT:-ram}" ;;
esac
case "$NGRAM_PLACEMENT" in
auto|vram|ram|disk) ;;
*) echo "REFUSED: FRANK_QWEN38_NGRAM_PLACEMENT must be auto|vram|ram|disk (got: $NGRAM_PLACEMENT)" >&2
exit 2 ;;
esac
# Every container this campaign boots is read through the ro /models mount, so # Every container this campaign boots is read through the ro /models mount, so
# the host path must live under it; the container-side spelling is what goes # the host path must live under it; the container-side spelling is what goes
# into the argv. Same rule for the container a score pass opens. # into the argv. Same rule for the container a score pass opens.
@@ -229,7 +249,7 @@ ARGS=(--tp 2 --placement expert_tiered --weights-disk-tier
--host-pool-mib "$HOST_POOL_MIB" --gpu-headroom-mib "$GPU_HEADROOM_MIB" --host-pool-mib "$HOST_POOL_MIB" --gpu-headroom-mib "$GPU_HEADROOM_MIB"
--deterministic --num-speculative-tokens 0 --deterministic --num-speculative-tokens 0
--kv-cache-dtype fp8 --max-model-len "$MAX_MODEL_LEN" --kv-cache-dtype fp8 --max-model-len "$MAX_MODEL_LEN"
--ngram-placement ram --ngram-placement "$NGRAM_PLACEMENT"
--override-chat-template /models/tpl.jinja) --override-chat-template /models/tpl.jinja)
MOUNTS=(-v "$MODELS:/models:ro") MOUNTS=(-v "$MODELS:/models:ro")
TAPENV=() TAPENV=()
@@ -250,7 +270,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 +297,28 @@ 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}")
# The tap's own authorization. A pusher must see this marker before it sends
# a prompt (calib-push.py --tap-marker), it names the container so a push
# aimed anywhere else refuses, and a boot that fails removes it again below.
# Measured 2026-10-29: with no marker, the chain pushed 4 rounds x 10M tokens
# into RESTORED PRODUCTION after the rig's boot died, and nothing recorded.
if [ "$DRY" = 1 ]; then
:
else
printf 'container=%s\nimage=%s\nmode=calib\nstarted=%s\npid=%s\n' \
"$C" "$IMAGE" "$(date -Is)" "$$" > "$CALIB_DIR/.rig-boot"
fi
;; ;;
esac esac
@@ -352,7 +392,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
@@ -360,5 +407,12 @@ podman rm -f "$C" >/dev/null 2>&1 || true
"${CMD[@]}" 2>&1 | tee "$LOG" "${CMD[@]}" 2>&1 | tee "$LOG"
rc=${PIPESTATUS[0]} rc=${PIPESTATUS[0]}
# A boot that failed must not leave the tap marked as rig-authorized: the marker
# is what calib-push.py trusts, and the point of it is that a dead rig can never
# be pushed into again.
if [ "$MODE" = calib ] && [ "$rc" -ne 0 ] && [ -f "$CALIB_DIR/.rig-boot" ]; then
mv -f "$CALIB_DIR/.rig-boot" "$CALIB_DIR/.rig-boot.failed-$(date +%Y%m%d-%H%M%S)" 2>/dev/null || true
echo "the rig boot failed (rc=$rc): the tap marker is retired, so no pusher can take this tap for a live rig"
fi
echo "run-qwen38-bf16 $MODE exited rc=$rc; $SERVICE restored; log: $LOG" echo "run-qwen38-bf16 $MODE exited rc=$rc; $SERVICE restored; log: $LOG"
exit "$rc" exit "$rc"
+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")
+36 -5
View File
@@ -562,6 +562,29 @@ vllm-radiance era (this table opens on 09-26; the era page claims 10-10).
fork bomb inside the container. Kill with `podman rm -f` (the cgroup owns every generation) fork bomb inside the container. Kill with `podman rm -f` (the cgroup owns every generation)
never pattern-kill an interpreter name on this host; use `TORCH_LOGS` for dynamo questions. never pattern-kill an interpreter name on this host; use `TORCH_LOGS` for dynamo questions.
### The quantization pipeline
- **A calibration tap is only a tap when the rig that opened it is running.** The runner writes
`.rig-boot` into the tap dir (container, image, pid) and retires it when its boot exits non-zero;
`calib-push.py` refuses without that marker, refuses when it names another container, and refuses
when the named container is not running; `qwen38-calib-window.sh` stops on a failed phase instead
of falling through its health wait. All three exist because on 2026-10-29 the unguarded chain's
rig boot died of an OOM, the health loop expired without failing, and the pusher fed four rounds
of 10M tokens into **restored production** — 86 minutes of live prefill that recorded nothing.
- **The n-gram table's pin is what starves a rig boot, and the tier is a knob.** On the BF16 payload
the table is 95.37 GiB, `--ngram-placement ram` pins it, and the calibration boot died at
`startup stopped: out of memory` with 230.22 GiB pinned of 246.07 GiB and 8.06 GiB available —
while the reference recording and every score boot succeeded at the same pool and slot count.
Calibration only needs the routed weights prefilled, so it reads the table from the container's
file tier (`--ngram-placement disk`); kldref and score keep production's placement, because their
numbers are only comparable at an identical shape.
- **Acceptance is three metrics and every slice, not a mean.** `scripts/kld-gate-qwen38.sh` grades
mean KL, the p99.9 tail, top-1 agreement and every `by_source` slice (code, prose, tools):
a mean can hide a few positions taking the whole hit, and an overall win can be bought by
regressing one distribution. The FP8-native anchor is an interval, not a requirement — without
the 172.8 GiB fetch the gate runs in two-report mode (candidate vs the served payload) and says
so on its first line.
## Contract (the wire) ## Contract (the wire)
- **Live surface:** `/health` (== `/ping`), `/v1/models`, `/v1/chat/completions` - **Live surface:** `/health` (== `/ping`), `/v1/models`, `/v1/chat/completions`
@@ -638,11 +661,19 @@ vllm-radiance era (this table opens on 09-26; the era page claims 10-10).
rebuild), `scripts/download-frank-archon-swift15.sh` (the BF16 source a w4dn8 re-convert rebuild), `scripts/download-frank-archon-swift15.sh` (the BF16 source a w4dn8 re-convert
needs), `scripts/hfget.sh` / `scripts/hf-fetch.sh`; template needs), `scripts/hfget.sh` / `scripts/hf-fetch.sh`; template
`scripts/qwen38-froggeric-v22.5.jinja`. `scripts/qwen38-froggeric-v22.5.jinja`.
- Convert campaign (dormant, kept for a future quant): `scripts/convert-swift15-w4dn8.sh` - The campaign (qwen38-w5nl64, live): `scripts/run-qwen38-bf16.sh` (kldref / calib / score — the
`scripts/recipes/qwen4exp-swift15-w4dn8-i8-hc8m.recipe`, `scripts/run-swift15-bf16.sh` ONE measuring shape), `scripts/qwen38-calib-window.sh` (the gated calibration window),
`scripts/kld-gate-swift15.sh`, `scripts/kld-then-calib.sh`, `scripts/calib-push.py` `scripts/convert-qwen38-w5nl64.sh` with
`scripts/make-calib-corpus.py`, `scripts/derive-protected-experts.py` `scripts/recipes/qwen4exp-qwen38-w5nl64-bf16trunk.recipe` (the arm driver and its template),
`scripts/apply-protected-cast.sh`, `scripts/verify-swift15-ple.py`. `scripts/kld-gate-qwen38.sh` (acceptance), `scripts/calib-push.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`.
- Retired swift15-era campaign scripts, superseded by the above and kept only as reference:
`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/verify-swift15-ple.py`.
- Probes: `scripts/probe-frank-archon-kv.py` (reads slots/window/dtype from the deployed env so - 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` it cannot drift; leg D = decode), `scripts/frank-archon-expert-activity.py`
`scripts/autoresearch-frank-archon.sh` + `scripts/autoresearch/frank-archon-measure.py`. `scripts/autoresearch-frank-archon.sh` + `scripts/autoresearch/frank-archon-measure.py`.
+174
View File
@@ -200,6 +200,9 @@ DERIVE="$REPO_ROOT/scripts/derive-protected-experts.py"
APPLY="$REPO_ROOT/scripts/apply-protected-cast.sh" APPLY="$REPO_ROOT/scripts/apply-protected-cast.sh"
CONVERT="$REPO_ROOT/scripts/convert-qwen38-w5nl64.sh" CONVERT="$REPO_ROOT/scripts/convert-qwen38-w5nl64.sh"
GATE="$REPO_ROOT/scripts/kld-gate-qwen38.sh" GATE="$REPO_ROOT/scripts/kld-gate-qwen38.sh"
RUNNER="$REPO_ROOT/scripts/run-qwen38-bf16.sh"
PUSHER="$REPO_ROOT/scripts/calib-push.py"
WINDOW="$REPO_ROOT/scripts/qwen38-calib-window.sh"
TEMPLATE="$REPO_ROOT/scripts/recipes/qwen4exp-qwen38-w5nl64-bf16trunk.recipe" TEMPLATE="$REPO_ROOT/scripts/recipes/qwen4exp-qwen38-w5nl64-bf16trunk.recipe"
# make_gram <dir> <layer> <expert> <energy> <rows> — a RADGRAM2 down-Gram whose # make_gram <dir> <layer> <expert> <energy> <rows> — a RADGRAM2 down-Gram whose
@@ -525,3 +528,174 @@ gate_case() { # gate_case <kl fp8> <kl iq4r> <kl cand> <top1 cand> [extra flags.
[[ "$output" == *"has no slice 'code'"* ]] [[ "$output" == *"has no slice 'code'"* ]]
[[ "$output" == *"it has: ops"* ]] [[ "$output" == *"it has: ops"* ]]
} }
# --- the n-gram tier: the pin that starved the calibration boot ---------------
@test "the calibration boot reads the n-gram table from the file tier" {
# 2026-10-29: 95.37 GiB pinned beside a 65.59 GiB/rank pool left 8.06 GiB of
# RAM and the boot died "startup stopped: out of memory". calib only needs the
# routed weights prefilled, so it reads the table off the container's file tier.
run bash "$RUNNER" calib --dry-run
[ "$status" -eq 0 ]
grep -A1 -x -- '--ngram-placement' <<<"$output" | grep -qx 'disk'
}
@test "the recording and every score pass keep production's n-gram placement" {
# Those numbers are only comparable when the shape is identical; the tier is
# part of the shape.
run bash "$RUNNER" kldref --dry-run
[ "$status" -eq 0 ]
grep -A1 -x -- '--ngram-placement' <<<"$output" | grep -qx 'ram'
}
@test "an unknown n-gram placement is refused by name before the boot" {
run env FRANK_QWEN38_NGRAM_PLACEMENT=ssd bash "$RUNNER" calib --dry-run
[ "$status" -eq 2 ]
[[ "$output" == *"auto|vram|ram|disk"* ]]
}
# --- the pusher: a tap without a live rig is not a tap ------------------------
# stub_podman <dir> <running|absent> — a podman that answers the one question the
# pusher asks: is the rig container running?
stub_podman() {
mkdir -p "$1/bin"
if [ "$2" = running ]; then
printf '#!/usr/bin/env bash\necho true\n' > "$1/bin/podman"
elif [ "$2" = stopped ]; then
printf '#!/usr/bin/env bash\necho false\n' > "$1/bin/podman"
else
printf '#!/usr/bin/env bash\nexit 125\n' > "$1/bin/podman"
fi
chmod +x "$1/bin/podman"
}
pusher_fixture() { # pusher_fixture <marker-container or empty>
D="$BATS_TEST_TMPDIR/push"
mkdir -p "$D/tap" "$D/bin"
printf '{"prompt":"hello"}\n' > "$D/corpus.jsonl"
if [ -n "$1" ]; then
printf 'container=%s\nimage=radiance@sha256:test\nmode=calib\npid=1\n' "$1" \
> "$D/tap/.rig-boot"
fi
echo "$D"
}
@test "the pusher refuses a tap that carries no rig marker: production is not a rig" {
# With no marker, the old driver fed four rounds of 10M tokens into restored
# production after the rig's boot had died, and recorded nothing.
D="$(pusher_fixture '')"
run python3 "$PUSHER" --corpus "$D/corpus.jsonl" --tap-dir "$D/tap" --port 1
[ "$status" -eq 2 ]
[[ "$output" == *"no rig marker"* ]]
}
@test "the pusher refuses a marker that names another container" {
D="$(pusher_fixture arbench)"
run python3 "$PUSHER" --corpus "$D/corpus.jsonl" --tap-dir "$D/tap" \
--container qwen38-bf16 --port 1
[ "$status" -eq 2 ]
[[ "$output" == *"names container 'arbench'"* ]]
[[ "$output" == *"targets 'qwen38-bf16'"* ]]
}
@test "the pusher refuses when the rig container is not running" {
D="$(pusher_fixture qwen38-bf16)"
stub_podman "$D" stopped
run env PATH="$D/bin:$PATH" python3 "$PUSHER" --corpus "$D/corpus.jsonl" \
--tap-dir "$D/tap" --container qwen38-bf16 --port 1
[ "$status" -eq 2 ]
[[ "$output" == *"is not running"* ]]
}
# --- the acceptance gate without an FP8 anchor -------------------------------
# kld_report_slices <path> <model> <kl all> <kl code> <top1 all> <top1 code>
kld_report_slices() {
python3 - "$@" <<'PY'
import json, sys
path, model, kl_all, kl_code, t_all, t_code = sys.argv[1:7]
def g(kl, t, n=25000):
return {"kld": {"mean": float(kl), "p99_9": float(kl) * 4},
"top1_agreement": float(t), "positions": n}
json.dump({"model": model, "all": g(kl_all, t_all),
"by_source": {"code": g(kl_code, t_code), "ops": g(kl_all, t_all)}},
open(path, "w"))
PY
}
gate2() { # gate2 <kl iq4r> <kl cand> <top1 iq4r> <top1 cand> [extra]
D="$BATS_TEST_TMPDIR/kl2"
mkdir -p "$D"
kld_report_slices "$D/iq4r.json" iq4r-moe "$1" "$1" "$3" "$3"
kld_report_slices "$D/cand.json" w5nl64 "$2" "$2" "$4" "$4"
run bash "$GATE" "$D/iq4r.json" "$D/cand.json" "${@:5}"
}
@test "a candidate that beats the served payload on mean, tail and top-1 passes without an anchor" {
gate2 0.0190 0.0090 0.962 0.971
[ "$status" -eq 0 ]
[[ "$output" == *"candidate vs served (no FP8 anchor)"* ]]
[[ "$output" == *"KLD GATE PASS"* ]]
}
@test "a better mean with a worse tail fails: the tail is where a mean hides its damage" {
D="$BATS_TEST_TMPDIR/kl3"
mkdir -p "$D"
kld_report_slices "$D/iq4r.json" iq4r-moe 0.0190 0.0190 0.962 0.962
python3 - "$D/cand.json" <<'PY'
import json, sys
# mean better than the served payload, p99.9 worse: the shape a mean-only gate waves through
g = {"kld": {"mean": 0.0090, "p99_9": 0.4000}, "top1_agreement": 0.971, "positions": 25000}
json.dump({"model": "w5nl64", "all": g, "by_source": {"code": dict(g), "ops": dict(g)}},
open(sys.argv[1], "w"))
PY
run bash "$GATE" "$D/iq4r.json" "$D/cand.json"
[ "$status" -eq 1 ]
[[ "$output" == *"[FAIL] 2."* ]]
[[ "$output" == *"KL p99.9"* ]]
}
@test "a candidate that wins overall but regresses one source slice fails" {
D="$BATS_TEST_TMPDIR/kl4"
mkdir -p "$D"
kld_report_slices "$D/iq4r.json" iq4r-moe 0.0190 0.0190 0.962 0.962
# 'all' is better, 'code' is worse: exactly the trade a mean-only gate cannot see
kld_report_slices "$D/cand.json" w5nl64 0.0090 0.0300 0.971 0.950
run bash "$GATE" "$D/iq4r.json" "$D/cand.json"
[ "$status" -eq 1 ]
[[ "$output" == *"[PASS] 1."* ]]
[[ "$output" == *"[FAIL] 4."* ]]
[[ "$output" == *"regressed: code"* ]]
}
# --- the calibration window: gated, and it says so ---------------------------
window_fixture() {
D="$BATS_TEST_TMPDIR/win"
mkdir -p "$D/campaign/kld-ref" "$D/campaign/calib" "$D/rad" "$D/corpora"
echo '{}' > "$D/campaign/kld-ref/kld.json"
: > "$D/rad/qwen38-bf16.rad"
: > "$D/corpora/calib.jsonl"
echo "$D"
}
@test "the calibration window refuses before touching a card without a reference recording" {
D="$BATS_TEST_TMPDIR/win-less"
mkdir -p "$D/campaign/calib"
run env FRANK_QWEN38_CAMPAIGN="$D/campaign" bash "$WINDOW"
[ "$status" -eq 2 ]
[[ "$output" == *"no reference recording"* ]]
}
@test "the calibration window's dry run prints the gated plan, including what it would bank" {
D="$(window_fixture)"
run env FRANK_QWEN38_CAMPAIGN="$D/campaign" \
FRANK_QWEN38_RAD="$D/rad/qwen38-bf16.rad" \
Q38_WINDOW_CALIB_CORPUS="$D/corpora/calib.jsonl" bash "$WINDOW" --dry-run
[ "$status" -eq 0 ]
[[ "$output" == *"run-qwen38-bf16.sh calib"* ]]
[[ "$output" == *"else kill the rig"* ]]
[[ "$output" == *"calib-push.py"* ]]
[[ "$output" == *"AGENTTRASH"* ]]
}